Update FeedableBodyGenerator to prevent feed logic execution from blocking the selector thread by checking if the current thread is the selector thread and if true, execute the task on a worker thread.

Author: rlubke

src/main/java/com/ning/http/client/providers/grizzly/FeedableBodyGenerator.java

@@ -29,6 +29,8 @@
 import org.glassfish.grizzly.http.HttpContent;
 import org.glassfish.grizzly.http.HttpRequestPacket;
 import org.glassfish.grizzly.impl.FutureImpl;
+import org.glassfish.grizzly.nio.NIOConnection;
+import org.glassfish.grizzly.nio.SelectorRunner;
 import org.glassfish.grizzly.ssl.SSLBaseFilter;
 import org.glassfish.grizzly.ssl.SSLFilter;
 import org.glassfish.grizzly.utils.Futures;
@@ -169,20 +171,46 @@ synchronized void initializeAsynchronousTransfer(final FilterChainContext contex
         }
         this.context = context;
         asyncTransferInitiated = true;
+        final Runnable r = new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    if (requestPacket.isSecure() &&
+                            (getSSLEngine(context.getConnection()) == null)) {
+                        flushOnSSLHandshakeComplete();
+                    } else {
+                        feeder.flush();
+                    }
+                } catch (IOException ioe) {
+                    GrizzlyAsyncHttpProvider.HttpTransactionContext ctx =
+                            GrizzlyAsyncHttpProvider.getHttpTransactionContext(
+                                    c);
+                    ctx.abort(ioe);
+                }
+            }
+        };
 
-        if (requestPacket.isSecure() &&
-                (getSSLEngine(context.getConnection()) == null)) {
-            flushOnSSLHandshakeComplete();
+        // If the current thread is a selector thread, we need to execute
+        // the remainder of the task on the worker thread to prevent
+        // it from being blocked.
+        if (isCurrentThreadSelectorRunner()) {
+            c.getTransport().getWorkerThreadPool().execute(r);
         } else {
-            feeder.flush();
+            r.run();
         }
-
     }
 
 
     // --------------------------------------------------------- Private Methods
 
 
+    private boolean isCurrentThreadSelectorRunner() {
+        final NIOConnection c = (NIOConnection) context.getConnection();
+        final SelectorRunner runner = c.getSelectorRunner();
+        return (Thread.currentThread() == runner.getRunnerThread());
+    }
+
+
     private void flushOnSSLHandshakeComplete() throws IOException {
         final FilterChain filterChain = context.getFilterChain();
         final int idx = filterChain.indexOfType(SSLFilter.class);
@@ -331,7 +359,6 @@ private static 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() {