Manipulatable pipelines

pom.xml

@@ -301,7 +301,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-shade-plugin</artifactId>
-                <version>1.5</version>
+                <version>1.6</version>
                 <executions>
                     <execution>
                         <phase>package</phase>
@@ -378,6 +378,17 @@
                                 <exclude>build.properties</exclude>
                             </excludes>
                         </filter>
+                        <filter>
+                            <artifact>io.netty:netty</artifact>
+                            <includes>
+                                <include>**</include>
+                                <exclude>META-INF/**</exclude>
+                                <exclude>LICENSE</exclude>
+                                <exclude>NOTICE</exclude>
+                                <exclude>/*.txt</exclude>
+                                <exclude>build.properties</exclude>
+                            </includes>
+                        </filter>
                     </filters>
                 </configuration>
             </plugin>

src/main/java/org/elasticsearch/common/netty/PipelineFactories.java

@@ -0,0 +1,40 @@
+package org.elasticsearch.common.netty;
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.ElasticSearchException;
+import org.elasticsearch.transport.netty.NettyTransport;
+import org.elasticsearch.transport.netty.MessageChannelHandler;
+import org.elasticsearch.common.netty.OpenChannelsHandler;
+import org.elasticsearch.common.inject.Inject;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.ChannelPipeline;
+import org.jboss.netty.channel.Channels;
+
+public class PipelineFactories {
+
+    @Inject()
+    public PipelineFactories() {
+    }
+
+    public ChannelPipelineFactory serverPipelineFactory(final NettyTransport transport, final OpenChannelsHandler serverOpenChannels, final ESLogger logger) throws ElasticSearchException {
+        return new ChannelPipelineFactory() {
+            @Override
+            public ChannelPipeline getPipeline() throws Exception {
+                ChannelPipeline pipeline = Channels.pipeline();
+                pipeline.addLast("openChannels", serverOpenChannels);
+                pipeline.addLast("dispatcher", new MessageChannelHandler(transport, logger));
+                return pipeline;
+            }
+        };
+    }
+
+    public ChannelPipelineFactory clientPipelineFactory(final NettyTransport transport, final ESLogger logger) throws ElasticSearchException {
+        return new ChannelPipelineFactory() {
+            @Override
+            public ChannelPipeline getPipeline() throws Exception {
+                ChannelPipeline pipeline = Channels.pipeline();
+                pipeline.addLast("dispatcher", new MessageChannelHandler(transport, logger));
+                return pipeline;
+            }
+        };
+    }
+}

src/main/java/org/elasticsearch/transport/netty/NettyTransport.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.common.io.stream.Streamable;
 import org.elasticsearch.common.netty.NettyStaticSetup;
 import org.elasticsearch.common.netty.OpenChannelsHandler;
+import org.elasticsearch.common.netty.PipelineFactories;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.network.NetworkUtils;
 import org.elasticsearch.common.settings.Settings;
@@ -119,14 +120,15 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
 
     private volatile OpenChannelsHandler serverOpenChannels;
 
+    private PipelineFactories pipelineFactories;
+
     private volatile ClientBootstrap clientBootstrap;
 
     private volatile ServerBootstrap serverBootstrap;
 
     // node id to actual channel
     final ConcurrentMap<DiscoveryNode, NodeChannels> connectedNodes = newConcurrentMap();
 
-
     private volatile Channel serverChannel;
 
     private volatile TransportServiceAdapter transportServiceAdapter;
@@ -136,18 +138,19 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
     private final Object[] connectMutex;
 
     public NettyTransport(ThreadPool threadPool) {
-        this(EMPTY_SETTINGS, threadPool, new NetworkService(EMPTY_SETTINGS));
+        this(EMPTY_SETTINGS, threadPool, new NetworkService(EMPTY_SETTINGS), new PipelineFactories());
     }
 
     public NettyTransport(Settings settings, ThreadPool threadPool) {
-        this(settings, threadPool, new NetworkService(settings));
+        this(settings, threadPool, new NetworkService(settings), new PipelineFactories());
     }
 
     @Inject
-    public NettyTransport(Settings settings, ThreadPool threadPool, NetworkService networkService) {
+    public NettyTransport(Settings settings, ThreadPool threadPool, NetworkService networkService, PipelineFactories pipelineFactories) {
         super(settings);
         this.threadPool = threadPool;
         this.networkService = networkService;
+        this.pipelineFactories = pipelineFactories;
 
         this.connectMutex = new Object[500];
         for (int i = 0; i < connectMutex.length; i++) {
@@ -202,14 +205,8 @@ protected void doStart() throws ElasticSearchException {
                     Executors.newCachedThreadPool(daemonThreadFactory(settings, "transport_client_worker")),
                     workerCount));
         }
-        ChannelPipelineFactory clientPipelineFactory = new ChannelPipelineFactory() {
-            @Override
-            public ChannelPipeline getPipeline() throws Exception {
-                ChannelPipeline pipeline = Channels.pipeline();
-                pipeline.addLast("dispatcher", new MessageChannelHandler(NettyTransport.this, logger));
-                return pipeline;
-            }
-        };
+        ChannelPipelineFactory clientPipelineFactory = pipelineFactories.clientPipelineFactory(this, logger);
+
         clientBootstrap.setPipelineFactory(clientPipelineFactory);
         clientBootstrap.setOption("connectTimeoutMillis", connectTimeout.millis());
         if (tcpNoDelay != null) {
@@ -244,15 +241,8 @@ public ChannelPipeline getPipeline() throws Exception {
                     Executors.newCachedThreadPool(daemonThreadFactory(settings, "transport_server_worker")),
                     workerCount));
         }
-        ChannelPipelineFactory serverPipelineFactory = new ChannelPipelineFactory() {
-            @Override
-            public ChannelPipeline getPipeline() throws Exception {
-                ChannelPipeline pipeline = Channels.pipeline();
-                pipeline.addLast("openChannels", serverOpenChannels);
-                pipeline.addLast("dispatcher", new MessageChannelHandler(NettyTransport.this, logger));
-                return pipeline;
-            }
-        };
+        ChannelPipelineFactory serverPipelineFactory = pipelineFactories.serverPipelineFactory(this, serverOpenChannels, logger);
+
         serverBootstrap.setPipelineFactory(serverPipelineFactory);
         if (tcpNoDelay != null) {
             serverBootstrap.setOption("child.tcpNoDelay", tcpNoDelay);