/*
 * Copyright (C) Alibaba Cloud Computing All rights reserved.
 */
package com.aliyun.openservices.log;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.aliyun.openservices.log.common.*;
import com.aliyun.openservices.log.common.Consts.CompressType;
import com.aliyun.openservices.log.common.Consts.CursorMode;
import com.aliyun.openservices.log.common.auth.*;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.http.client.*;
import com.aliyun.openservices.log.http.comm.*;
import com.aliyun.openservices.log.http.signer.SignVersion;
import com.aliyun.openservices.log.http.signer.SlsSigner;
import com.aliyun.openservices.log.http.signer.SlsSignerBase;
import com.aliyun.openservices.log.http.utils.CodingUtils;
import com.aliyun.openservices.log.internal.ErrorCodes;
import com.aliyun.openservices.log.request.*;
import com.aliyun.openservices.log.response.*;
import com.aliyun.openservices.log.util.*;
import org.apache.commons.codec.binary.Base64;
import org.apache.http.conn.HttpClientConnectionManager;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;

import static com.aliyun.openservices.log.common.Consts.CONST_LOGSTORE_REPLICATION;

/**
 * Client class is the main class in the sdk, it implements the interfaces
 * defined in LogService. It can be used to send request to the log service
 * server to put/get data.
 * <p></p>
 * It is highly recommended to use {@link ClientBuilder} to build {@link Client} instance.
 */
public class Client implements LogService {
	private static final String DEFAULT_USER_AGENT = VersionInfoUtils.getDefaultUserAgent();
	private String httpType;
	private String hostName;
	private CredentialsProvider credentialsProvider;
	private String sourceIp;
	private ServiceClient serviceClient;
	private String realIpForConsole;
	private Boolean useSSLForConsole;
	private String userAgent = DEFAULT_USER_AGENT;
	private boolean mUUIDTag = false;
	private boolean useDirectMode = false;
	/**
	 * Real backend server's IP address. If not null, skip resolving DNS
	 */
	private String realServerIP = null;
	private String resourceOwnerAccount = null;
	private SlsSigner signer;
	private boolean useMetricStoreUrl;
	private ClientConfiguration clientConfiguration;
	private boolean isCname = false;

	public boolean isUseMetricStoreUrl() {
		return useMetricStoreUrl;
	}

	public void setUseMetricStoreUrl(final boolean useMetricStoreUrl) {
		this.useMetricStoreUrl = useMetricStoreUrl;
	}

	public String getUserAgent() {
		return userAgent;
	}

	public void setUserAgent(String userAgent) {
		this.userAgent = userAgent;
	}

	public boolean isCname() {
        return isCname;
    }

    public void setCname(boolean cname) {
        isCname = cname;
    }

	public String getRealIpForConsole() {
		return realIpForConsole;
	}

	public void setRealIpForConsole(String realIpForConsole) {
		this.realIpForConsole = realIpForConsole;
	}

	public boolean isUseSSLForConsole() {
		return useSSLForConsole;
	}

	public void setUseSSLForConsole(boolean useSSLForConsole) {
		this.useSSLForConsole = useSSLForConsole;
	}

	public void ClearConsoleResources() {
		realIpForConsole = null;
		useSSLForConsole = null;
	}

	public void EnableUUIDTag() { mUUIDTag = true; }

	public void DisableUUIDTag() { mUUIDTag = false; }

	@Deprecated
	public void EnableDirectMode() {
		setUseDirectMode(true);
	}

	@Deprecated
	public void DisableDirectMode() {
		setUseDirectMode(false);
	}

	public void setUseDirectMode(boolean useDirectMode) {
		this.useDirectMode = useDirectMode;
	}

	public boolean isUseDirectMode() {
		return useDirectMode;
	}

	public String getResourceOwnerAccount() {
		return resourceOwnerAccount;
	}

	public void setResourceOwnerAccount(String resourceOwnerAccount) {
		this.resourceOwnerAccount = resourceOwnerAccount;
	}

	public HttpClientConnectionManager getConnectionManager() {
        return serviceClient.getConnectionManager();
    }

