Port changes from 1.7.

Author: rlubke

api/src/test/java/org/asynchttpclient/async/NoNullResponseTest.java

@@ -39,7 +39,7 @@ public void multipleSslRequestsWithDelayAndKeepAlive() throws Throwable {
         try {
             final BoundRequestBuilder builder = client.prepareGet(GOOGLE_HTTPS_URL);
             final Response response1 = builder.execute().get();
-            Thread.sleep(5000);
+            Thread.sleep(4000);
             final Response response2 = builder.execute().get();
             if (response2 != null) {
                 System.out.println("Success (2nd response was not null).");

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/FeedableBodyGenerator.java

@@ -12,17 +12,27 @@
  */
 package org.asynchttpclient.providers.grizzly;
 
-import org.asynchttpclient.Body;
-import org.asynchttpclient.BodyGenerator;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
+
+import org.asynchttpclient.Body;
+import org.asynchttpclient.BodyGenerator;
 import org.glassfish.grizzly.Buffer;
+import org.glassfish.grizzly.Connection;
+import org.glassfish.grizzly.WriteHandler;
 import org.glassfish.grizzly.filterchain.FilterChainContext;
 import org.glassfish.grizzly.http.HttpContent;
 import org.glassfish.grizzly.http.HttpRequestPacket;
+import org.glassfish.grizzly.impl.FutureImpl;
+import org.glassfish.grizzly.utils.Futures;
+
+import static java.lang.Boolean.TRUE;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.glassfish.grizzly.utils.Exceptions.*;
 
 /**
  * {@link BodyGenerator} which may return just part of the payload at the time
@@ -38,51 +48,145 @@ public class FeedableBodyGenerator implements BodyGenerator {
 
     private volatile HttpRequestPacket requestPacket;
     private volatile FilterChainContext context;
-    
+    private volatile HttpContent.Builder contentBuilder;
+
+    private final EmptyBody EMPTY_BODY = new EmptyBody();
+
+
+
+    // ---------------------------------------------- Methods from BodyGenerator
+
+
     @Override
     public Body createBody() throws IOException {
-        return new EmptyBody();
+        return EMPTY_BODY;
     }
-    
+
+
+    // ---------------------------------------------------------- Public Methods
+
+
+    /**
+     * Feeds the specified buffer.  This buffer may be queued to be sent later
+     * or sent immediately.  Note that this method may block if data is being
+     * fed faster than it is being consumed by the peer.
+     *
+     * The maximum duration that this method may block is dependent on
+     * the current value of {@link org.glassfish.grizzly.Transport#getWriteTimeout(java.util.concurrent.TimeUnit)}.
+     * This value can be customized by using a {@link TransportCustomizer} to
+     * fine-tune the transport used by the client instance.
+     *
+     * @param buffer the {@link Buffer} to feed.
+     * @param last flag indicating if this is the final buffer of the message.
+     * @throws IOException if an I/O error occurs.
+     *
+     * @see TransportCustomizer
+     * @see GrizzlyAsyncHttpProviderConfig#addProperty(GrizzlyAsyncHttpProviderConfig.Property, Object)
+     * @see GrizzlyAsyncHttpProviderConfig.Property#TRANSPORT_CUSTOMIZER
+     */
     @SuppressWarnings("UnusedDeclaration")
-    public void feed(final Buffer buffer, final boolean isLast)
-            throws IOException {
-        queue.offer(new BodyPart(buffer, isLast));
+    public void feed(final Buffer buffer, final boolean last)
+    throws IOException {
+        queue.offer(new BodyPart(buffer, last));
         queueSize.incrementAndGet();
 
         if (context != null) {
-            flushQueue();            
+            flushQueue(true);
         }
     }
-    
+
     public void initializeAsynchronousTransfer(final FilterChainContext context,
-            final HttpRequestPacket requestPacket) {
+                                        final HttpRequestPacket requestPacket)
+    throws IOException {
         this.context = context;
         this.requestPacket = requestPacket;
-        flushQueue();
+        this.contentBuilder = HttpContent.builder(requestPacket);
+        // don't block here.  If queue is full at the time of the next feed()
+        // call, it will block.
+        flushQueue(false);
     }
 
+
+    // --------------------------------------------------------- Private Methods
+
+
     @SuppressWarnings("unchecked")
-    private void flushQueue() {
+    private void flushQueue(final boolean allowBlocking) throws IOException {
         if (queueSize.get() > 0) {
             synchronized(this) {
+                final Connection c = context.getConnection();
                 while(queueSize.get() > 0) {
+                    if (allowBlocking) {
+                        blockUntilQueueFree(c);
+                    }
                     final BodyPart bodyPart = queue.poll();
                     queueSize.decrementAndGet();
                     final HttpContent content =
-                            requestPacket.httpContentBuilder()
-                            .content(bodyPart.buffer)
-                            .last(bodyPart.isLast)
+                            contentBuilder.content(bodyPart.buffer)
+                                    .last(bodyPart.isLast)
                             .build();
-                    context.write(content, ((!requestPacket.isCommitted()) ?
-                            context.getTransportContext().getCompletionHandler() :
-                            null));
-                    
+                    context.write(content,
+                                  ((!requestPacket.isCommitted())
+                                          ? context.getTransportContext()
+                                                .getCompletionHandler()
+                                          : null));
                 }
             }
         }
     }
-    
+
+    /**
+     * This method will block if the async write queue is currently larger
+     * than the configured maximum.  The amount of time that this method
+     * will block is dependent on the write timeout of the transport
+     * associated with the specified connection.
+     */
+    private void blockUntilQueueFree(final Connection c) {
+        if (!c.canWrite()) {
+            final FutureImpl<Boolean> future =
+                    Futures.createSafeFuture();
+
+            // Connection may be obtained by calling FilterChainContext.getConnection().
+            c.notifyCanWrite(new WriteHandler() {
+
+                @Override
+                public void onWritePossible() throws Exception {
+                    future.result(TRUE);
+                }
+
+                @Override
+                public void onError(Throwable t) {
+                    future.failure(makeIOException(t));
+                }
+            });
+
+            block(c, future);
+        }
+    }
+
+    private void block(final Connection c,
+                       final FutureImpl<Boolean> future) {
+        try {
+            final long writeTimeout =
+                    c.getTransport().getWriteTimeout(MILLISECONDS);
+            if (writeTimeout != -1) {
+                future.get(writeTimeout, MILLISECONDS);
+            } else {
+                future.get();
+            }
+        } catch (ExecutionException e) {
+            HttpTransactionContext httpCtx = HttpTransactionContext.get(c);
+            httpCtx.abort(e.getCause());
+        } catch (Exception e) {
+            HttpTransactionContext httpCtx = HttpTransactionContext.get(c);
+            httpCtx.abort(e);
+        }
+    }
+
+
+    // ----------------------------------------------------------- Inner Classes
+
+
     private final class EmptyBody implements Body {
 
         @Override
@@ -100,9 +204,14 @@ public void close() throws IOException {
             context.completeAndRecycle();
             context = null;
             requestPacket = null;
+            contentBuilder = null;
         }
     }
-    
+
+
+    // ---------------------------------------------------------- Nested Classes
+
+
     private final static class BodyPart {
         private final boolean isLast;
         private final Buffer buffer;

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/GrizzlyAsyncHttpProvider.java

@@ -86,6 +86,7 @@
 import static org.asynchttpclient.providers.grizzly.GrizzlyAsyncHttpProviderConfig.Property;
 import static org.asynchttpclient.providers.grizzly.GrizzlyAsyncHttpProviderConfig.Property.CONNECTION_POOL;
 import static org.asynchttpclient.providers.grizzly.GrizzlyAsyncHttpProviderConfig.Property.MAX_HTTP_PACKET_HEADER_SIZE;
+import static org.glassfish.grizzly.asyncqueue.AsyncQueueWriter.AUTO_SIZE;
 
 /**
  * A Grizzly 2.0-based implementation of {@link AsyncHttpProvider}.
@@ -182,7 +183,7 @@ public void close() {
 
         try {
             connectionManager.destroy();
-            clientTransport.stop();
+            clientTransport.shutdownNow();
             final ExecutorService service = clientConfig.executorService();
             // service may be null due to a custom configuration that
             // leverages Grizzly's SameThreadIOStrategy.
@@ -254,6 +255,7 @@ void initializeTransport(final AsyncHttpClientConfig clientConfig) {
         final int timeout = clientConfig.getRequestTimeoutInMs();
         if (timeout > 0) {
             int delay = 500;
+            //noinspection ConstantConditions
             if (timeout < delay) {
                 delay = timeout - 10;
                 if (delay <= 0) {
@@ -382,7 +384,8 @@ public void onTimeout(Connection connection) {
         }
 
         // Don't limit the number of bytes the client can have queued to write.
-        clientTransport.getAsyncQueueIO().getWriter().setMaxPendingBytesPerConnection(-1);
+        clientTransport.getAsyncQueueIO().getWriter()
+                .setMaxPendingBytesPerConnection(AUTO_SIZE);
 
         // Install the HTTP filter chain.
         //clientTransport.setProcessor(fcb.build());