Switch to NIO

pom.xml

@@ -410,6 +410,28 @@
       </build>
     </profile>
 
+    <!--
+      Profile to activate the NIO mode in the test suite:
+      mvn verify -P use-nio
+    -->
+    <profile>
+      <id>use-nio</id>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-failsafe-plugin</artifactId>
+            <version>2.19.1</version>
+            <configuration>
+              <systemPropertyVariables>
+                <use.nio>true</use.nio>
+              </systemPropertyVariables>
+            </configuration>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+
     <profile>
       <!--
       The "release" Maven profile is used to push release artifacts to a

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -16,6 +16,8 @@
 package com.rabbitmq.client;
 
 import com.rabbitmq.client.impl.*;
+import com.rabbitmq.client.impl.nio.NioParams;
+import com.rabbitmq.client.impl.nio.SocketChannelFrameHandlerFactory;
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 
 import javax.net.SocketFactory;
@@ -108,6 +110,12 @@ public class ConnectionFactory implements Cloneable {
 
     private MetricsCollector metricsCollector;
 
+    private boolean nio = false;
+    private FrameHandlerFactory frameHandlerFactory;
+    private NioParams nioParams = new NioParams();
+
+    private SSLContext sslContext;
+
     /** @return the default host to use for connections */
     public String getHost() {
         return host;
@@ -541,7 +549,7 @@ public boolean isSSL(){
     }
 
     /**
-     * Convenience method for setting up a SSL socket factory, using
+     * Convenience method for setting up a SSL socket factory/engine, using
      * the DEFAULT_SSL_PROTOCOL and a trusting TrustManager.
      */
     public void useSslProtocol()
@@ -551,7 +559,7 @@ public void useSslProtocol()
     }
 
     /**
-     * Convenience method for setting up a SSL socket factory, using
+     * Convenience method for setting up a SSL socket factory/engine, using
      * the supplied protocol and a very trusting TrustManager.
      */
     public void useSslProtocol(String protocol)
@@ -561,7 +569,7 @@ public void useSslProtocol(String protocol)
     }
 
     /**
-     * Convenience method for setting up an SSL socket factory.
+     * Convenience method for setting up an SSL socket factory/engine.
      * Pass in the SSL protocol to use, e.g. "TLSv1" or "TLSv1.2".
      *
      * @param protocol SSL protocol to use.
@@ -575,13 +583,14 @@ public void useSslProtocol(String protocol, TrustManager trustManager)
     }
 
     /**
-     * Convenience method for setting up an SSL socket factory.
+     * Convenience method for setting up an SSL socket factory/engine.
      * Pass in an initialized SSLContext.
      *
      * @param context An initialized SSLContext
      */
     public void useSslProtocol(SSLContext context) {
         setSocketFactory(context.getSocketFactory());
+        this.sslContext = context;
     }
 
     public static String computeDefaultTlsProcotol(String[] supportedProtocols) {
@@ -641,8 +650,19 @@ public MetricsCollector getMetricsCollector() {
         return metricsCollector;
     }
 
-    protected FrameHandlerFactory createFrameHandlerFactory() throws IOException {
-        return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL(), this.shutdownExecutor);
+    protected synchronized FrameHandlerFactory createFrameHandlerFactory() throws IOException {
+        if(nio) {
+            if(this.frameHandlerFactory == null) {
+                if(this.nioParams.getNioExecutor() == null && this.nioParams.getThreadFactory() == null) {
+                    this.nioParams.setThreadFactory(getThreadFactory());
+                }
+                this.frameHandlerFactory = new SocketChannelFrameHandlerFactory(connectionTimeout, nioParams, isSSL(), sslContext);
+            }
+            return this.frameHandlerFactory;
+        } else {
+            return new SocketFrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL(), this.shutdownExecutor);
+        }
+
     }
 
     /**
@@ -1019,4 +1039,44 @@ public void setNetworkRecoveryInterval(int networkRecoveryInterval) {
     public void setNetworkRecoveryInterval(long networkRecoveryInterval) {
         this.networkRecoveryInterval = networkRecoveryInterval;
     }
+
+    /**
+     * Sets the parameters when using NIO.
+     *
+     *
+     * @param nioParams
+     * @see NioParams
+     */
+    public void setNioParams(NioParams nioParams) {
+        this.nioParams = nioParams;
+    }
+
+    /**
+     * Use non-blocking IO (NIO) for communication with the server.
+     * With NIO, several connections created from the same {@link ConnectionFactory}
+     * can use the same IO thread.
+     *
+     * A client process using a lot of not-so-active connections can benefit
+     * from NIO, as it would use fewer threads than with the traditional, blocking IO mode.
+     *
+     * Use {@link NioParams} to tune NIO and a {@link SocketChannelConfigurator} to
+     * configure the underlying {@link java.nio.channels.SocketChannel}s for connections.
+     *
+     * @see NioParams
+     * @see SocketChannelConfigurator
+     * @see java.nio.channels.SocketChannel
+     * @see java.nio.channels.Selector
+     */
+    public void useNio() {
+        this.nio = true;
+    }
+
+    /**
+     * Use blocking IO for communication with the server.
+     * With blocking IO, each connection creates its own thread
+     * to read data from the server.
+     */
+    public void useBlockingIo() {
+        this.nio = false;
+    }
 }

src/main/java/com/rabbitmq/client/DefaultSocketChannelConfigurator.java

@@ -0,0 +1,38 @@
+// Copyright (c) 2007-Present Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+
+package com.rabbitmq.client;
+
+import java.io.IOException;
+import java.nio.channels.SocketChannel;
+
+public class DefaultSocketChannelConfigurator implements SocketChannelConfigurator {
+
+    /**
+     *  Provides a hook to insert custom configuration of the {@link SocketChannel}s
+     *  used to connect to an AMQP server before they connect.
+     *
+     *  The default behaviour of this method is to disable Nagle's
+     *  algorithm to get more consistently low latency.  However it
+     *  may be overridden freely and there is no requirement to retain
+     *  this behaviour.
+     *
+     *  @param socketChannel The socket channel that is to be used for the Connection
+     */
+    @Override
+    public void configure(SocketChannel socketChannel) throws IOException {
+        socketChannel.socket().setTcpNoDelay(true);
+    }
+}

src/main/java/com/rabbitmq/client/SocketChannelConfigurator.java

@@ -0,0 +1,29 @@
+// Copyright (c) 2007-Present Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+
+package com.rabbitmq.client;
+
+import java.io.IOException;
+import java.nio.channels.SocketChannel;
+
+public interface SocketChannelConfigurator {
+
+    /**
+     * Provides a hook to insert custom configuration of the {@link SocketChannel}s
+     * used to connect to an AMQP server before they connect.
+     */
+    void configure(SocketChannel socketChannel) throws IOException;
+
+}

src/main/java/com/rabbitmq/client/impl/AMQChannel.java

@@ -143,8 +143,12 @@ public void handleCompleteInboundCommand(AMQCommand command) throws IOException
         if (!processAsync(command)) {
             // The filter decided not to handle/consume the command,
             // so it must be some reply to an earlier RPC.
-            nextOutstandingRpc().handleCommand(command);
-            markRpcFinished();
+            RpcContinuation nextOutstandingRpc = nextOutstandingRpc();
+            // the outstanding RPC can be null when calling Channel#asyncRpc
+            if(nextOutstandingRpc != null) {
+                nextOutstandingRpc.handleCommand(command);
+                markRpcFinished();
+            }
         }
     }