	/**
	 * Construct the sls client with accessId, accessKey and server address, all
	 * other parameters will be set to default value
	 *
	 * @throws NullPointerException
	 *             if the input parameter is null
	 * @throws IllegalArgumentException
	 *             if the input parameter is empty
	 *
	 * @param endpoint
	 *            the log service server address
	 * @param accessId
	 *            aliyun accessId
	 * @param accessKey
	 *            aliyun accessKey
	 */
	public Client(String endpoint, String accessId, String accessKey) {
		this(endpoint, new DefaultCredentials(accessId, accessKey), null);
	}

	public Client(String endpoint, String accessId, String accessKey, ClientConfiguration configuration) {
		this(endpoint, accessId, accessKey, null, configuration);
	}

	public Client(String endpoint, String roleName) {
		this(endpoint, new ECSRoleCredentialsProvider(roleName), null);
	}

	/**
	 * Construct the sls client with accessId, accessKey , server address and
	 * client ip address, all other parameters will be set to default value
	 *
	 *
	 * @throws NullPointerException
	 *             if the input parameter is null
	 * @throws IllegalArgumentException
	 *             if the input parameter is empty
	 *
	 * @param endpoint
	 *            the log service server address
	 * @param accessId
	 *            aliyun accessId
	 * @param accessKey
	 *            aliyun accessKey
	 * @param sourceIp
	 *            client ip address
	 */
	public Client(String endpoint, String accessId, String accessKey, String sourceIp) {
		this(endpoint, new DefaultCredentials(accessId, accessKey), sourceIp);
	}

	public Client(String endpoint, Credentials credentials, String sourceIp) {
		this.clientConfiguration = getDefaultClientConfiguration();
		this.serviceClient = buildServiceClient(clientConfiguration);
		configure(endpoint, credentials, sourceIp);
	}

	public Client(String endpoint, CredentialsProvider credentialsProvider) {
		this(endpoint, credentialsProvider, "");
	}

	static ClientConfiguration getDefaultClientConfiguration() {
		ClientConfiguration clientConfig = new ClientConfiguration();
		clientConfig.setMaxConnections(Consts.HTTP_CONNECT_MAX_COUNT);
		clientConfig.setConnectionTimeout(Consts.HTTP_CONNECT_TIME_OUT);
		clientConfig.setSocketTimeout(Consts.HTTP_SEND_TIME_OUT);
		return clientConfig;
	}

	static ServiceClient buildServiceClient(ClientConfiguration clientConfig) {
		if (clientConfig.isRequestTimeoutEnabled()) {
			return new TimeoutServiceClient(clientConfig);
		}
		return new DefaultServiceClient(clientConfig);
	}
	/**
	 * @param endpoint            required not null, the log service server address
	 * @param credentialsProvider required not null, interface which provide credentials
	 * @param sourceIp            nullable, client ip address
	 */
	public Client(String endpoint, CredentialsProvider credentialsProvider, String sourceIp) {
		this(endpoint, credentialsProvider, new DefaultServiceClient(getDefaultClientConfiguration()), sourceIp);
	}

	/**
	 * @param endpoint            required not null, the log service server address
	 * @param credentialsProvider required not null, interface which provide credentials
	 * @param serviceClient       required not null, customized service client
	 * @param sourceIp            nullable, client ip address
	 */
	public Client(String endpoint, CredentialsProvider credentialsProvider, ServiceClient serviceClient, String sourceIp) {
		this.serviceClient = serviceClient;
		this.clientConfiguration = serviceClient.getClientConfiguration();
		configure(endpoint, credentialsProvider, sourceIp);
	}

	/**
	 * @deprecated Use Client(String endpoint, String accessId, String accessKey, String sourceIp,
	 * ClientConfiguration config) instead.
	 */
	@Deprecated
	public Client(String endpoint, String accessId, String accessKey, String sourceIp,
				  int connectMaxCount, int connectTimeout, int sendTimeout) {
		ClientConfiguration clientConfig = new ClientConfiguration();
		clientConfig.setMaxConnections(connectMaxCount);
		clientConfig.setConnectionTimeout(connectTimeout);
		clientConfig.setSocketTimeout(sendTimeout);
		this.clientConfiguration = clientConfig;
		this.serviceClient = buildServiceClient(clientConfig);
		configure(endpoint, new DefaultCredentials(accessId, accessKey), sourceIp);
	}

