Provide a way to configure ChannelPipelines, close AsyncHttpClient#717

Author: Stephane Landelle

providers/netty/src/main/java/org/asynchttpclient/providers/netty/NettyAsyncHttpProviderConfig.java

@@ -16,6 +16,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.Timer;
 
@@ -87,9 +88,9 @@ public Set<Map.Entry<ChannelOption<Object>, Object>> propertiesSet() {
         return properties.entrySet();
     }
 
-    public static interface AdditionalChannelInitializer {
+    public static interface AdditionalPipelineInitializer {
 
-        void initChannel(Channel ch) throws Exception;
+        void initPipeline(ChannelPipeline pipeline) throws Exception;
     }
 
     public static interface ResponseBodyPartFactory {
@@ -130,10 +131,10 @@ public NettyWebSocket newNettyWebSocket(Channel channel, NettyAsyncHttpProviderC
      */
     private EventLoopGroup eventLoopGroup;
 
-    private AdditionalChannelInitializer httpAdditionalChannelInitializer;
-    private AdditionalChannelInitializer wsAdditionalChannelInitializer;
-    private AdditionalChannelInitializer httpsAdditionalChannelInitializer;
-    private AdditionalChannelInitializer wssAdditionalChannelInitializer;
+    private AdditionalPipelineInitializer httpAdditionalPipelineInitializer;
+    private AdditionalPipelineInitializer wsAdditionalPipelineInitializer;
+    private AdditionalPipelineInitializer httpsAdditionalPipelineInitializer;
+    private AdditionalPipelineInitializer wssAdditionalPipelineInitializer;
 
     /**
      * Allow configuring Netty's HttpClientCodecs.
@@ -176,36 +177,36 @@ public void setEventLoopGroup(EventLoopGroup eventLoopGroup) {
         this.eventLoopGroup = eventLoopGroup;
     }
 
-    public AdditionalChannelInitializer getHttpAdditionalChannelInitializer() {
-        return httpAdditionalChannelInitializer;
+    public AdditionalPipelineInitializer getHttpAdditionalPipelineInitializer() {
+        return httpAdditionalPipelineInitializer;
     }
 
-    public void setHttpAdditionalChannelInitializer(AdditionalChannelInitializer httpAdditionalChannelInitializer) {
-        this.httpAdditionalChannelInitializer = httpAdditionalChannelInitializer;
+    public void setHttpAdditionalPipelineInitializer(AdditionalPipelineInitializer httpAdditionalPipelineInitializer) {
+        this.httpAdditionalPipelineInitializer = httpAdditionalPipelineInitializer;
     }
 
-    public AdditionalChannelInitializer getWsAdditionalChannelInitializer() {
-        return wsAdditionalChannelInitializer;
+    public AdditionalPipelineInitializer getWsAdditionalPipelineInitializer() {
+        return wsAdditionalPipelineInitializer;
     }
 
-    public void setWsAdditionalChannelInitializer(AdditionalChannelInitializer wsAdditionalChannelInitializer) {
-        this.wsAdditionalChannelInitializer = wsAdditionalChannelInitializer;
+    public void setWsAdditionalPipelineInitializer(AdditionalPipelineInitializer wsAdditionalPipelineInitializer) {
+        this.wsAdditionalPipelineInitializer = wsAdditionalPipelineInitializer;
     }
 
-    public AdditionalChannelInitializer getHttpsAdditionalChannelInitializer() {
-        return httpsAdditionalChannelInitializer;
+    public AdditionalPipelineInitializer getHttpsAdditionalPipelineInitializer() {
+        return httpsAdditionalPipelineInitializer;
     }
 
-    public void setHttpsAdditionalChannelInitializer(AdditionalChannelInitializer httpsAdditionalChannelInitializer) {
-        this.httpsAdditionalChannelInitializer = httpsAdditionalChannelInitializer;
+    public void setHttpsAdditionalPipelineInitializer(AdditionalPipelineInitializer httpsAdditionalPipelineInitializer) {
+        this.httpsAdditionalPipelineInitializer = httpsAdditionalPipelineInitializer;
     }
 
-    public AdditionalChannelInitializer getWssAdditionalChannelInitializer() {
-        return wssAdditionalChannelInitializer;
+    public AdditionalPipelineInitializer getWssAdditionalPipelineInitializer() {
+        return wssAdditionalPipelineInitializer;
     }
 
-    public void setWssAdditionalChannelInitializer(AdditionalChannelInitializer wssAdditionalChannelInitializer) {
-        this.wssAdditionalChannelInitializer = wssAdditionalChannelInitializer;
+    public void setWssAdditionalPipelineInitializer(AdditionalPipelineInitializer wssAdditionalPipelineInitializer) {
+        this.wssAdditionalPipelineInitializer = wssAdditionalPipelineInitializer;
     }
 
     public int getHttpClientCodecMaxInitialLineLength() {

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/ChannelManager.java

@@ -191,8 +191,8 @@ protected void initChannel(Channel ch) throws Exception {
                         .addLast(CHUNKED_WRITER_HANDLER, new ChunkedWriteHandler())//
                         .addLast(HTTP_PROCESSOR, httpProcessor);
 
-                if (nettyConfig.getHttpAdditionalChannelInitializer() != null)
-                    nettyConfig.getHttpAdditionalChannelInitializer().initChannel(ch);
+                if (nettyConfig.getHttpAdditionalPipelineInitializer() != null)
+                    nettyConfig.getHttpAdditionalPipelineInitializer().initPipeline(ch.pipeline());
             }
         });
 
@@ -203,9 +203,8 @@ protected void initChannel(Channel ch) throws Exception {
                         .addLast(HTTP_HANDLER, newHttpClientCodec())//
                         .addLast(WS_PROCESSOR, wsProcessor);
 
-                if (nettyConfig.getWsAdditionalChannelInitializer() != null) {
-                    nettyConfig.getWsAdditionalChannelInitializer().initChannel(ch);
-                }
+                if (nettyConfig.getWsAdditionalPipelineInitializer() != null)
+                    nettyConfig.getWsAdditionalPipelineInitializer().initPipeline(ch.pipeline());
             }
         });
 
@@ -220,8 +219,8 @@ protected void initChannel(Channel ch) throws Exception {
                         .addLast(CHUNKED_WRITER_HANDLER, new ChunkedWriteHandler())//
                         .addLast(HTTP_PROCESSOR, httpProcessor);
 
-                if (nettyConfig.getHttpsAdditionalChannelInitializer() != null)
-                    nettyConfig.getHttpsAdditionalChannelInitializer().initChannel(ch);
+                if (nettyConfig.getHttpsAdditionalPipelineInitializer() != null)
+                    nettyConfig.getHttpsAdditionalPipelineInitializer().initPipeline(ch.pipeline());
             }
         });
 
@@ -234,9 +233,8 @@ protected void initChannel(Channel ch) throws Exception {
                         .addLast(HTTP_HANDLER, newHttpClientCodec())//
                         .addLast(WS_PROCESSOR, wsProcessor);
 
-                if (nettyConfig.getWssAdditionalChannelInitializer() != null) {
-                    nettyConfig.getWssAdditionalChannelInitializer().initChannel(ch);
-                }
+                if (nettyConfig.getWssAdditionalPipelineInitializer() != null)
+                    nettyConfig.getWssAdditionalPipelineInitializer().initPipeline(ch.pipeline());
             }
         });
     }

providers/netty/src/test/java/org/asynchttpclient/providers/netty/NettyAsyncProviderPipelineTest.java

@@ -22,12 +22,12 @@
 import org.asynchttpclient.RequestBuilder;
 import org.asynchttpclient.Response;
 import org.asynchttpclient.async.AbstractBasicTest;
-import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig.AdditionalChannelInitializer;
+import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig.AdditionalPipelineInitializer;
 import org.testng.annotations.Test;
 
-import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.http.HttpMessage;
 
 import java.util.concurrent.CountDownLatch;
@@ -44,10 +44,9 @@ public AsyncHttpClient getAsyncHttpClient(AsyncHttpClientConfig config) {
     public void asyncPipelineTest() throws Exception {
 
         NettyAsyncHttpProviderConfig nettyConfig = new NettyAsyncHttpProviderConfig();
-        nettyConfig.setHttpAdditionalChannelInitializer(new AdditionalChannelInitializer() {
-            public void initChannel(Channel ch) throws Exception {
-                // super.initPlainChannel(ch);
-                ch.pipeline().addBefore("inflater", "copyEncodingHeader", new CopyEncodingHandler());
+        nettyConfig.setHttpAdditionalPipelineInitializer(new AdditionalPipelineInitializer() {
+            public void initPipeline(ChannelPipeline pipeline) throws Exception {
+                pipeline.addBefore("inflater", "copyEncodingHeader", new CopyEncodingHandler());
             }
         });
         AsyncHttpClient p = getAsyncHttpClient(new AsyncHttpClientConfig.Builder()