Add handler callbacks, close AsyncHttpClient#673

Author: Stephane Landelle

api/src/main/java/org/asynchttpclient/AsyncHandlerExtensions.java

@@ -19,21 +19,39 @@
  * 
  * More additional hooks might come, such as:
  * <ul>
- *   <li>onConnected()</li>
  *   <li>onConnectionClosed()</li>
  *   <li>onBytesSent(long numberOfBytes)</li>
  *   <li>onBytesReceived(long numberOfBytes)</li>
  * </ul>
  */
 public interface AsyncHandlerExtensions {
 
+    /**
+     * 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 being 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.

providers/netty/src/main/java/org/asynchttpclient/providers/netty/request/NettyConnectListener.java

@@ -13,9 +13,21 @@
  */
 package org.asynchttpclient.providers.netty.request;
 
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.handler.ssl.SslHandler;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 
+import java.net.ConnectException;
+import java.nio.channels.ClosedChannelException;
+
+import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLSession;
+
+import org.asynchttpclient.AsyncHandlerExtensions;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.providers.netty.channel.ChannelManager;
 import org.asynchttpclient.providers.netty.channel.Channels;
@@ -25,18 +37,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.handler.ssl.SslHandler;
-
-import javax.net.ssl.HostnameVerifier;
-import javax.net.ssl.SSLEngine;
-import javax.net.ssl.SSLSession;
-
-import java.net.ConnectException;
-import java.nio.channels.ClosedChannelException;
-
 /**
  * Non Blocking connect.
  */
@@ -79,6 +79,9 @@ private void writeRequest(Channel channel) {
             return;
         }
 
+        if (future.getAsyncHandler() instanceof AsyncHandlerExtensions)
+            AsyncHandlerExtensions.class.cast(future.getAsyncHandler()).onConnectionOpen();
+
         channelManager.registerOpenChannel(channel);
         future.attachChannel(channel, false);
         requestSender.writeRequest(future, channel);

providers/netty/src/main/java/org/asynchttpclient/providers/netty/request/NettyRequestSender.java

@@ -159,8 +159,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
@@ -190,12 +189,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);
 
@@ -254,6 +256,9 @@ private <T> ListenableFuture<T> sendRequestWithNewChannel(//
             channelPreempted = preemptChannel(asyncHandler, poolKey);
         }
 
+        if (asyncHandler instanceof AsyncHandlerExtensions)
+            AsyncHandlerExtensions.class.cast(asyncHandler).onOpenConnection();
+
         try {
             ChannelFuture channelFuture = connect(request, uri, proxy, useProxy, bootstrap);
             channelFuture.addListener(new NettyConnectListener<T>(config, future, this, channelManager, channelPreempted, poolKey));
@@ -303,7 +308,7 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
             if (!future.isHeadersAlreadyWrittenOnContinue()) {
                 try {
                     if (future.getAsyncHandler() instanceof AsyncHandlerExtensions)
-                        AsyncHandlerExtensions.class.cast(future.getAsyncHandler()).onRequestSent();
+                        AsyncHandlerExtensions.class.cast(future.getAsyncHandler()).onSendRequest();
 
                     channel.writeAndFlush(httpRequest, channel.newProgressivePromise()).addListener(new ProgressListener(config, future.getAsyncHandler(), future, true, 0L));
                 } catch (Throwable cause) {
@@ -474,7 +479,11 @@ private boolean validateWebSocketRequest(Request request, AsyncHandler<?> asyncH
         return request.getMethod().equals(HttpMethod.GET.name()) && asyncHandler instanceof WebSocketUpgradeHandler;
     }
 
-    private Channel pollAndVerifyCachedChannel(UriComponents uri, ProxyServer proxy, ConnectionPoolKeyStrategy connectionPoolKeyStrategy) {
+    private 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) {