Introduce SSLEngine configuration hook for NIO

Can be useful to pass in SSLParameters to the SSLEngine (e.g. for SNI support).
This is specific to NIO, as SSLParameters can be passed in to the SSLSocket
with the SocketConfigurator in blocking IO mode.

Fixes #274

Author: acogoluegnes

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

@@ -0,0 +1,30 @@
+// Copyright (c) 2017-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
+// info@rabbitmq.com.
+
+package com.rabbitmq.client;
+
+import javax.net.ssl.SSLEngine;
+import java.io.IOException;
+
+public interface SslEngineConfigurator {
+
+    /**
+     * Provides a hook to insert custom configuration of the {@link SSLEngine}s
+     * used to connect to an AMQP server before they connect.
+     * Note this is used only when NIO are in use.
+     */
+    void configure(SSLEngine sslEngine) throws IOException;
+
+}

src/main/java/com/rabbitmq/client/impl/nio/NioParams.java

@@ -17,7 +17,10 @@
 
 import com.rabbitmq.client.DefaultSocketChannelConfigurator;
 import com.rabbitmq.client.SocketChannelConfigurator;
+import com.rabbitmq.client.SslEngineConfigurator;
 
+import javax.net.ssl.SSLEngine;
+import java.io.IOException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadFactory;
 
@@ -51,6 +54,12 @@ public class NioParams {
     /** the hook to configure the socket channel before it's open */
     private SocketChannelConfigurator socketChannelConfigurator = new DefaultSocketChannelConfigurator();
 
+    /** the hook to configure the SSL engine before the connection is open */
+    private SslEngineConfigurator sslEngineConfigurator = new SslEngineConfigurator() {
+        @Override
+        public void configure(SSLEngine sslEngine) throws IOException { }
+    };
+
     public NioParams() {
     }
 
@@ -62,6 +71,7 @@ public NioParams(NioParams nioParams) {
         setWriteQueueCapacity(nioParams.getWriteQueueCapacity());
         setNioExecutor(nioParams.getNioExecutor());
         setThreadFactory(nioParams.getThreadFactory());
+        setSslEngineConfigurator(nioParams.getSslEngineConfigurator());
     }
 
     public int getReadByteBufferSize() {
@@ -248,4 +258,21 @@ public void setSocketChannelConfigurator(SocketChannelConfigurator configurator)
     public SocketChannelConfigurator getSocketChannelConfigurator() {
         return socketChannelConfigurator;
     }
+
+    /**
+     * Set the {@link SSLEngine} configurator.
+     * This gets a change to "configure" the SSL engine
+     * before the connection has been opened. This can be
+     * used e.g. to set {@link javax.net.ssl.SSLParameters}.
+     * The default implementation doesn't do anything.
+     *
+     * @param configurator the configurator to use
+     */
+    public void setSslEngineConfigurator(SslEngineConfigurator configurator) {
+        this.sslEngineConfigurator = configurator;
+    }
+
+    public SslEngineConfigurator getSslEngineConfigurator() {
+        return sslEngineConfigurator;
+    }
 }

src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerFactory.java

@@ -70,6 +70,9 @@ public FrameHandler create(Address addr) throws IOException {
             if (ssl) {
                 sslEngine = sslContext.createSSLEngine(addr.getHost(), portNumber);
                 sslEngine.setUseClientMode(true);
+                if (nioParams.getSslEngineConfigurator() != null) {
+                    nioParams.getSslEngineConfigurator().configure(sslEngine);
+                }
             }
 
             SocketAddress address = new InetSocketAddress(addr.getHost(), portNumber);

src/test/java/com/rabbitmq/client/test/ssl/NioTlsUnverifiedConnection.java

@@ -16,15 +16,17 @@
 package com.rabbitmq.client.test.ssl;
 
 import com.rabbitmq.client.*;
+import com.rabbitmq.client.impl.nio.NioParams;
 import com.rabbitmq.client.test.BrokerTestCase;
-import org.junit.Assert;
 import org.junit.Test;
 import org.slf4j.LoggerFactory;
 
+import javax.net.ssl.SSLEngine;
 import java.io.IOException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -66,6 +68,32 @@ public void connectionGetConsume() throws Exception {
         assertTrue("Message has not been received", messagesReceived);
     }
 
+    @Test public void socketChannelConfigurator() throws Exception {
+        ConnectionFactory connectionFactory = new ConnectionFactory();
+        connectionFactory.useNio();
+        connectionFactory.useSslProtocol();
+        NioParams nioParams = new NioParams();
+        final AtomicBoolean sslEngineHasBeenCalled = new AtomicBoolean(false);
+        nioParams.setSslEngineConfigurator(new SslEngineConfigurator() {
+            @Override
+            public void configure(SSLEngine sslEngine) throws IOException {
+                sslEngineHasBeenCalled.set(true);
+            }
+        });
+
+        connectionFactory.setNioParams(nioParams);
+
+        Connection connection = null;
+        try {
+            connection = connectionFactory.newConnection();
+            assertTrue("The SSL engine configurator should have called", sslEngineHasBeenCalled.get());
+        } finally {
+            if (connection != null) {
+                connection.close();
+            }
+        }
+    }
+
     private Connection basicGetBasicConsume(Connection connection, String queue, final CountDownLatch latch)
         throws IOException, TimeoutException {
         Channel channel = connection.createChannel();