    public Client(String endpoint, String accessId, String accessKey, ServiceClient serviceClient) {
        this.serviceClient = serviceClient;
		this.clientConfiguration = serviceClient.getClientConfiguration();
		configure(endpoint, new DefaultCredentials(accessId, accessKey), null);
    }

    public synchronized void setEndpoint(String endpoint) {
		Args.notNullOrEmpty(endpoint, "endpoint");
		endpoint = endpoint.trim();
		if (endpoint.startsWith("http://")) {
			this.hostName = endpoint.substring(7);
			this.httpType = "http://";
		} else if (endpoint.startsWith("https://")) {
			this.hostName = endpoint.substring(8);
			this.httpType = "https://";
		} else {
			this.hostName = endpoint;
			this.httpType = "http://";
		}
		hostName = Utils.normalizeHostName(hostName);
		if (hostName == null) {
			throw new IllegalArgumentException("Invalid endpoint: " + endpoint);
		}
		if (NetworkUtils.isIPAddr(this.hostName)) {
			throw new IllegalArgumentException("The ip address is not supported");
		}
		if (getClientConfiguration().getRegion() == null) {
			setSignV4IfInAcdr(endpoint);
		}
	}

	private void configure(String endpoint, Credentials credentials, String sourceIp) {
		configure(endpoint, new StaticCredentialsProvider(credentials), sourceIp);
	}

	private void configure(String endpoint, CredentialsProvider credentialsProvider, String sourceIp) {
		setEndpoint(endpoint);
		this.sourceIp = sourceIp;
		if (sourceIp == null || sourceIp.isEmpty()) {
			this.sourceIp = NetworkUtils.getLocalMachineIP();
		}
		this.credentialsProvider = credentialsProvider;
		updateSigner(credentialsProvider);
	}

	/**
	 * @param endpoint required not null, the log service server address
	 * @param credentialsProvider required not null, interface which provide credentials
	 * @param config   required not null, client configuration
	 * @param sourceIp nullable, client ip address
	 */
	public Client(String endpoint, CredentialsProvider credentialsProvider,
				  ClientConfiguration config,
				  String sourceIp) {
		Args.notNull(config, "Config");
		this.clientConfiguration = config;
		this.serviceClient = buildServiceClient(config);
		configure(endpoint, credentialsProvider, sourceIp);
	}

	/**
	 * @param endpoint  required not null, the log service server address
	 * @param accessId  required not null, access key id
	 * @param accessKey required not null, access key secret
	 * @param config    required not null, client configuration
	 * @param sourceIp  nullable, client ip address
	 */
	@Deprecated
	public Client(String endpoint, String accessId, String accessKey, String sourceIp,
				  ClientConfiguration config) {
		this(endpoint, new StaticCredentialsProvider(new DefaultCredentials(accessId, accessKey)), config, sourceIp);
	}

	/**
	 * set credentialProvider and update the signer.
	 *
	 * @param credentialsProvider
	 */
	public void setCredentialsProvider(CredentialsProvider credentialsProvider) {
		this.credentialsProvider = credentialsProvider;
		this.updateSigner(credentialsProvider);
	}

	private void updateSigner(CredentialsProvider credentialsProvider) {
		ClientConfiguration clientConfiguration = serviceClient.getClientConfiguration();
		this.signer = SlsSignerBase.createRequestSigner(clientConfiguration, credentialsProvider);
	}

	private void ensureStaticCredentialProvider() {
		if (!(this.credentialsProvider instanceof StaticCredentialsProvider)) {
			throw new RuntimeException("can only set or get AccessId/AccessKey/SecurityToken on StaticCredentialProvider");
		}
	}

	/**
	 * If client's credentialsProvider is not of type StaticCredentialsProvider, a RuntimeException will be thrown.
	 */
	public String getAccessId() {
		ensureStaticCredentialProvider();
		return credentialsProvider.getCredentials().getAccessKeyId();
	}

	/**
	 * If client's credentialsProvider is not of type StaticCredentialsProvider, a RuntimeException will be thrown.
	 */
	public void setAccessId(String accessId) {
		ensureStaticCredentialProvider();
		StaticCredentialsProvider scp = (StaticCredentialsProvider) this.credentialsProvider;
		scp.setAccessKeyId(accessId);
	}

