Add opt-in to enable server hostname verification

Hostname verification isn't performed by default in the TLS handshake,
this commit makes it easier to enable server hostname verification for
both blocking and non-blocking IO modes.

References #394

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

@@ -15,8 +15,6 @@
 
 package com.rabbitmq.client;
 
-import static java.util.concurrent.TimeUnit.*;
-
 import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.ConnectionParams;
 import com.rabbitmq.client.impl.CredentialsProvider;
@@ -32,6 +30,10 @@
 import com.rabbitmq.client.impl.recovery.RetryHandler;
 import com.rabbitmq.client.impl.recovery.TopologyRecoveryFilter;
 
+import javax.net.SocketFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -50,10 +52,8 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Predicate;
-import javax.net.SocketFactory;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLSocketFactory;
-import javax.net.ssl.TrustManager;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
 
 /**
  * Convenience factory class to facilitate opening a {@link Connection} to a RabbitMQ node.
@@ -132,7 +132,7 @@ public class ConnectionFactory implements Cloneable {
     // connections uses, see rabbitmq/rabbitmq-java-client#86
     private ExecutorService shutdownExecutor;
     private ScheduledExecutorService heartbeatExecutor;
-    private SocketConfigurator socketConf           = new DefaultSocketConfigurator();
+    private SocketConfigurator socketConf           = SocketConfigurators.defaultConfigurator();
     private ExceptionHandler exceptionHandler       = new DefaultExceptionHandler();
     private CredentialsProvider credentialsProvider = new DefaultCredentialsProvider(DEFAULT_USER, DEFAULT_PASS);
 
@@ -729,6 +729,44 @@ public void useSslProtocol(SSLContext context) {
         setSocketFactory(context.getSocketFactory());
     }
 
+    /**
+     * Enable server hostname verification for TLS connections.
+     * <p>
+     * This enables hostname verification regardless of the IO mode
+     * used (blocking or non-blocking IO).
+     * <p>
+     * This can be called typically after setting the {@link SSLContext}
+     * with one of the <code>useSslProtocol</code> methods.
+     *
+     * @see NioParams#enableHostnameVerification()
+     * @see NioParams#setSslEngineConfigurator(SslEngineConfigurator)
+     * @see SslEngineConfigurators#ENABLE_HOSTNAME_VERIFICATION
+     * @see SocketConfigurators#ENABLE_HOSTNAME_VERIFICATION
+     * @see ConnectionFactory#useSslProtocol(String)
+     * @see ConnectionFactory#useSslProtocol(SSLContext)
+     * @see ConnectionFactory#useSslProtocol()
+     * @see ConnectionFactory#useSslProtocol(String, TrustManager)
+     */
+    public void enableHostnameVerification() {
+        enableHostnameVerificationForNio();
+        enableHostnameVerificationForBlockingIo();
+    }
+
+    protected void enableHostnameVerificationForNio() {
+        if (this.nioParams == null) {
+            this.nioParams = new NioParams();
+        }
+        this.nioParams = this.nioParams.enableHostnameVerification();
+    }
+
+    protected void enableHostnameVerificationForBlockingIo() {
+        if (this.socketConf == null) {
+            this.socketConf = SocketConfigurators.builder().defaultConfigurator().enableHostnameVerification().build();
+        } else {
+            this.socketConf = this.socketConf.andThen(SocketConfigurators.enableHostnameVerification());
+        }
+    }
+
     public static String computeDefaultTlsProcotol(String[] supportedProtocols) {
         if(supportedProtocols != null) {
             for (String supportedProtocol : supportedProtocols) {

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

@@ -17,7 +17,9 @@
 
 import java.io.IOException;
 import java.nio.channels.SocketChannel;
+import java.util.Objects;
 
+@FunctionalInterface
 public interface SocketChannelConfigurator {
 
     /**
@@ -26,4 +28,18 @@ public interface SocketChannelConfigurator {
      */
     void configure(SocketChannel socketChannel) throws IOException;
 
+    /**
+     * Returns a composed configurator that performs, in sequence, this
+     * operation followed by the {@code after} operation.
+     *
+     * @param after the operation to perform after this operation
+     * @return a composed configurator that performs in sequence this
+     * operation followed by the {@code after} operation
+     * @throws NullPointerException if {@code after} is null
+     */
+    default SocketChannelConfigurator andThen(SocketChannelConfigurator after) {
+        Objects.requireNonNull(after);
+        return t -> { configure(t); after.configure(t); };
+    }
+
 }

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

@@ -0,0 +1,111 @@
+// Copyright (c) 2018 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;
+
+/**
+ * Ready-to-use instances and builder for {@link SocketChannelConfigurator}.
+ * <p>
+ * Note {@link SocketChannelConfigurator}s can be combined with
+ * {@link SocketChannelConfigurator#andThen(SocketChannelConfigurator)}.
+ *
+ * @since 5.5.0
+ */
+public abstract class SocketChannelConfigurators {
+
+    /**
+     * Disable Nagle's algorithm.
+     */
+    public static final SocketChannelConfigurator DISABLE_NAGLE_ALGORITHM =
+        socketChannel -> SocketConfigurators.DISABLE_NAGLE_ALGORITHM.configure(socketChannel.socket());
+
+    /**
+     * Default {@link SocketChannelConfigurator} that disables Nagle's algorithm.
+     */
+    public static final SocketChannelConfigurator DEFAULT = DISABLE_NAGLE_ALGORITHM;
+
+    /**
+     * The default {@link SocketChannelConfigurator} that disables Nagle's algorithm.
+     *
+     * @return
+     */
+    public static SocketChannelConfigurator defaultConfigurator() {
+        return DEFAULT;
+    }
+
+    /**
+     * {@link SocketChannelConfigurator} that disables Nagle's algorithm.
+     *
+     * @return
+     */
+    public static SocketChannelConfigurator disableNagleAlgorithm() {
+        return DISABLE_NAGLE_ALGORITHM;
+    }
+
+    /**
+     * Builder to configure and creates a {@link SocketChannelConfigurator} instance.
+     *
+     * @return
+     */
+    public static SocketChannelConfigurators.Builder builder() {
+        return new SocketChannelConfigurators.Builder();
+    }
+
+    public static class Builder {
+
+        private SocketChannelConfigurator configurator = channel -> {
+        };
+
+        /**
+         * Set default configuration.
+         *
+         * @return
+         */
+        public Builder defaultConfigurator() {
+            configurator = configurator.andThen(DEFAULT);
+            return this;
+        }
+
+        /**
+         * Disable Nagle's Algorithm.
+         *
+         * @return
+         */
+        public Builder disableNagleAlgorithm() {
+            configurator = configurator.andThen(DISABLE_NAGLE_ALGORITHM);
+            return this;
+        }
+
+        /**
+         * Add an extra configuration step.
+         *
+         * @param extraConfiguration
+         * @return
+         */
+        public Builder add(SocketChannelConfigurator extraConfiguration) {
+            configurator = configurator.andThen(extraConfiguration);
+            return this;
+        }
+
+        /**
+         * Return the configured {@link SocketConfigurator}.
+         *
+         * @return
+         */
+        public SocketChannelConfigurator build() {
+            return configurator;
+        }
+    }
+}

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

@@ -17,11 +17,31 @@
 
 import java.io.IOException;
 import java.net.Socket;
+import java.util.Objects;
 
+@FunctionalInterface
 public interface SocketConfigurator {
+
     /**
      * Provides a hook to insert custom configuration of the sockets
      * used to connect to an AMQP server before they connect.
      */
     void configure(Socket socket) throws IOException;
+
+    /**
+     * Returns a composed configurator that performs, in sequence, this
+     * operation followed by the {@code after} operation.
+     *
+     * @param after the operation to perform after this operation
+     * @return a composed configurator that performs in sequence this
+     * operation followed by the {@code after} operation
+     * @throws NullPointerException if {@code after} is null
+     */
+    default SocketConfigurator andThen(SocketConfigurator after) {
+        Objects.requireNonNull(after);
+        return t -> {
+            configure(t);
+            after.configure(t);
+        };
+    }
 }