[1.9.x] + grizzly refactoring and NTLM support

Author: oleksiys

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

@@ -86,39 +86,42 @@ static boolean isConnectionCacheable(final Connection c) {
         return (canCache != null) ? canCache : false;
     }
 
-    void doAsyncTrackedConnection(final Request request,
+    void getConnectionAsync(final Request request,
             final GrizzlyResponseFuture requestFuture,
             final CompletionHandler<Connection> connectHandler)
             throws IOException, ExecutionException, InterruptedException {
 
-        Connection c = pool.poll(getPartitionId(request, requestFuture.getProxy()));
+        final ProxyServer proxy = ProxyUtils.getProxyServer(config, request);
+        
+        Connection c = pool.poll(getPartitionId(request, proxy));
         if (c == null) {
             if (!connectionMonitor.acquire()) {
                 throw new IOException("Max connections exceeded");
             }
-            doAsyncConnect(request, requestFuture, connectHandler);
+            openConnectionAsync(request, proxy, requestFuture, connectHandler);
         } else {
             provider.touchConnection(c, request);
             connectHandler.completed(c);
         }
     }
 
-    Connection obtainConnection(final Request request,
+    Connection openConnectionSync(final Request request,
             final GrizzlyResponseFuture requestFuture)
             throws IOException, ExecutionException,
             InterruptedException, TimeoutException {
 
-        final Connection c = obtainConnection0(request, requestFuture);
+        final Connection c = openConnectionSync0(request,
+                ProxyUtils.getProxyServer(config, request), requestFuture);
         DO_NOT_CACHE.set(c, Boolean.TRUE);
         return c;
     }
 
-    void doAsyncConnect(final Request request,
+    private void openConnectionAsync(final Request request,
+            final ProxyServer proxy,
             final GrizzlyResponseFuture requestFuture,
             final CompletionHandler<Connection> connectHandler)
             throws IOException, ExecutionException, InterruptedException {
 
-        ProxyServer proxy = requestFuture.getProxy();
         final Uri uri = request.getUri();
         String host = (proxy != null) ? proxy.getHost() : uri.getHost();
         int port = (proxy != null) ? proxy.getPort() : uri.getPort();
@@ -132,40 +135,33 @@ void doAsyncConnect(final Request request,
                     HostnameVerifierListener.wrapWithHostnameVerifierHandler(
                             completionHandler, verifier, uri.getHost());
         }
-        if (request.getLocalAddress() != null) {
-            connectionHandler.connect(
-                    new InetSocketAddress(host, getPort(uri, port)),
-                    new InetSocketAddress(request.getLocalAddress(), 0),
-                    completionHandler);
-        } else {
-            connectionHandler.connect(
-                    new InetSocketAddress(host, getPort(uri, port)),
-                    completionHandler);
-        }
+        
+        connectionHandler.connect(
+                new InetSocketAddress(host, getPort(uri, port)),
+                new InetSocketAddress(request.getLocalAddress(), 0),
+                completionHandler);
     }
 
-    private Connection obtainConnection0(final Request request,
+    private Connection openConnectionSync0(final Request request,
+            final ProxyServer proxy,
             final GrizzlyResponseFuture requestFuture)
             throws IOException, ExecutionException,
             InterruptedException, TimeoutException {
 
         final Uri uri = request.getUri();
-        final ProxyServer proxy = requestFuture.getProxy();
         String host = (proxy != null) ? proxy.getHost() : uri.getHost();
         int port = (proxy != null) ? proxy.getPort() : uri.getPort();
         int cTimeout = config.getConnectTimeout();
         FutureImpl<Connection> future = Futures.createSafeFuture();
         CompletionHandler<Connection> ch = Futures.toCompletionHandler(future,
                 createConnectionCompletionHandler(request, requestFuture, null));
-        if (cTimeout > 0) {
-            connectionHandler.connect(
-                    new InetSocketAddress(host, getPort(uri, port)), ch);
-            return future.get(cTimeout, TimeUnit.MILLISECONDS);
-        } else {
-            connectionHandler.connect(
-                    new InetSocketAddress(host, getPort(uri, port)), ch);
-            return future.get();
-        }
+
+        connectionHandler.connect(
+                new InetSocketAddress(host, getPort(uri, port)), ch);
+        
+        return cTimeout > 0
+                ? future.get(cTimeout, TimeUnit.MILLISECONDS)
+                : future.get();
     }
 
     boolean returnConnection(final Request request, final Connection c) {
@@ -188,7 +184,7 @@ void destroy() {
         pool.destroy();
     }
 
-    CompletionHandler<Connection> createConnectionCompletionHandler(
+    private CompletionHandler<Connection> createConnectionCompletionHandler(
             final Request request, final GrizzlyResponseFuture future,
             final CompletionHandler<Connection> wrappedHandler) {
 
@@ -201,16 +197,15 @@ public void cancelled() {
                 }
             }
 
-            public void failed(Throwable throwable) {
+            public void failed(final Throwable throwable) {
                 if (wrappedHandler != null) {
                     wrappedHandler.failed(throwable);
                 } else {
                     future.abort(throwable);
                 }
             }
 
-            public void completed(Connection connection) {
-                future.setConnection(connection);
+            public void completed(final Connection connection) {
                 provider.touchConnection(connection, request);
                 if (wrappedHandler != null) {
                     connection.addCloseListener(connectionMonitor);

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

@@ -37,6 +37,7 @@
 import static java.lang.Boolean.TRUE;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.glassfish.grizzly.ssl.SSLUtils.getSSLEngine;
+import org.glassfish.grizzly.utils.Exceptions;
 import static org.glassfish.grizzly.utils.Exceptions.*;
 
 /**
@@ -177,7 +178,7 @@ public void run() {
                         feeder.flush();
                     }
                 } catch (IOException ioe) {
-                    HttpTransactionContext.currentTransaction(c).abort(ioe);
+                    HttpTransactionContext.currentTransaction(requestPacket).abort(ioe);
                 }
             }
         };
@@ -212,8 +213,8 @@ public void onStart(Connection connection) {
             }
 
             @Override
-            public void onFailure(Connection connection, Throwable t) {
-                HttpTransactionContext.currentTransaction(c).abort(t);
+            public void onFailure(final Connection connection, final Throwable t) {
+                connection.closeWithReason(Exceptions.makeIOException(t));
             }
 
             public void onComplete(Connection connection) {
@@ -222,7 +223,7 @@ public void onComplete(Connection connection) {
                     try {
                         feeder.flush();
                     } catch (IOException ioe) {
-                        HttpTransactionContext.currentTransaction(c).abort(ioe);
+                        connection.closeWithReason(ioe);
                     }
                 }
             }
@@ -379,9 +380,9 @@ private static void block(final Connection c,
                     future.get();
                 }
             } catch (ExecutionException e) {
-                HttpTransactionContext.currentTransaction(c).abort(e.getCause());
+                c.closeWithReason(Exceptions.makeIOException(e.getCause()));
             } catch (Exception e) {
-                HttpTransactionContext.currentTransaction(c).abort(e);
+                c.closeWithReason(Exceptions.makeIOException(e));
             }
         }
 
@@ -587,7 +588,7 @@ public void onWritePossible() throws Exception {
             @Override
             public void onError(Throwable t) {
                 c.setMaxAsyncWriteQueueSize(feedableBodyGenerator.origMaxPendingBytes);
-                HttpTransactionContext.currentTransaction(c).abort(t);
+                c.closeWithReason(Exceptions.makeIOException(t));
             }
 
         } // END WriteHandlerImpl
@@ -607,7 +608,7 @@ public void ready() {
                 } catch (IOException e) {
                     final Connection c = feedableBodyGenerator.context.getConnection();
                     c.setMaxAsyncWriteQueueSize(feedableBodyGenerator.origMaxPendingBytes);
-                    HttpTransactionContext.currentTransaction(c).abort(e);
+                    c.closeWithReason(Exceptions.makeIOException(e));
                 }
             }