	/**
	 * If client's credentialsProvider is not of type StaticCredentialsProvider, a RuntimeException will be thrown.
	 */
	public String getAccessKey() {
		ensureStaticCredentialProvider();
		return credentialsProvider.getCredentials().getAccessKeySecret();
	}

	/**
	 * If client's credentialsProvider is not of type StaticCredentialsProvider, a RuntimeException will be thrown.
	 */
	public void setAccessKey(String accessKey) {
		ensureStaticCredentialProvider();
		StaticCredentialsProvider scp = (StaticCredentialsProvider) this.credentialsProvider;
		scp.setAccessKeySecret(accessKey);
	}

	/**
	 * If client's credentialsProvider is not of type StaticCredentialsProvider, a RuntimeException will be thrown.
	 */
	public String getSecurityToken() {
		ensureStaticCredentialProvider();
		return credentialsProvider.getCredentials().getSecurityToken();
	}

	/**
	 * If client's credentialsProvider is not of type StaticCredentialsProvider, a RuntimeException will be thrown.
	 */
	public void setSecurityToken(String securityToken) {
		ensureStaticCredentialProvider();
		StaticCredentialsProvider scp = (StaticCredentialsProvider) this.credentialsProvider;
		scp.setSecurityToken(securityToken);
	}

	public void shutdown() {
		serviceClient.shutdown();
	}

	public ClientConfiguration getClientConfiguration() {
		return serviceClient.getClientConfiguration();
	}

	URI GetHostURI(String project) {
		String endPointUrl = this.httpType + this.hostName;
		if (project != null && !project.isEmpty()) {
			if (!Utils.validateProject(project)) {
				throw new IllegalArgumentException("Invalid project: " + project);
			}
			if (!isCname) {
				endPointUrl = this.httpType + project + "." + this.hostName;
			}
		}
		try {
			return new URI(endPointUrl);
		} catch (URISyntaxException e) {
			throw new IllegalArgumentException("Invalid endpoint: " + endPointUrl, e);
		}
	}

	private URI GetHostURIByIp(String ip_addr) throws LogException {
		String endPointUrl = this.httpType + ip_addr;
		try {
			return new URI(endPointUrl);
		} catch (URISyntaxException e) {
			throw new LogException(ErrorCodes.ENDPOINT_INVALID,
					"Failed to get real server ip when direct mode in enabled", "");
		}
	}

	private static byte[] encodeToUtf8(String source) throws LogException {
		try {
			return source.getBytes(Consts.UTF_8_ENCODING);
		} catch (UnsupportedEncodingException e) {
			throw new LogException(ErrorCodes.ENCODING_EXCEPTION, e.getMessage(), "");
		}
	}

	private static String encodeResponseBodyToUtf8String(ResponseMessage response, String requestId) throws LogException {
		byte[] body = response.GetRawBody();
		if (body == null) {
			throw new LogException(ErrorCodes.BAD_RESPONSE, "The response body is null", null, requestId);
		}
		try {
			return new String(body, Consts.UTF_8_ENCODING);
		} catch (UnsupportedEncodingException e) {
			throw new LogException(ErrorCodes.BAD_RESPONSE,
					"The response is not valid utf-8 string: ", e, requestId);
		}
	}

	public GetLogtailProfileResponse ExtractLogtailProfile(Map<String, String> resHeaders, JSONObject object) throws LogException {
		try {
			int count = object.getIntValue("count");
			int total = object.getIntValue("total");
			JSONArray array = object.getJSONArray("profile");
			List<LogtailProfile> logtailProfiles = new ArrayList<LogtailProfile>();
			if (array != null) {
				for (int i = 0; i < array.size(); i++) {
					JSONObject profileObj = array.getJSONObject(i);
					if (profileObj == null) {
						continue;
					}
					LogtailProfile logtailProfile = new LogtailProfile();
					logtailProfile.FromJsonObject(profileObj);
					logtailProfiles.add(logtailProfile);
				}
			}
			return new GetLogtailProfileResponse(resHeaders, count, total, logtailProfiles);
		} catch (LogException e) {
			throw new LogException(e.getErrorCode(), e.getMessage(),
					e.getCause(), GetRequestId(resHeaders));
		}
	}

