package com.sonic.common.rpc;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Function;

import com.sonic.common.rpc.exception.RpcNetworkException;
import com.sonic.common.rpc.exception.RpcProtocolException;
import com.sonic.common.rpc.exception.TokenExpiredException;

import lombok.extern.slf4j.Slf4j;
import okhttp3.*;

/**
 * @author code
 */
@Slf4j
public class OkHttpClientImpl implements HttpClient {

    private OkHttpClient okHttpClient;

    private Config config;

    private List<Interceptor> interceptorList;

    private final static long MAX_RESPONSE_LOG_LENGTH = 10_240L;

    public OkHttpClientImpl() {
        initClient();
    }

    public OkHttpClientImpl(Config config) {
        this.config = config;
        initClient();
    }

    public OkHttpClientImpl(Config config, List<Interceptor> interceptorList) {
        this.config = config;
        this.interceptorList = interceptorList;
        initClient();
    }

    @Override
    public String execute(HttpMethod httpMethod, String url, RequestParams requestParams) {
        Objects.requireNonNull(url);

        Request.Builder builder = new Request.Builder()
                .url(httpMethod.getUrlBuilder().apply(url, requestParams))
                .headers(Headers.of(Optional.ofNullable(requestParams.getHeaders()).orElse(Collections.emptyMap())));

        return execute(httpMethod.getRequestBuilder().apply(builder, requestParams), requestParams, this::getStringResponse);
    }

    @Override
    public <R> R execute(HttpMethod httpMethod, String url, RequestParams requestParams, BiFunction<byte[], Map<String, List<String>>, R> responder) {
        Objects.requireNonNull(url);

        Request.Builder builder = new Request.Builder()
                .url(httpMethod.getUrlBuilder().apply(url, requestParams))
                .headers(Headers.of(Optional.ofNullable(requestParams.getHeaders()).orElse(Collections.emptyMap())));

        return execute(httpMethod.getRequestBuilder().apply(builder, requestParams), requestParams, response -> applyResponder(response, responder));
    }

    private String getStringResponse(Response response) {
        try {
            return response.body().string();
        } catch (IOException e) {
            log.error("OkHttpClient apply string responder error", e);
            throw new RpcNetworkException("OkHttpClient apply string responder error", e);
        }
    }

    private <R> R applyResponder(Response response, BiFunction<byte[], Map<String, List<String>>, R> responder) {
        try {
            return responder.apply(response.body().bytes(), response.headers().toMultimap());
        } catch (IOException e) {
            log.error("OkHttpClient apply responder error", e);
            throw new RpcNetworkException("OkHttpClient apply responder error", e);
        }
    }

    private <R> R execute(Request.Builder requestBuilder, RequestParams requestParams,
                          Function<Response, R> responseProcessor) {
        Request request = requestBuilder.build();

        if (requestParams.getReadTimeoutMillis() != null) {
            OkHttpClient.Builder builder = okHttpClient.newBuilder();
            builder.readTimeout(requestParams.getReadTimeoutMillis(), TimeUnit.MILLISECONDS);
        }

        try (Response response = okHttpClient.newCall(request).execute()) {
            if (requestParams.isLogEnabled()) {
                log.info("OkHttpClient request success, url = {}, method = {}, param = {}, resp = {}",
                        request.url().toString(), request.method(), requestParams, response.peekBody(MAX_RESPONSE_LOG_LENGTH).string());
            }

            if (response.code() == TokenExpiredException.STATUS_CODE) {
                log.warn("OkHttpClient token expired, url = {},  method = {}, param = {}", request.url().toString(), request.method(), requestParams);
                throw new TokenExpiredException(String.format("OkHttpClient token expired, url = %s, method = %s, param = %s", request.url().toString(), request.method(), requestParams));
            }

            if (response.code() >= 500) {
                String body = response.peekBody(MAX_RESPONSE_LOG_LENGTH).string();
                log.error("OkHttpClient server error, url = {}, param = {}, response code = {}, body = {}",
                        request.url().toString(), requestParams, response.code(), body);
                throw new RpcProtocolException(response.code(), body,
                        String.format("OkHttpClient server error, url =%s, param = %s, code = %s",
                                request.url().toString(), requestParams, response.code()));
            }

            return responseProcessor.apply(response);
        } catch (IOException e) {
            log.error("OkHttpClient network error, url = {}, method = {},  param = {}", request.url().toString(), request.method(), requestParams, e);
            throw new RpcNetworkException(String.format("OkHttpClient network error, url = %s, method = %s, param = %s", request.url().toString(), request.method(), requestParams), e);
        }
    }

    private void initClient() {
        OkHttpClient.Builder builder = new OkHttpClient.Builder();
        Dispatcher dispatcher = new Dispatcher();

        config = config == null ? Config.EMPTY : config;

        dispatcher.setMaxRequests(config.getAsyncMaxRequests());
        dispatcher.setMaxRequestsPerHost(config.getAsyncMaxRequestsPerHost());
        builder.connectTimeout(config.getConnectTimeOut(), TimeUnit.MILLISECONDS)
                .readTimeout(config.getReadTimeOut(), TimeUnit.MILLISECONDS)
                .writeTimeout(config.getWriteTimeOut(), TimeUnit.MILLISECONDS)
                .connectionPool(new ConnectionPool(config.getMaxIdleConnections(), config.getKeepAliveMinutes(), TimeUnit.MINUTES))
                .dispatcher(dispatcher);
        Optional.ofNullable(interceptorList).orElse(Collections.emptyList()).
                forEach(interceptor -> builder.addInterceptor(interceptor));

        okHttpClient = builder.build();
    }
}
