/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package com.treasuredata.client.api;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.guava.GuavaModule;
import com.fasterxml.jackson.datatype.jsonorg.JsonOrgModule;
import com.google.common.base.Optional;
import com.google.common.base.Throwables;
import com.treasuredata.client.ErrorCode;
import com.treasuredata.client.ExponentialBackOffRetry;
import com.treasuredata.client.TDClientConfig;
import com.treasuredata.client.TDClientException;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.ContentResponse;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.util.HttpCookieStore;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * An extension of Jetty HttpClient with request retry handler
 */
public class TDHttpClient
{
    private static final Logger logger = LoggerFactory.getLogger(TDHttpClient.class);
    private final TDClientConfig config;
    private final HttpClient httpClient;
    private final ObjectMapper objectMapper;

    public TDHttpClient(TDClientConfig config)
    {
        this.config = config;
        this.httpClient = new HttpClient();
        httpClient.setConnectTimeout(3000);
        httpClient.setAddressResolutionTimeout(3000);
        httpClient.setIdleTimeout(10000);
        httpClient.setTCPNoDelay(true);
        httpClient.setExecutor(new QueuedThreadPool());
        httpClient.setMaxConnectionsPerDestination(64);
        httpClient.setCookieStore(new HttpCookieStore.Empty());

        try {
            httpClient.start();
        }
        catch (Exception e) {
            logger.error("Failed to initialize Jetty client", e);
            throw Throwables.propagate(e);
        }

        this.objectMapper = new ObjectMapper()
                .registerModule(new JsonOrgModule()) // for mapping query json to JSONObject
                .registerModule(new GuavaModule())   // for mapping to Guava Optional class
                .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }

    public void close()
    {
        try {
            httpClient.stop();
        }
        catch (Exception e) {
            logger.error("Failed to terminate Jetty client", e);
            throw Throwables.propagate(e);
        }
    }

    public ContentResponse submit(ApiRequest apiRequest)
            throws TDClientException
    {
        ExponentialBackOffRetry retry = new ExponentialBackOffRetry(config.getRetryLimit(), config.getRetryInitialWaitMillis(), config.getRetryWaitMillis());
        Optional<Exception> rootCause = Optional.absent();
        try {

            Optional<Integer> nextInterval = Optional.absent();
            do {
                if(retry.getExecutionCount() > 0) {
                    int waitTimeMillis = nextInterval.get();
                    logger.warn(String.format("Retrying request to %s (%d/%d) in %.2f sec.", apiRequest.getPath(), retry.getExecutionCount(), retry.getMaxRetryCount(), waitTimeMillis / 1000.0));
                    Thread.sleep(waitTimeMillis);
                }

                try {
                    if(logger.isDebugEnabled()) {
                        logger.debug(String.format("Sending API request to %s", apiRequest.getPath()));
                    }
                    Request request = apiRequest.newJettyRequest(httpClient, config);
                    ContentResponse response = request.send();
                    int code = response.getStatus();
                    if (HttpStatus.isSuccess(code)) {
                        // 2xx success
                        if(logger.isDebugEnabled()) {
                            logger.debug(String.format("API request to %s succeeded with code %d", request.getPath(), code));
                        }
                        return response;
                    }
                    else if (HttpStatus.isClientError(code)) {
                        // 4xx errors
                        throw new TDClientException(ErrorCode.API_CLIENT_ERROR, response.getReason());
                    }
                    else if (HttpStatus.isServerError(code)) {
                        // 5xx errors
                        logger.warn("API request to %s failed with internal error code %d: %s", request.getPath(), code, response.getReason());
                    }
                    else {
                        logger.warn("API request to %s failed with code %d: %s", request.getPath(), code, response.getReason());
                    }
                }
                catch (ExecutionException e) {
                    rootCause = Optional.<Exception>of(e);
                    logger.warn("API request failed", e);
                }
                catch (TimeoutException e) {
                    rootCause = Optional.<Exception>of(e);
                    logger.warn(String.format("API request to %s timed out", apiRequest.getPath()), e);
                }
            }
            while((nextInterval = retry.nextWaitTimeMillis()).isPresent());
        }
        catch (InterruptedException e) {
            throw new TDClientException(ErrorCode.API_EXECUTION_INTERRUPTED, e);
        }
        throw new TDClientException(ErrorCode.API_RETRY_LIMIT_EXCEEDED, String.format("Failed to process the API request to %s", apiRequest.getPath()), rootCause);
    }

    public <Result> Result submit(ApiRequest request, Class<Result> resultType)
            throws TDClientException
    {
        try {
            ContentResponse response = submit(request);
            if (logger.isTraceEnabled()) {
                logger.trace("response json:\n" + response.getContentAsString());
            }
            return objectMapper.readValue(response.getContent(), resultType);
        }
        catch (JsonMappingException e) {
            logger.error("Jackson mapping error", e);
            throw new TDClientException(ErrorCode.API_INVALID_JSON_RESPONSE, e);
        }
        catch (IOException e) {
            throw new TDClientException(ErrorCode.API_INVALID_JSON_RESPONSE, e);
        }
    }
}

@@ -1832,46 +1832,8 @@
 on;%0A
-import java.util.concurrent.TimeUnit;%0A
 impo
@@ -1874,16 +1874,16 @@
 eption;%0A
+
 %0A/**%0A * 
@@ -3004,16 +3004,26 @@
 ry json 
+strings in
 to JSONO
@@ -3884,16 +3884,17 @@
       if
+ 
 (retry.g
@@ -4280,32 +4280,33 @@
               if
+ 
 (logger.isDebugE
@@ -4764,16 +4764,17 @@
       if
+ 
 (logger.
@@ -6111,32 +6111,32 @@
 %7D%0A            %7D%0A
-
             whil
@@ -6136,16 +6136,17 @@
    while
+ 
 ((nextIn

Enhance log message