	// use `tagResources(TagResourcesRequest request)` instead
	@Deprecated
	public TagResourcesResponse tagResources(String tagResourcesStr) throws LogException {
		CodingUtils.assertParameterNotNull(tagResourcesStr, "tagResourcesStr");
		Map<String, String> headParameter = GetCommonHeadPara("");
		byte[] body = encodeToUtf8(tagResourcesStr);
		String resourceUri = "/tag";
		Map<String, String> urlParameter = new HashMap<String, String>();
		ResponseMessage response = SendData("", HttpMethod.POST, resourceUri, urlParameter, headParameter, body);
        Map<String, String> resHeaders = response.getHeaders();
        return new TagResourcesResponse(resHeaders);
	}

	public TagResourcesResponse tagResources(TagResourcesRequest request) throws LogException {
		Args.notNull(request, "request");
		Map<String, String> headParameter = GetCommonHeadPara("");
		byte[] body = encodeToUtf8(JsonUtils.serialize(request));
		String resourceUri = "/tag";
		Map<String, String> urlParameter = new HashMap<String, String>();
		ResponseMessage response = SendData(request.getProject(), HttpMethod.POST, resourceUri, urlParameter, headParameter, body);
        Map<String, String> resHeaders = response.getHeaders();
        return new TagResourcesResponse(resHeaders);
	}

	// use `untagResources(UntagResourcesRequest request)` instead
	@Deprecated
	public UntagResourcesResponse untagResources(String untagResourcesStr) throws LogException {
		CodingUtils.assertParameterNotNull(untagResourcesStr, "tagResourcesStr");
		Map<String, String> headParameter = GetCommonHeadPara("");
		byte[] body = encodeToUtf8(untagResourcesStr);
		String resourceUri = "/untag";
		Map<String, String> urlParameter = new HashMap<String, String>();
		ResponseMessage response = SendData("", HttpMethod.POST, resourceUri, urlParameter, headParameter, body);
        Map<String, String> resHeaders = response.getHeaders();
        return new UntagResourcesResponse(resHeaders);
	}

	public UntagResourcesResponse untagResources(UntagResourcesRequest request) throws LogException {
		Args.notNull(request, "request");
		Map<String, String> headParameter = GetCommonHeadPara("");
		byte[] body = encodeToUtf8(JsonUtils.serialize(request));
		String resourceUri = "/untag";
		Map<String, String> urlParameter = new HashMap<String, String>();
		ResponseMessage response = SendData(request.getProject(), HttpMethod.POST, resourceUri, urlParameter, headParameter, body);
        Map<String, String> resHeaders = response.getHeaders();
        return new UntagResourcesResponse(resHeaders);
	}

	public TagResourcesResponse tagResourcesSystemTags(TagResourcesSystemTagsRequest request) throws LogException {
		Args.notNull(request, "request");
		Map<String, String> headParameter = GetCommonHeadPara("");
		String tagResourcesStr = JsonUtils.serialize(request);
		byte[] body = encodeToUtf8(tagResourcesStr);
		String resourceUri = "/systemtag";
		Map<String, String> urlParameter = new HashMap<String, String>();
		ResponseMessage response = SendData("", HttpMethod.POST, resourceUri, urlParameter, headParameter, body);
		Map<String, String> resHeaders = response.getHeaders();
		return new TagResourcesResponse(resHeaders);
	}

	public UntagResourcesResponse untagResourcesSystemTags(UntagResourcesSystemTagsRequest request) throws LogException {
		Args.notNull(request, "request");
		String untagResourcesStr = JsonUtils.serialize(request);
		Map<String, String> headParameter = GetCommonHeadPara("");
		byte[] body = encodeToUtf8(untagResourcesStr);
		String resourceUri = "/systemuntag";
		Map<String, String> urlParameter = new HashMap<String, String>();
		ResponseMessage response = SendData("", HttpMethod.POST, resourceUri, urlParameter, headParameter, body);
		Map<String, String> resHeaders = response.getHeaders();
		return new UntagResourcesResponse(resHeaders);
	}

