package org.lxp.crawler;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CloseableHttpAsyncClientHelper extends AbstractHttpClientHelper {
    private static final Logger LOGGER = LoggerFactory.getLogger(CloseableHttpAsyncClientHelper.class);
    private PoolingNHttpClientConnectionManager connManager;

    public CloseableHttpAsyncClientHelper() throws IOReactorException {
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(Runtime.getRuntime().availableProcessors()).setSoKeepAlive(true).build();
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);

        connManager = new PoolingNHttpClientConnectionManager(ioReactor);
        connManager.setMaxTotal(maxTotalConnection);
        connManager.setDefaultMaxPerRoute(maxTotalConnection);
    }

    @Override
    public List<Integer> batchGet(List<String> urls) throws IOException {
        try (CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setConnectionManager(connManager)
                .setDefaultRequestConfig(requestConfig).build()) {
            httpclient.start();

            List<Future<HttpResponse>> respList = urls.stream().map(url -> {
                final HttpGet request = buildGet(url);
                return httpclient.execute(request, null);
            }).collect(Collectors.toCollection(LinkedList::new));

            return respList.stream().map(response -> {
                try {
                    return response.get().getStatusLine().getStatusCode();
                } catch (InterruptedException | ExecutionException e) {
                    LOGGER.error(e.getMessage(), e);
                    return null;
                }
            }).collect(Collectors.toList());
        }
    }

    @Override
    public Integer get(String url) throws Exception {
        try (CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setConnectionManager(connManager)
                .setDefaultRequestConfig(requestConfig).build()) {
            httpclient.start();

            final HttpGet request = buildGet(url);
            Future<HttpResponse> response = httpclient.execute(request, null);

            return response.get().getStatusLine().getStatusCode();
        }
    }
}

@@ -304,16 +304,66 @@
 ttpGet;%0A
+import org.apache.http.concurrent.FutureCallback;%0A
 import o
@@ -2105,20 +2105,30 @@
 equest, 
-null
+futureCallback
 );%0A     
@@ -2981,20 +2981,30 @@
 equest, 
-null
+futureCallback
 );%0A%0A    
@@ -3081,11 +3081,416 @@
  %7D%0A    %7D
+%0A%0A    private FutureCallback%3CHttpResponse%3E futureCallback = new FutureCallback%3CHttpResponse%3E() %7B%0A        public void completed(HttpResponse httpResponse) %7B%0A            LOGGER.info(%22request complete%22);%0A        %7D%0A%0A        public void failed(Exception e) %7B%0A            LOGGER.error(%22request failed%22);%0A        %7D%0A%0A        public void cancelled() %7B%0A            LOGGER.info(%22request cancelled%22);%0A        %7D%0A    %7D;
 %0A%7D%0A

add FutureCallback