Merge remote-tracking branch 'upstream/master'

Author: lquerel

pom.xml

@@ -198,6 +198,7 @@
                 <configuration>
                     <manifestLocation>META-INF</manifestLocation>
                     <instructions>
+                        <Bundle-Version>$(replace;$(project.version);-SNAPSHOT;.$(tstamp;yyyyMMdd-HHmm))</Bundle-Version>
                         <Bundle-Vendor>Sonatype</Bundle-Vendor>
                         <Import-Package>
                             org.jboss.netty.*;resolution:=optional,
@@ -206,7 +207,7 @@
                             *
                         </Import-Package>
                         <Export-Package>
-                            com.ning.http.*;version="1.6.1"
+                            com.ning.http.*;version="$(replace;$(project.version);-SNAPSHOT;"")"
                         </Export-Package>
                     </instructions>
                 </configuration>

src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

@@ -720,14 +720,14 @@ public Response prepareResponse(final HttpResponseStatus status,
     /* @Override */
 
     public <T> ListenableFuture<T> execute(Request request, final AsyncHandler<T> asyncHandler) throws IOException {
-        return doConnect(request, asyncHandler, null, true);
+        return doConnect(request, asyncHandler, null, true, executeConnectAsync);
     }
 
-    private <T> void execute(final Request request, final NettyResponseFuture<T> f, boolean useCache) throws IOException {
-        doConnect(request, f.getAsyncHandler(), f, useCache);
+    private <T> void execute(final Request request, final NettyResponseFuture<T> f, boolean useCache, boolean asyncConnect) throws IOException {
+        doConnect(request, f.getAsyncHandler(), f, useCache, asyncConnect);
     }
 
-    private <T> ListenableFuture<T> doConnect(final Request request, final AsyncHandler<T> asyncHandler, NettyResponseFuture<T> f, boolean useCache) throws IOException {
+    private <T> ListenableFuture<T> doConnect(final Request request, final AsyncHandler<T> asyncHandler, NettyResponseFuture<T> f, boolean useCache, boolean asyncConnect) throws IOException {
 
         if (isClose.get()) {
             throw new IOException("Closed");
@@ -827,7 +827,7 @@ private <T> ListenableFuture<T> doConnect(final Request request, final AsyncHand
             directInvokation = false;
         }
 
-        if (directInvokation && !executeConnectAsync && request.getFile() == null) {
+        if (directInvokation && !asyncConnect && request.getFile() == null) {
             int timeOut = config.getConnectionTimeoutInMs() > 0 ? config.getConnectionTimeoutInMs() : Integer.MAX_VALUE;
             if (!channelFuture.awaitUninterruptibly(timeOut, TimeUnit.MILLISECONDS)) {
                 abort(c.future(), new ConnectException("Connect times out"));
@@ -906,7 +906,6 @@ public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) thr
             HttpChunk chunk = (HttpChunk) e.getMessage();
             if (chunk.isLast()) {
                 AsyncCallable ac = (AsyncCallable) ctx.getAttachment();
-                ctx.setAttachment(ac.future());
                 ac.call();
             }
             return;
@@ -1049,17 +1048,7 @@ public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) thr
                         future.attachChannel(ctx.getChannel(), true);
                     }
 
-                    // We must consume the body first in order to re-use the connection.
-                    if (response.isChunked()) {
-                        ctx.setAttachment(new AsyncCallable(future) {
-                            public Object call() throws Exception {
-                                nextRequest(builder.setHeaders(headers).setRealm(nr).build(), future);
-                                return null;
-                            }
-                        });
-                    } else {
-                        nextRequest(builder.setHeaders(headers).setRealm(nr).build(), future);
-                    }
+                    nextRequest(builder.setHeaders(headers).setRealm(nr).build(), future);
                     return;
                 }
 
@@ -1078,16 +1067,7 @@ public Object call() throws Exception {
 
                     log.debug("Sending proxy authentication to {}", request.getUrl());
 
-                    if (response.isChunked()) {
-                        ctx.setAttachment(new AsyncCallable(future) {
-                            public Object call() throws Exception {
-                                nextRequest(future.getRequest(), future);
-                                return null;
-                            }
-                        });
-                    } else {
-                        nextRequest(future.getRequest(), future);
-                    }
+                    nextRequest(future.getRequest(), future);
                     return;
                 }
 
@@ -1125,36 +1105,18 @@ public Object call() throws Exception {
                         if (!uri.toString().equalsIgnoreCase(future.getURI().toString())) {
                             final RequestBuilder builder = new RequestBuilder(future.getRequest());
                             final URI initialConnectionUri = future.getURI();
-                            final boolean initialConnectionKeepAlive = future.getKeepAlive();
                             future.setURI(uri);
                             final String newUrl = uri.toString();
 
                             log.debug("Redirecting to {}", newUrl);
 
                             if (response.isChunked()) {
-                                ctx.setAttachment(new AsyncCallable(future) {
-                                    public Object call() throws Exception {
-                                        if (initialConnectionKeepAlive) {
-                                            if (!connectionsPool.offer(AsyncHttpProviderUtils.getBaseUrl(initialConnectionUri), ctx.getChannel())) {
-                                                finishChannel(ctx);
-                                            }
-                                        } else {
-                                            closeChannel(ctx);
-                                        }
-                                        nextRequest(builder.setUrl(newUrl).build(), future);
-                                        return null;
-                                    }
-                                });
+                                drainChannel(ctx, future, initialConnectionUri);
                             } else {
-                                if (initialConnectionKeepAlive) {
-                                    if (!connectionsPool.offer(AsyncHttpProviderUtils.getBaseUrl(initialConnectionUri), ctx.getChannel())) {
-                                        finishChannel(ctx);
-                                    }
-                                } else {
-                                    closeChannel(ctx);
-                                }
-                                nextRequest(builder.setUrl(newUrl).build(), future);
+                                closeChannel(ctx);
                             }
+
+                            nextRequest(builder.setUrl(newUrl).build(), future);
                             return;
                         }
                     } else {
@@ -1179,7 +1141,9 @@ public Object call() throws Exception {
 
                 if (nettyRequest.getMethod().equals(HttpMethod.HEAD)) {
                     updateBodyAndInterrupt(handler, new ResponseBodyPart(future.getURI(), response, this));
-                    markAsDoneAndCacheConnection(future, ctx, ctx.getChannel().isReadable());
+                    markAsDoneAndCacheConnection(future, ctx);
+                    drainChannel(ctx, future, future.getURI());
+                    return;
                 }
 
             } else if (e.getMessage() instanceof HttpChunk) {
@@ -1216,6 +1180,21 @@ public Object call() throws Exception {
         }
     }
 
+    private void drainChannel(final ChannelHandlerContext ctx, final NettyResponseFuture<?> future, final URI uri){
+        ctx.setAttachment(new AsyncCallable(future) {
+            public Object call() throws Exception {
+                if (future.getKeepAlive() && ctx.getChannel().isReadable()) {
+                    if (!connectionsPool.offer(AsyncHttpProviderUtils.getBaseUrl(uri), ctx.getChannel())) {
+                        finishChannel(ctx);
+                    }
+                } else {
+                    finishChannel(ctx);
+                }
+                return null;
+            }
+        });
+    }
+
     private FilterContext handleIoException(FilterContext fc, NettyResponseFuture<?> future) {
         for (IOExceptionFilter asyncFilter : config.getIOExceptionFilters()) {
             try {
@@ -1237,18 +1216,8 @@ private void replayRequest(final NettyResponseFuture<?> future, FilterContext fc
         future.touch();
 
         log.debug("\n\nReplaying Request {}\n for Future {}\n", newRequest, future);
-
-        // We must consume the body first in order to re-use the connection.
-        if (response != null && response.isChunked()) {
-            ctx.setAttachment(new AsyncCallable(future) {
-                public Object call() throws Exception {
-                    nextRequest(newRequest, future);
-                    return null;
-                }
-            });
-        } else {
-            nextRequest(newRequest, future);
-        }
+        drainChannel(ctx, future, future.getURI());
+        nextRequest(newRequest, future);
         return;
     }
 
@@ -1267,7 +1236,7 @@ private void nextRequest(final Request request, final NettyResponseFuture<?> fut
     }
 
     private void nextRequest(final Request request, final NettyResponseFuture<?> future, final boolean useCache) throws IOException {
-        execute(request, future, useCache);
+        execute(request, future, useCache, true);
     }
 
     private void abort(NettyResponseFuture<?> future, Throwable t) {
@@ -1386,20 +1355,10 @@ protected boolean remotelyClosed(Channel channel, NettyResponseFuture<?> future)
         return false;
     }
 
-    private void markAsDoneAndCacheConnection(final NettyResponseFuture<?> future, final ChannelHandlerContext ctx,
-                                              final boolean cache) throws MalformedURLException {
+    private void markAsDoneAndCacheConnection(final NettyResponseFuture<?> future, final ChannelHandlerContext ctx) throws MalformedURLException {
         // We need to make sure everything is OK before adding the connection back to the pool.
         try {
-            future.done(new Callable<Boolean>() {
-                public Boolean call() throws Exception {
-                    if (future.getKeepAlive() && cache) {
-                        if (!connectionsPool.offer(AsyncHttpProviderUtils.getBaseUrl(future.getURI()), ctx.getChannel())) {
-                            finishChannel(ctx);
-                        }
-                    }
-                    return false;
-                }
-            });
+            future.done(null);
         } catch (Throwable t) {
             // Never propagate exception once we know we are done.
             log.debug(t.getMessage(), t);
@@ -1412,15 +1371,15 @@ public Boolean call() throws Exception {
 
     private void finishUpdate(final NettyResponseFuture<?> future, final ChannelHandlerContext ctx, boolean isChunked) throws IOException {
         if (isChunked && future.getKeepAlive()) {
-            ctx.setAttachment(new AsyncCallable(future) {
-                public Object call() throws Exception {
-                    markAsDoneAndCacheConnection(future, ctx, ctx.getChannel().isReadable());
-                    return null;
-                }
-            });
+            drainChannel(ctx, future, future.getURI());
         } else {
-            markAsDoneAndCacheConnection(future, ctx, markChannelNotReadable(ctx));
+             if (future.getKeepAlive() && ctx.getChannel().isReadable()) {
+                if (!connectionsPool.offer(AsyncHttpProviderUtils.getBaseUrl(future.getURI()), ctx.getChannel())) {
+                    finishChannel(ctx);
+                }
+            }
         }
+        markAsDoneAndCacheConnection(future, ctx);
     }
 
     private boolean markChannelNotReadable(final ChannelHandlerContext ctx) {

src/main/java/com/ning/http/multipart/Part.java

@@ -23,7 +23,7 @@
  * This class is an adaptation of the Apache HttpClient implementation
  * @link http://hc.apache.org/httpclient-3.x/
  */
-public abstract class Part {
+public abstract class Part implements com.ning.http.client.Part {
 
     /**
      * The boundary

src/main/java/com/ning/http/util/AsyncHttpProviderUtils.java

@@ -194,7 +194,9 @@ public final static MultipartRequestEntity createMultipartRequestEntity(List<Par
         int i = 0;
 
         for (Part part : params) {
-            if (part instanceof StringPart) {
+            if (part instanceof com.ning.http.multipart.Part) {
+                parts[i] = (com.ning.http.multipart.Part) part;
+            } else if (part instanceof StringPart) {
                 parts[i] = new com.ning.http.multipart.StringPart(part.getName(),
                         ((StringPart) part).getValue(),
                         "UTF-8");

src/test/java/com/ning/http/client/async/RetryNonBlockingIssue.java

@@ -27,7 +27,6 @@
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
-
 import javax.servlet.ServletException;
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
@@ -106,7 +105,7 @@ public void stop() {
     }
 
     private ListenableFuture<Response> testMethodRequest(AsyncHttpClient
-            fetcher, int requests,String action, String id) throws IOException {
+            fetcher, int requests, String action, String id) throws IOException {
         RequestBuilder builder = new RequestBuilder("GET");
         builder.addQueryParameter(action, "1");
 
@@ -137,7 +136,7 @@ public void testRetryNonBlocking() throws IOException, InterruptedException,
 
             bc.setAllowPoolingConnection(true);
             bc.setMaximumConnectionsTotal(100);
-            bc.setConnectionTimeoutInMs(30000);
+            bc.setConnectionTimeoutInMs(60000);
             bc.setRequestTimeoutInMs(30000);
 
             NettyAsyncHttpProviderConfig config = new
@@ -146,13 +145,54 @@ public void testRetryNonBlocking() throws IOException, InterruptedException,
             bc.setAsyncHttpClientProviderConfig(config);
             c = new AsyncHttpClient(bc.build());
 
-            res.add(testMethodRequest(c,
-                    3, "servlet", UUID.randomUUID().toString()));
-            res.add(testMethodRequest(c, 3, "io", UUID.randomUUID().toString()));
-            res.add(testMethodRequest(c,
-                    3, "normal", UUID.randomUUID().toString()));
-            res.add(testMethodRequest(c, 3, "500", UUID.randomUUID().toString()));
-            res.add(testMethodRequest(c, 3, "500", UUID.randomUUID().toString()));
+            for (int i = 0; i < 32; i++) {
+                res.add(testMethodRequest(c, 3, "servlet", UUID.randomUUID().toString()));
+            }
+
+            StringBuilder b = new StringBuilder();
+            for (ListenableFuture<Response> r : res) {
+                Response theres = r.get();
+                b.append("==============\r\n");
+                b.append("Response Headers\r\n");
+                Map<String, List<String>> heads = theres.getHeaders();
+                b.append(heads + "\r\n");
+                b.append("==============\r\n");
+                assertTrue(heads.size() > 0);
+            }
+            System.out.println(b.toString());
+            System.out.flush();
+
+        }
+        finally {
+            if (c != null) c.close();
+        }
+    }
+
+    @Test
+    public void testRetryNonBlockingAsyncConnect() throws IOException, InterruptedException,
+            ExecutionException {
+        AsyncHttpClient c = null;
+        List<ListenableFuture<Response>> res = new
+                ArrayList<ListenableFuture<Response>>();
+        try {
+            AsyncHttpClientConfig.Builder bc =
+                    new AsyncHttpClientConfig.Builder();
+
+            bc.setAllowPoolingConnection(true);
+            bc.setMaximumConnectionsTotal(100);
+            bc.setConnectionTimeoutInMs(60000);
+            bc.setRequestTimeoutInMs(30000);
+
+            NettyAsyncHttpProviderConfig config = new
+                    NettyAsyncHttpProviderConfig();
+            config.addProperty(NettyAsyncHttpProviderConfig.EXECUTE_ASYNC_CONNECT, "true");
+
+            bc.setAsyncHttpClientProviderConfig(config);
+            c = new AsyncHttpClient(bc.build());
+
+            for (int i = 0; i < 32; i++) {
+                res.add(testMethodRequest(c, 3, "servlet", UUID.randomUUID().toString()));
+            }
 
             StringBuilder b = new StringBuilder();
             for (ListenableFuture<Response> r : res) {
@@ -191,17 +231,13 @@ public void testRetryBlocking() throws IOException, InterruptedException,
             NettyAsyncHttpProviderConfig config = new
                     NettyAsyncHttpProviderConfig();
             config.addProperty(NettyAsyncHttpProviderConfig.USE_BLOCKING_IO, "true");
-            
+
             bc.setAsyncHttpClientProviderConfig(config);
             c = new AsyncHttpClient(bc.build());
 
-            res.add(testMethodRequest(c,
-                    3, "servlet", UUID.randomUUID().toString()));
-            res.add(testMethodRequest(c, 3, "io", UUID.randomUUID().toString()));
-            res.add(testMethodRequest(c,
-                    3, "normal", UUID.randomUUID().toString()));
-            res.add(testMethodRequest(c, 3, "500", UUID.randomUUID().toString()));
-            res.add(testMethodRequest(c, 3, "500", UUID.randomUUID().toString()));
+            for (int i = 0; i < 32; i++) {
+                res.add(testMethodRequest(c, 3, "servlet", UUID.randomUUID().toString()));
+            }
 
             StringBuilder b = new StringBuilder();
             for (ListenableFuture<Response> r : res) {