	public ListTagResourcesResponse listSystemTagResources(ListSystemTagResourcesRequest request) throws LogException {
		CodingUtils.assertParameterNotNull(request, "request");
		Map<String, String> urlParameter = request.GetAllParams();
		Map<String, String> headParameter = GetCommonHeadPara("");
		String resourceUri = "/systemtags";
		ResponseMessage response = SendData("", HttpMethod.GET, resourceUri, urlParameter, headParameter);
		Map<String, String> resHeaders = response.getHeaders();
		String requestId = GetRequestId(response.getHeaders());
		JSONObject object = parseResponseBody(response, requestId);
		List<TagResource> tagResources = ExtractTagResources(object, requestId);
		String nextToken = object.getString("nextToken");
		return new ListTagResourcesResponse(resHeaders, nextToken, tagResources);
	}

	protected List<TagResource> ExtractTagResources(JSONObject object, String requestId)
			throws LogException {
		List<TagResource> tagResources = new ArrayList<TagResource>();
		if (object == null) {
			return tagResources;
		}
		try {
			JSONArray array = object.getJSONArray("tagResources");
			if (array == null) {
				return tagResources;
			}

			for (int index = 0; index < array.size(); index++) {
				JSONObject item = array.getJSONObject(index);
				if(item != null) {
					tagResources.add(TagResource.FromJsonObject(item));
				}
			}
		} catch (JSONException e) {
			throw new LogException(ErrorCodes.BAD_RESPONSE, "The response is not valid config json array string : " + object.toString(), e, requestId);
		}

		return tagResources;
	}


	public ListTagResourcesResponse listTagResources(ListTagResourcesRequest request) throws LogException {
		CodingUtils.assertParameterNotNull(request, "request");
		Map<String, String> urlParameter = request.GetAllParams();
		Map<String, String> headParameter = GetCommonHeadPara("");
		String resourceUri = "/tags";
		ResponseMessage response = SendData("", HttpMethod.GET, resourceUri, urlParameter, headParameter);
		Map<String, String> resHeaders = response.getHeaders();
		String requestId = GetRequestId(response.getHeaders());
		JSONObject object = parseResponseBody(response, requestId);
		List<TagResource> tagResources = ExtractTagResources(object, requestId);
		String nextToken = object.getString("nextToken");
		return new ListTagResourcesResponse(resHeaders, nextToken, tagResources);
	}

	public GetLogtailProfileResponse GetLogtailProfile(String project, String logstore, String source,
			int line, int offset) throws LogException {
		CodingUtils.assertStringNotNullOrEmpty(project, "project");
		CodingUtils.assertStringNotNullOrEmpty(logstore, "logstore");
		GetLogtailProfileRequest request = new GetLogtailProfileRequest(project, logstore, source, line, offset);
		return GetLogtailProfile(request);
	}

	public GetLogtailProfileResponse GetLogtailProfile(GetLogtailProfileRequest request) throws LogException {
		CodingUtils.assertParameterNotNull(request, "request");
		Map<String, String> urlParameter = request.GetAllParams();
		String project = request.GetProject();
		Map<String, String> headParameter = GetCommonHeadPara(project);
		CodingUtils.validateLogstore(request.getLogStore());
		String resourceUri = "/logstores/" + request.getLogStore() + Consts.CONST_GETLOGTAILPROFILE_URI;
		ResponseMessage response = SendData(project, HttpMethod.GET,
				resourceUri, urlParameter, headParameter);
		Map<String, String> resHeaders = response.getHeaders();
		String requestId = GetRequestId(resHeaders);
		JSONObject object = parseResponseBody(response, requestId);
		return ExtractLogtailProfile(resHeaders, object);
	}

	public GetHistogramsResponse GetHistograms(String project, String logStore,
			int from, int to, String topic, String query) throws LogException {
		CodingUtils.assertStringNotNullOrEmpty(project, "project");
		CodingUtils.assertStringNotNullOrEmpty(logStore, "logStore");
		CodingUtils.assertParameterNotNull(topic, "topic");
		CodingUtils.assertParameterNotNull(query, "query");
		GetHistogramsRequest request = new GetHistogramsRequest(project,
				logStore, topic, query, from, to);
		return GetHistograms(request);
	}

