Provide a way to configure ChannelPipelines, close AsyncHttpClient#717

Author: Stephane Landelle

src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProviderConfig.java

@@ -14,6 +14,7 @@
 package com.ning.http.client.providers.netty;
 
 import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelPipeline;
 import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
 import org.jboss.netty.util.Timer;
 
@@ -101,6 +102,11 @@ public Set<Map.Entry<String, Object>> propertiesSet() {
      */
     private ExecutorService bossExecutorService;
 
+    private AdditionalPipelineInitializer httpAdditionalPipelineInitializer;
+    private AdditionalPipelineInitializer wsAdditionalPipelineInitializer;
+    private AdditionalPipelineInitializer httpsAdditionalPipelineInitializer;
+    private AdditionalPipelineInitializer wssAdditionalPipelineInitializer;
+
     /**
      * Allow configuring Netty's HttpClientCodecs.
      */
@@ -152,6 +158,38 @@ public ExecutorService getBossExecutorService() {
     public void setBossExecutorService(ExecutorService bossExecutorService) {
         this.bossExecutorService = bossExecutorService;
     }
+    
+    public AdditionalPipelineInitializer getHttpAdditionalPipelineInitializer() {
+        return httpAdditionalPipelineInitializer;
+    }
+
+    public void setHttpAdditionalPipelineInitializer(AdditionalPipelineInitializer httpAdditionalPipelineInitializer) {
+        this.httpAdditionalPipelineInitializer = httpAdditionalPipelineInitializer;
+    }
+
+    public AdditionalPipelineInitializer getWsAdditionalPipelineInitializer() {
+        return wsAdditionalPipelineInitializer;
+    }
+
+    public void setWsAdditionalPipelineInitializer(AdditionalPipelineInitializer wsAdditionalPipelineInitializer) {
+        this.wsAdditionalPipelineInitializer = wsAdditionalPipelineInitializer;
+    }
+
+    public AdditionalPipelineInitializer getHttpsAdditionalPipelineInitializer() {
+        return httpsAdditionalPipelineInitializer;
+    }
+
+    public void setHttpsAdditionalPipelineInitializer(AdditionalPipelineInitializer httpsAdditionalPipelineInitializer) {
+        this.httpsAdditionalPipelineInitializer = httpsAdditionalPipelineInitializer;
+    }
+
+    public AdditionalPipelineInitializer getWssAdditionalPipelineInitializer() {
+        return wssAdditionalPipelineInitializer;
+    }
+
+    public void setWssAdditionalPipelineInitializer(AdditionalPipelineInitializer wssAdditionalPipelineInitializer) {
+        this.wssAdditionalPipelineInitializer = wssAdditionalPipelineInitializer;
+    }
 
     public int getHttpClientCodecMaxInitialLineLength() {
         return httpClientCodecMaxInitialLineLength;
@@ -261,6 +299,11 @@ public static interface NettyWebSocketFactory {
         NettyWebSocket newNettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig);
     }
 
+    public static interface AdditionalPipelineInitializer {
+
+        void initPipeline(ChannelPipeline pipeline) throws Exception;
+    }
+
     public class DefaultNettyWebSocketFactory implements NettyWebSocketFactory {
 
         @Override

src/main/java/com/ning/http/client/providers/netty/channel/ChannelManager.java

@@ -207,6 +207,10 @@ public ChannelPipeline getPipeline() throws Exception {
                 pipeline.addLast(INFLATER_HANDLER, new HttpContentDecompressor());
                 pipeline.addLast(CHUNKED_WRITER_HANDLER, new ChunkedWriteHandler());
                 pipeline.addLast(HTTP_PROCESSOR, httpProcessor);
+
+                if (nettyConfig.getHttpAdditionalPipelineInitializer() != null)
+                    nettyConfig.getHttpAdditionalPipelineInitializer().initPipeline(pipeline);
+
                 return pipeline;
             }
         });
@@ -217,6 +221,10 @@ public ChannelPipeline getPipeline() throws Exception {
                 ChannelPipeline pipeline = pipeline();
                 pipeline.addLast(HTTP_HANDLER, newHttpClientCodec());
                 pipeline.addLast(WS_PROCESSOR, wsProcessor);
+
+                if (nettyConfig.getWsAdditionalPipelineInitializer() != null)
+                    nettyConfig.getWsAdditionalPipelineInitializer().initPipeline(pipeline);
+
                 return pipeline;
             }
         });
@@ -230,6 +238,10 @@ public ChannelPipeline getPipeline() throws Exception {
                 pipeline.addLast(INFLATER_HANDLER, new HttpContentDecompressor());
                 pipeline.addLast(CHUNKED_WRITER_HANDLER, new ChunkedWriteHandler());
                 pipeline.addLast(HTTP_PROCESSOR, httpProcessor);
+
+                if (nettyConfig.getHttpsAdditionalPipelineInitializer() != null)
+                    nettyConfig.getHttpsAdditionalPipelineInitializer().initPipeline(pipeline);
+
                 return pipeline;
             }
         });
@@ -241,6 +253,10 @@ public ChannelPipeline getPipeline() throws Exception {
                 pipeline.addLast(SSL_HANDLER, new SslInitializer(ChannelManager.this));
                 pipeline.addLast(HTTP_HANDLER, newHttpClientCodec());
                 pipeline.addLast(WS_PROCESSOR, wsProcessor);
+
+                if (nettyConfig.getWssAdditionalPipelineInitializer() != null)
+                    nettyConfig.getWssAdditionalPipelineInitializer().initPipeline(pipeline);
+
                 return pipeline;
             }
         });