upgrade AHC to v2.0.15 #37

CHANGELOG.md

@@ -1,6 +1,16 @@
 Parallec Change Log
 ==========
 
+
+
+## Version 0.20.0-SNAPSHOT
+
+_2016-09-30_
+
+* Issue [#37](https://github.com/eBay/parallec/issues/37). Upgraded Async-http-client to version 2.0.15.  This version has not been tested in PROD environment. The client auth plugin is to be tested. The code is in branch `issue37`
+
+
+
 ## Version 0.10.1-beta
 
 

pom.xml

@@ -4,7 +4,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>io.parallec</groupId>
 	<artifactId>parallec-core</artifactId>
-	<version>0.10.1-beta</version>
+	<version>0.20.1-SNAPSHOT</version>
 	<packaging>jar</packaging>
 	<name>io.parallec:parallec-core</name>
 	<url>https://github.com/eBay/parallec</url>
@@ -56,7 +56,7 @@
 		<commons-codec.version>1.10</commons-codec.version>
 		<commons-io.version>2.3</commons-io.version>
 
-		<async-http-client.version>1.6.5</async-http-client.version>
+		<async-http-client.version>2.0.15</async-http-client.version>
 		<akka-actor_2.10.version>2.3.3</akka-actor_2.10.version>
 		<akka-cluster_2.10.version>2.3.3</akka-cluster_2.10.version>
 
@@ -119,11 +119,18 @@
 			<artifactId>commons-lang3</artifactId>
 			<version>${commons-lang3.version}</version>
 		</dependency>
-		<dependency>
+<!-- 		<dependency>
 			<groupId>com.ning</groupId>
 			<artifactId>async-http-client</artifactId>
 			<version>${async-http-client.version}</version>
-		</dependency>
+		</dependency> -->
+		<dependency>
+		    <groupId>org.asynchttpclient</groupId>
+		    <artifactId>async-http-client</artifactId>
+		    <version>${async-http-client.version}</version>
+		</dependency>		
+
+
 		<dependency>
 			<groupId>com.typesafe.akka</groupId>
 			<artifactId>akka-actor_2.10</artifactId>
@@ -205,8 +212,8 @@
 					<groupId>org.apache.maven.plugins</groupId>
 					<artifactId>maven-compiler-plugin</artifactId>
 					<configuration>
-						<source>1.7</source>
-						<target>1.7</target>
+						<source>1.8</source>
+						<target>1.8</target>
 					</configuration>
 				</plugin>
 			</plugins>
@@ -257,8 +264,8 @@
 							<groupId>org.apache.maven.plugins</groupId>
 							<artifactId>maven-compiler-plugin</artifactId>
 							<configuration>
-								<source>1.6</source>
-								<target>1.6</target>
+								<source>1.8</source>
+								<target>1.8</target>
 							</configuration>
 						</plugin>
 					</plugins>

src/main/java/io/parallec/core/ParallelClient.java

@@ -25,7 +25,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.ning.http.client.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClient;
 
 /**
  * 

src/main/java/io/parallec/core/ParallelTask.java

@@ -52,7 +52,7 @@
 
 import akka.actor.ActorRef;
 
-import com.ning.http.client.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClient;
 
 /**
  * The key class represents a onetime execution on multiple requests. It

src/main/java/io/parallec/core/ParallelTaskBuilder.java

@@ -48,7 +48,7 @@
 import org.slf4j.LoggerFactory;
 
 import com.google.common.base.Strings;
-import com.ning.http.client.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClient;
 
 
 // TODO: Auto-generated Javadoc

src/main/java/io/parallec/core/actor/ExecutionManager.java

@@ -68,7 +68,7 @@
 import akka.actor.UntypedActor;
 
 import com.google.common.base.Strings;
-import com.ning.http.client.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClient;
 
 /**
  * This is the Akka Actor that executes the ParallelTask.

src/main/java/io/parallec/core/actor/HttpWorker.java

@@ -24,14 +24,19 @@
 import io.parallec.core.util.PcHttpUtils;
 import io.parallec.core.util.PcStringUtils;
 
-import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import org.asynchttpclient.AsyncCompletionHandler;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.BoundRequestBuilder;
+import org.asynchttpclient.ListenableFuture;
+import org.asynchttpclient.Response;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,13 +48,7 @@
 import akka.actor.UntypedActor;
 
 import com.google.common.base.Strings;
-import com.ning.http.client.AsyncCompletionHandler;
-import com.ning.http.client.AsyncHttpClient;
-import com.ning.http.client.AsyncHttpClient.BoundRequestBuilder;
-import com.ning.http.client.ListenableFuture;
-import com.ning.http.client.Response;
 
-// TODO: Auto-generated Javadoc
 /**
  * This is an akka actor with async http client.
  *
@@ -367,18 +366,24 @@ public ResponseOnSingeRequest onComplete(Response response) {
         cancelCancellable();
         try {
             Map<String, List<String>> responseHeaders = null;
+            //TODO FIX NOT TESTED
             if (responseHeaderMeta != null) {
                 responseHeaders = new LinkedHashMap<String, List<String>>();
                 if (responseHeaderMeta.isGetAll()) {
-                    for (Map.Entry<String, List<String>> header : response
-                            .getHeaders()) {
-                        responseHeaders.put(header.getKey().toLowerCase(Locale.ROOT), header.getValue());
+                    for (Map.Entry<String,String> header : response.getHeaders()
+                            .entries()) {
+                        List<String> list = new ArrayList<String>();
+                        list.add( header.getValue());
+                        responseHeaders.put(header.getKey().toLowerCase(Locale.ROOT),  
+                               list);
                     }
                 } else {
                     for (String key : responseHeaderMeta.getKeys()) {
-                        if (response.getHeaders().containsKey(key)) {
+                        if (response.getHeaders().contains(key)) {
+                            List<String> list = new ArrayList<String>();
+                            list.add(response.getHeaders().get(key));
                             responseHeaders.put(key.toLowerCase(Locale.ROOT),
-                                    response.getHeaders().get(key));
+                                   list );
                         }
                     }
                 }
@@ -389,7 +394,7 @@ public ResponseOnSingeRequest onComplete(Response response) {
 
             reply(response.getResponseBody(), false, null, null, statusCode,
                     statusCodeInt, responseHeaders);
-        } catch (IOException e) {
+        } catch (Exception e) {
             getLogger().error("fail response.getResponseBody " + e);
         }
 

src/main/java/io/parallec/core/actor/OperationWorker.java

@@ -43,7 +43,7 @@
 import akka.actor.Props;
 import akka.actor.UntypedActor;
 
-import com.ning.http.client.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClient;
 
 
 

src/main/java/io/parallec/core/bean/HttpMeta.java

@@ -19,11 +19,10 @@
 import io.parallec.core.resources.HttpMethod;
 import io.parallec.core.util.PcConstants;
 
+import org.asynchttpclient.AsyncHttpClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.ning.http.client.AsyncHttpClient;
-
 // TODO: Auto-generated Javadoc
 /**
  * The metadata about the HTTP request (url/port/header/concurrency etc) It does

src/main/java/io/parallec/core/resources/AsyncHttpClientFactoryEmbed.java

@@ -32,12 +32,13 @@
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.X509TrustManager;
 
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.DefaultAsyncHttpClient;
+import org.asynchttpclient.DefaultAsyncHttpClientConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.ning.http.client.AsyncHttpClient;
-import com.ning.http.client.AsyncHttpClientConfigBean;
-
 
 /**
  * The Class HttpClientFactoryEmbed.
@@ -65,29 +66,28 @@ public AsyncHttpClientFactoryEmbed() {
 
         try {
             // create and configure async http client
-            AsyncHttpClientConfigBean configFastClient = new AsyncHttpClientConfigBean();
-
+            AsyncHttpClientConfig configFastClient = new DefaultAsyncHttpClientConfig.Builder()
+            .setConnectTimeout(ParallecGlobalConfig.ningFastClientConnectionTimeoutMillis)
+            .setRequestTimeout(ParallecGlobalConfig.ningFastClientConnectionTimeoutMillis)
+            .build();
             logger.info(
                     "FastClient: ningFastClientConnectionTimeoutMillis: {}",
                     ParallecGlobalConfig.ningFastClientConnectionTimeoutMillis);
-            configFastClient
-                    .setConnectionTimeOutInMs(ParallecGlobalConfig.ningFastClientConnectionTimeoutMillis);
-
+
             logger.info("FastClient: ningFastClientRequestTimeoutMillis: {}",
                     ParallecGlobalConfig.ningFastClientRequestTimeoutMillis);
-            configFastClient
-                    .setRequestTimeoutInMs(ParallecGlobalConfig.ningFastClientRequestTimeoutMillis);
-            fastClient = new AsyncHttpClient(configFastClient);
+            fastClient = new DefaultAsyncHttpClient(configFastClient);
+
 
             // TODO added
             // configFastClient.setMaxRequestRetry(3);
 
-            AsyncHttpClientConfigBean configSlowClient = new AsyncHttpClientConfigBean();
-            configSlowClient
-                    .setConnectionTimeOutInMs(ParallecGlobalConfig.ningSlowClientConnectionTimeoutMillis);
-            configSlowClient
-                    .setRequestTimeoutInMs(ParallecGlobalConfig.ningSlowClientRequestTimeoutMillis);
-            slowClient = new AsyncHttpClient(configSlowClient);
+            //TODO UNTESTED
+            AsyncHttpClientConfig configSlowClient = new DefaultAsyncHttpClientConfig.Builder()
+            .setConnectTimeout(ParallecGlobalConfig.ningSlowClientConnectionTimeoutMillis)
+            .setRequestTimeout(ParallecGlobalConfig.ningSlowClientRequestTimeoutMillis)
+            .build();
+             slowClient = new DefaultAsyncHttpClient(configSlowClient);
 
             disableCertificateVerification();
         } catch (Exception e) {

src/main/java/io/parallec/core/resources/HttpClientStore.java

@@ -12,11 +12,14 @@
  */
 package io.parallec.core.resources;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
 
-import com.ning.http.client.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 
 /**
@@ -35,6 +38,10 @@
  */
 public class HttpClientStore {
 
+    /** The logger. */
+    private static Logger logger = LoggerFactory
+            .getLogger(HttpClientStore.class);
+
     /** The http client type current default. */
     private HttpClientType httpClientTypeCurrentDefault;
 
@@ -93,7 +100,11 @@ public void shutdown() {
         for (Entry<HttpClientType, AsyncHttpClient> entry : map.entrySet()) {
             AsyncHttpClient client = entry.getValue();
             if (client != null)
-                client.close();
+                try {
+                    client.close();
+                } catch (IOException ex) {
+                    logger.error("error",ex);
+                }
         }
 
     }

src/main/java/io/parallec/core/util/PcHttpUtils.java

@@ -20,7 +20,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.ning.http.client.AsyncHttpClient.BoundRequestBuilder;
+import org.asynchttpclient.BoundRequestBuilder;
 
 
 /**