	public GetHistogramsResponse GetHistograms(GetHistogramsRequest request)
			throws LogException {
		CodingUtils.assertParameterNotNull(request, "request");
		Map<String, String> urlParameter = request.GetAllParams();
		String project = request.GetProject();
		Map<String, String> headParameter = GetCommonHeadPara(project);
		CodingUtils.validateLogstore(request.GetLogStore());
		String resourceUri = "/logstores/" + request.GetLogStore() + "/index";
		ResponseMessage response = SendData(project, HttpMethod.GET,
				resourceUri, urlParameter, headParameter);
		Map<String, String> resHeaders = response.getHeaders();
		String requestId = GetRequestId(resHeaders);
		JSONArray object = ParseResponseMessageToArray(response, requestId);
		GetHistogramsResponse histogramResponse = new GetHistogramsResponse(resHeaders);
		histogramResponse.fromJSON(object);
		return histogramResponse;
	}

	public PutLogsResponse PutLogs(String project, String logStore, byte[] logGroupBytes, String compressType, String shardHash) throws LogException {
		if (isUseMetricStoreUrl()) {
			shardHash = Consts.METRICS_STORE_AUTO_HASH;
		}
		CodingUtils.assertStringNotNullOrEmpty(project, "project");
		CodingUtils.assertStringNotNullOrEmpty(logStore, "logStore");
		CodingUtils.assertParameterNotNull(logGroupBytes, "logGroupBytes");
		PutLogsRequest request = new PutLogsRequest(project, logStore, null,
				null, logGroupBytes, shardHash);
		request.SetCompressType(CompressType.fromString(compressType));
		return PutLogs(request);
	}

	@Override
	public PutLogsResponse PutLogs(String project, String logStore,
			String topic, List<LogItem> logItems, String source,
			String shardHash) throws LogException {
		if (isUseMetricStoreUrl()) {
			shardHash = Consts.METRICS_STORE_AUTO_HASH;
		}
		CodingUtils.assertStringNotNullOrEmpty(project, "project");
		CodingUtils.assertStringNotNullOrEmpty(logStore, "logStore");
		CodingUtils.assertParameterNotNull(topic, "topic");
		CodingUtils.assertParameterNotNull(logItems, "logGroup");
		PutLogsRequest request = new PutLogsRequest(project, logStore, topic,
				source, logItems, shardHash);
		request.SetCompressType(CompressType.LZ4);
		return PutLogs(request);
	}

	@Override
	public PutLogsResponse PutLogs(String project, String logStore,
			String topic, List<LogItem> logItems, String source)
			throws LogException {
		String shardHash = isUseMetricStoreUrl() ? Consts.METRICS_STORE_AUTO_HASH : null;
		CodingUtils.assertStringNotNullOrEmpty(project, "project");
		CodingUtils.assertStringNotNullOrEmpty(logStore, "logStore");
		CodingUtils.assertParameterNotNull(topic, "topic");
		CodingUtils.assertParameterNotNull(logItems, "logGroup");
		PutLogsRequest request = new PutLogsRequest(project, logStore, topic,
				source, logItems, shardHash);
		request.SetCompressType(CompressType.LZ4);
		return PutLogs(request);
	}

	@Override
	public BatchPutLogsResponse batchPutLogs(BatchPutLogsRequest request)
		throws LogException {
		String project = request.GetProject();
		CodingUtils.assertStringNotNullOrEmpty(project, "project");
		CodingUtils.assertStringNotNullOrEmpty(request.getLogStore(), "logStore");
		final CompressType compressType = request.getCompressType();
		CodingUtils.assertParameterNotNull(compressType, "compressType");
		if (compressType != CompressType.ZSTD && compressType != CompressType.LZ4) {
			throw new LogException(ErrorCodes.INVALID_PARAMETER, "Unsupported compress type:" + compressType, "");
		}
		List<LogGroup> logGroups = request.getLogGroups();
		if (logGroups == null || logGroups.isEmpty()) {
			throw new LogException(ErrorCodes.INVALID_PARAMETER, "LogGroups is empty", "");
		}

		byte[] logBytes = request.serializeToPb();
		Map<String, String> headParameter = GetCommonHeadPara(project);
		headParameter.put(Consts.CONST_X_LOG_MODE, Consts.CONST_BATCH_GROUP);
		headParameter.put(Consts.CONST_CONTENT_TYPE, Consts.CONST_PROTO_BUF);
		headParameter.put(Consts.CONST_X_SLS_BODYRAWSIZE, String.valueOf(logBytes.length));
		headParameter.put(Consts.CONST_X_SLS_COMPRESSTYPE, compressType.toString());
		checkBodyRawSize(logBytes.length);
		String resourceUri = "/logstores/" + request.getLogStore();
		String shardKey = request.getHashKey();
		Map<String, String> urlParameter = request.GetAllParams();
		if (shardKey == null || shardKey.isEmpty()) {
			resourceUri += "/shards/lb";
		} else {
			resourceUri += "/shards/route";
			urlParameter.put("key", shardKey);
		}

		ResponseMessage response = sendLogBytes(project, logBytes, resourceUri, urlParameter, headParameter);
		if (response != null) {
			return new BatchPutLogsResponse(response.getHeaders());
		}
		// never happen
		return null;
	}

