Add handler callbacks, close AsyncHttpClient#673

Author: Stephane Landelle

src/main/java/com/ning/http/client/AsyncHandlerExtensions.java

@@ -19,7 +19,6 @@
  * 
  * More additional hooks might come, such as:
  * <ul>
- *   <li>onConnected()</li>
  *   <li>onConnectionClosed()</li>
  *   <li>onBytesSent(long numberOfBytes)</li>
  *   <li>onBytesReceived(long numberOfBytes)</li>
@@ -28,12 +27,31 @@
 public interface AsyncHandlerExtensions {
 
     /**
-     * Notify the callback when a request is being written on the wire.
+     * Notify the callback when trying to open a new connection.
+     */
+    void onOpenConnection();
+
+    /**
+     * Notify the callback when a new connection was successfully opened.
+     */
+    void onConnectionOpen();
+
+    /**
+     * Notify the callback when trying to fetch a connection from the pool.
+     */
+    void onPoolConnection();
+
+    /**
+     * Notify the callback when a new connection was successfully fetched from the pool.
+     */
+    void onConnectionPooled();
+
+    /**
+     * Notify the callback when a request is about to be written on the wire.
      * If the original request causes multiple requests to be sent, for example, because of authorization or retry,
      * it will be notified multiple times.
-     * Currently only supported by the Netty provider.
      */
-    void onRequestSent();
+    void onSendRequest();
 
     /**
      * Notify the callback every time a request is being retried.

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

@@ -1174,7 +1174,7 @@ protected void onHttpHeadersEncoded(HttpHeader httpHeader, FilterChainContext ct
                 ((TransferCompletionHandler) handler).onHeaderWriteCompleted();
             }
             if (handler instanceof AsyncHandlerExtensions) {
-                ((AsyncHandlerExtensions) handler).onRequestSent();
+                ((AsyncHandlerExtensions) handler).onSendRequest();
             }
         }
 

src/main/java/com/ning/http/client/providers/netty/request/NettyRequestSender.java

@@ -162,8 +162,7 @@ private <T> ListenableFuture<T> sendRequestThroughSslProxy(//
                     newFuture = newNettyRequestAndResponseFuture(request, asyncHandler, future, uri, proxyServer, false);
 
             if (Channels.isChannelValid(channel))
-                // if the channel is still active, we can use it, otherwise try
-                // gain
+                // if the channel is still active, we can use it, otherwise try gain
                 return sendRequestWithCachedChannel(request, uri, proxyServer, newFuture, asyncHandler, channel);
             else
                 // pool is empty
@@ -194,12 +193,15 @@ private Channel getCachedChannel(NettyResponseFuture<?> future, UriComponents ur
         if (future != null && future.reuseChannel() && Channels.isChannelValid(future.channel()))
             return future.channel();
         else
-            return pollAndVerifyCachedChannel(uri, proxyServer, poolKeyGen);
+            return pollAndVerifyCachedChannel(uri, proxyServer, poolKeyGen, future.getAsyncHandler());
     }
 
     private <T> ListenableFuture<T> sendRequestWithCachedChannel(Request request, UriComponents uri, ProxyServer proxy,
             NettyResponseFuture<T> future, AsyncHandler<T> asyncHandler, Channel channel) throws IOException {
 
+        if (asyncHandler instanceof AsyncHandlerExtensions)
+            AsyncHandlerExtensions.class.cast(asyncHandler).onConnectionPooled();
+
         future.setState(NettyResponseFuture.STATE.POOLED);
         future.attachChannel(channel, false);
 
@@ -258,6 +260,9 @@ private <T> ListenableFuture<T> sendRequestWithNewChannel(//
         }
 
         try {
+            if (asyncHandler instanceof AsyncHandlerExtensions)
+                AsyncHandlerExtensions.class.cast(asyncHandler).onOpenConnection();
+
             ChannelFuture channelFuture = connect(request, uri, proxy, useProxy, bootstrap);
             channelFuture.addListener(new NettyConnectListener<T>(config, future, this, channelManager, channelPreempted, poolKey));
 
@@ -306,9 +311,8 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
 
             if (!future.isHeadersAlreadyWrittenOnContinue()) {
                 try {
-                    if (future.getAsyncHandler() instanceof AsyncHandlerExtensions) {
-                        AsyncHandlerExtensions.class.cast(future.getAsyncHandler()).onRequestSent();
-                    }
+                    if (future.getAsyncHandler() instanceof AsyncHandlerExtensions)
+                        AsyncHandlerExtensions.class.cast(future.getAsyncHandler()).onSendRequest();
                     channel.write(httpRequest).addListener(new ProgressListener(config, future.getAsyncHandler(), future, true));
                 } catch (Throwable cause) {
                     // FIXME why not notify?
@@ -481,7 +485,11 @@ private boolean validateWebSocketRequest(Request request, AsyncHandler<?> asyncH
         return request.getMethod().equals(HttpMethod.GET.getName()) && asyncHandler instanceof WebSocketUpgradeHandler;
     }
 
-    public Channel pollAndVerifyCachedChannel(UriComponents uri, ProxyServer proxy, ConnectionPoolKeyStrategy connectionPoolKeyStrategy) {
+    public Channel pollAndVerifyCachedChannel(UriComponents uri, ProxyServer proxy, ConnectionPoolKeyStrategy connectionPoolKeyStrategy, AsyncHandler<?> asyncHandler) {
+ 
+        if (asyncHandler instanceof AsyncHandlerExtensions)
+            AsyncHandlerExtensions.class.cast(asyncHandler).onPoolConnection();
+
         final Channel channel = channelManager.poll(connectionPoolKeyStrategy.getKey(uri, proxy));
 
         if (channel != null) {

src/main/java/com/ning/http/client/providers/netty/request/body/NettyConnectListener.java

@@ -20,6 +20,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.ning.http.client.AsyncHandlerExtensions;
 import com.ning.http.client.AsyncHttpClientConfig;
 import com.ning.http.client.providers.netty.channel.ChannelManager;
 import com.ning.http.client.providers.netty.channel.Channels;
@@ -79,6 +80,9 @@ private void writeRequest(Channel channel, String poolKey) {
             return;
         }
 
+        if (future.getAsyncHandler() instanceof AsyncHandlerExtensions)
+            AsyncHandlerExtensions.class.cast(future.getAsyncHandler()).onConnectionOpen();
+
         channelManager.registerOpenChannel(channel);
         future.attachChannel(channel, false);
         requestSender.writeRequest(future, channel);