	@Override
	public PutObjectResponse putObject(String project, String logStore, String objectName, InputStream content) throws LogException {
		return putObject(project, logStore, objectName, content, null);
	}

    @Override
    public PutObjectResponse putObject(String project, String logStore, String objectName, InputStream content,
            Map<String, String> headers) throws LogException {
        CodingUtils.assertStringNotNullOrEmpty(project, "project");
        CodingUtils.assertStringNotNullOrEmpty(logStore, "logStore");
        CodingUtils.assertParameterNotNull(objectName, "objectName");
        CodingUtils.validateLogstore(logStore);

        String resourceUri = "/logstores/" + logStore + "/objects/" + encodeObjectName(objectName);
        Map<String, String> urlParameter = new HashMap<String, String>();
        Map<String, String> headParameter = GetCommonHeadPara(project);

        // Add custom headers if provided
        if (headers != null) {
            headParameter.putAll(headers);
        }

        // Sign with empty bytes (not with input content)
        if (resourceOwnerAccount != null && !resourceOwnerAccount.isEmpty()) {
            headParameter.put(Consts.CONST_X_LOG_RESOURCEOWNERACCOUNT, resourceOwnerAccount);
        }
        try {
            signer.sign(HttpMethod.PUT, headParameter, resourceUri, urlParameter, new byte[0]);
        } catch (Exception e) {
            throw new LogException("ClientSignatureError",
                    "Fail to calculate signature for request, error:" + e.getMessage(), "");
        }

        // Build request with InputStream directly (do not read it)
        URI uri = getHostURI(project, null);
        RequestMessage requestMessage = BuildRequest(uri, HttpMethod.PUT,
                resourceUri, urlParameter, headParameter,
                content != null ? content : new java.io.ByteArrayInputStream(new byte[0]),
                -1);

        ResponseMessage response = null;
        try {
            response = this.serviceClient.sendRequest(requestMessage, Consts.UTF_8_ENCODING);
            throwServerErrorIfPresent(response);
            ExtractResponseBody(response);
            Map<String, String> resHeaders = response.getHeaders();
            return new PutObjectResponse(resHeaders);
        } catch (ServiceException e) {
            throw new LogException("RequestError", "Web request failed: " + e.getMessage(), e, "");
        } catch (ClientException e) {
            throw new LogException("RequestError", "Web request failed: " + e.getMessage(), e, "");
        } catch (LogException e) {
            throw e;
        } catch (Exception e) {
            throw new LogException(ErrorCodes.BAD_RESPONSE,
                    "Fail to send request, error:" + e.getMessage(), e, getRequestId(response));
        } finally {
            safeClose(response);
        }
    }


    // when statusCode is not 200, throw LogException, else do nothing
    // must be called before ExtractResponseBody
    // we dont close response here, the caller is responsible for closing it
    private void throwServerErrorIfPresent(ResponseMessage response) throws LogException {
        int statusCode = response.getStatusCode();
        if (statusCode == Consts.CONST_HTTP_OK) {
            return;
        }
        ExtractResponseBody(response);
        String requestId = GetRequestId(response.getHeaders());
        try {
            String responseBody = encodeResponseBodyToUtf8String(response, requestId);
            JSONObject object = null;
            try {
