Use ChannelCustomizer to set SNI

References #11, #12

Author: acogoluegnes

src/docs/asciidoc/api.adoc

@@ -201,10 +201,6 @@ The server certificate chain and the client private key are the typical
 elements that need to be configured.
 |The JDK trust manager and no client private key.
 
-|`tls#sslParameters`
-|Set the `SSLParameters` for the `SSLEngine`. The provided parameters will be merged into the parameters returned by `SSLEngine#getSSLParameters()`. Can be used to set SNI information with `SSLParameters#setServerNames(List)`.
-|`null`
-
 |`tls#trustEverything`
 |Helper to configure a `SslContext` that trusts all server certificates
 and does not use a client private key. **Only for development**.

src/main/java/com/rabbitmq/stream/ChannelCustomizer.java

@@ -14,6 +14,7 @@
 package com.rabbitmq.stream;
 
 import io.netty.channel.Channel;
+import java.util.Objects;
 
 /**
  * An extension point to customize Netty's {@link io.netty.channel.Channel}s used for connection.
@@ -23,4 +24,12 @@
 public interface ChannelCustomizer {
 
   void customize(Channel channel);
+
+  default ChannelCustomizer andThen(ChannelCustomizer after) {
+    Objects.requireNonNull(after);
+    return ch -> {
+      customize(ch);
+      after.customize(ch);
+    };
+  }
 }

src/main/java/com/rabbitmq/stream/EnvironmentBuilder.java

@@ -25,8 +25,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
-import javax.net.ssl.SSLEngine;
-import javax.net.ssl.SSLParameters;
 
 /**
  * API to configure and create an {@link Environment}.
@@ -334,21 +332,6 @@ interface TlsConfiguration {
      */
     TlsConfiguration sslContext(SslContext sslContext);
 
-    /**
-     * Set {@link SSLParameters} for the {@link javax.net.ssl.SSLEngine}.
-     *
-     * <p>Provided {@link SSLParameters} will be merged into the {@link SSLParameters} returned by
-     * {@link SSLEngine#getSSLParameters()}, that is non-null property values from the provided
-     * instance will override those in the original instance.
-     *
-     * <p>This is typically use to provide SNI information with {@link
-     * SSLParameters#setServerNames(List)}.
-     *
-     * @param sslParameters
-     * @return the TLS configuration helper
-     */
-    TlsConfiguration sslParameters(SSLParameters sslParameters);
-
     /**
      * Convenience method to set a {@link SslContext} that trusts all servers.
      *

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -285,15 +285,13 @@ public void write(
             if (parameters.sslContext != null) {
               SslHandler sslHandler =
                   parameters.sslContext.newHandler(ch.alloc(), parameters.host, parameters.port);
-              SSLEngine sslEngine = sslHandler.engine();
-              SSLParameters sslParameters = sslEngine.getSSLParameters();
+
               if (parameters.tlsHostnameVerification) {
+                SSLEngine sslEngine = sslHandler.engine();
+                SSLParameters sslParameters = sslEngine.getSSLParameters();
                 sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
+                sslEngine.setSSLParameters(sslParameters);
               }
-              if (parameters.sslParameters != null) {
-                Utils.mergeSslParameters(sslParameters, parameters.sslParameters);
-              }
-              sslEngine.setSSLParameters(sslParameters);
 
               ch.pipeline().addFirst("ssl", sslHandler);
             }
@@ -1923,7 +1921,6 @@ public static class ClientParameters {
     private ChunkChecksum chunkChecksum = JdkChunkChecksum.CRC32_SINGLETON;
     private MetricsCollector metricsCollector = NoOpMetricsCollector.SINGLETON;
     private SslContext sslContext;
-    private SSLParameters sslParameters;
     private boolean tlsHostnameVerification = true;
     private ByteBufAllocator byteBufAllocator;
     private Duration rpcTimeout;
@@ -2070,11 +2067,6 @@ public ClientParameters sslContext(SslContext sslContext) {
       return this;
     }
 
-    public ClientParameters sslParameters(SSLParameters sslParameters) {
-      this.sslParameters = sslParameters;
-      return this;
-    }
-
     public ClientParameters tlsHostnameVerification(boolean tlsHostnameVerification) {
       this.tlsHostnameVerification = tlsHostnameVerification;
       return this;

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -145,7 +145,6 @@ class StreamEnvironment implements Environment {
         clientParametersPrototype.sslContext(sslContext);
         clientParametersPrototype.tlsHostnameVerification(
             tlsConfiguration.hostnameVerificationEnabled());
-        clientParametersPrototype.sslParameters(tlsConfiguration.sslParameters());
 
       } catch (SSLException e) {
         throw new StreamException("Error while creating Netty SSL context", e);

src/main/java/com/rabbitmq/stream/impl/StreamEnvironmentBuilder.java

@@ -38,7 +38,6 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Collectors;
 import javax.net.ssl.SSLException;
-import javax.net.ssl.SSLParameters;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -301,7 +300,6 @@ static final class DefaultTlsConfiguration implements TlsConfiguration {
     private boolean enabled = false;
     private boolean hostnameVerification = true;
     private SslContext sslContext;
-    private SSLParameters sslParameters;
 
     private DefaultTlsConfiguration(EnvironmentBuilder environmentBuilder) {
       this.environmentBuilder = environmentBuilder;
@@ -325,12 +323,6 @@ public TlsConfiguration sslContext(SslContext sslContext) {
       return this;
     }
 
-    @Override
-    public TlsConfiguration sslParameters(SSLParameters sslParameters) {
-      this.sslParameters = sslParameters;
-      return this;
-    }
-
     @Override
     public TlsConfiguration trustEverything() {
       LOGGER.warn(
@@ -368,9 +360,5 @@ public boolean hostnameVerificationEnabled() {
     public SslContext sslContext() {
       return sslContext;
     }
-
-    public SSLParameters sslParameters() {
-      return sslParameters;
-    }
   }
 }

src/main/java/com/rabbitmq/stream/impl/Utils.java

@@ -27,7 +27,6 @@
 import java.util.function.LongConsumer;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
-import javax.net.ssl.SSLParameters;
 import javax.net.ssl.X509TrustManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -216,72 +215,6 @@ public X509Certificate[] getAcceptedIssuers() {
     }
   }
 
-  static void mergeSslParameters(SSLParameters original, SSLParameters provided) {
-    if (notEmptyArray(provided.getCipherSuites())) {
-      LOGGER.debug(
-          "Setting SSLParameters cipherSuites from {} to {}",
-          arrayToString(original.getCipherSuites()),
-          arrayToString(provided.getCipherSuites()));
-      original.setCipherSuites(provided.getCipherSuites());
-    }
-    if (notEmptyArray(provided.getProtocols())) {
-      LOGGER.debug(
-          "Setting SSLParameters protocols from {} to {}",
-          arrayToString(original.getProtocols()),
-          arrayToString(provided.getProtocols()));
-      original.setProtocols(provided.getProtocols());
-    }
-    if (original.getWantClientAuth() != provided.getWantClientAuth()) {
-      LOGGER.debug(
-          "Setting SSLParameters wantClientAuth from {} to {}",
-          original.getWantClientAuth(),
-          provided.getWantClientAuth());
-      original.setWantClientAuth(provided.getWantClientAuth());
-    }
-    if (original.getNeedClientAuth() != provided.getNeedClientAuth()) {
-      LOGGER.debug(
-          "Setting SSLParameters needClientAuth from {} to {}",
-          original.getNeedClientAuth(),
-          provided.getNeedClientAuth());
-      original.setNeedClientAuth(provided.getNeedClientAuth());
-    }
-    if (notNullOrBlank(provided.getEndpointIdentificationAlgorithm())) {
-      LOGGER.debug(
-          "Setting SSLParameters endpointIdentificationAlgorithm from {} to {}",
-          original.getEndpointIdentificationAlgorithm(),
-          provided.getEndpointIdentificationAlgorithm());
-      original.setEndpointIdentificationAlgorithm(provided.getEndpointIdentificationAlgorithm());
-    }
-    if (provided.getAlgorithmConstraints() != null) {
-      LOGGER.debug(
-          "Setting SSLParameters algorithmConstraints from {} to {}",
-          original.getAlgorithmConstraints(),
-          provided.getAlgorithmConstraints());
-      original.setAlgorithmConstraints(provided.getAlgorithmConstraints());
-    }
-    if (provided.getServerNames() != null) {
-      LOGGER.debug(
-          "Setting SSLParameters serverNames from {} to {}",
-          original.getServerNames(),
-          provided.getServerNames());
-      original.setServerNames(provided.getServerNames());
-    }
-    if (provided.getSNIMatchers() != null) {
-      LOGGER.debug(
-          "Setting SSLParameters SNIMatchers from {} to {}",
-          original.getSNIMatchers(),
-          provided.getSNIMatchers());
-      original.setSNIMatchers(provided.getSNIMatchers());
-    }
-    if (original.getUseCipherSuitesOrder() != provided.getUseCipherSuitesOrder()) {
-      LOGGER.debug(
-          "Setting SSLParameters useCipherSuitesOrder from {} to {}",
-          original.getUseCipherSuitesOrder(),
-          provided.getUseCipherSuitesOrder());
-      original.setUseCipherSuitesOrder(provided.getUseCipherSuitesOrder());
-    }
-  }
-
   private static boolean notNullOrBlank(String str) {
     return str != null && !str.trim().isEmpty();
   }

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -19,6 +19,7 @@
 import com.rabbitmq.stream.Address;
 import com.rabbitmq.stream.AddressResolver;
 import com.rabbitmq.stream.ByteCapacity;
+import com.rabbitmq.stream.ChannelCustomizer;
 import com.rabbitmq.stream.Codec;
 import com.rabbitmq.stream.ConfirmationHandler;
 import com.rabbitmq.stream.Constants;
@@ -46,6 +47,7 @@
 import io.netty.buffer.ByteBufAllocatorMetric;
 import io.netty.buffer.ByteBufAllocatorMetricProvider;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
 import io.netty.util.internal.PlatformDependent;
 import java.io.PrintStream;
 import java.io.PrintWriter;
@@ -501,22 +503,31 @@ public Integer call() throws Exception {
             .maxTrackingConsumersByConnection(this.trackingConsumersByConnection)
             .maxConsumersByConnection(this.consumersByConnection);
 
+    ChannelCustomizer channelCustomizer = channel -> {};
+
     if (tls) {
       TlsConfiguration tlsConfiguration = environmentBuilder.tls();
       tlsConfiguration =
           tlsConfiguration.sslContext(
               SslContextBuilder.forClient()
                   .trustManager(Utils.TRUST_EVERYTHING_TRUST_MANAGER)
                   .build());
+      environmentBuilder = tlsConfiguration.environmentBuilder();
       if (!this.sniServerNames.isEmpty()) {
-        SSLParameters sslParameters = new SSLParameters();
-        sslParameters.setServerNames(this.sniServerNames);
-        tlsConfiguration = tlsConfiguration.sslParameters(sslParameters);
+        channelCustomizer =
+            channelCustomizer.andThen(
+                ch -> {
+                  SslHandler sslHandler = ch.pipeline().get(SslHandler.class);
+                  if (sslHandler != null) {
+                    SSLParameters sslParameters = sslHandler.engine().getSSLParameters();
+                    sslParameters.setServerNames(this.sniServerNames);
+                    sslHandler.engine().setSSLParameters(sslParameters);
+                  }
+                });
       }
-      environmentBuilder = tlsConfiguration.environmentBuilder();
     }
 
-    Environment environment = environmentBuilder.build();
+    Environment environment = environmentBuilder.channelCustomizer(channelCustomizer).build();
     shutdownService.wrap(closeStep("Closing environment(s)", () -> environment.close()));
 
     streams = Utils.streams(this.streamCount, this.streams);

src/test/java/com/rabbitmq/stream/impl/StreamEnvironmentTest.java

@@ -25,6 +25,7 @@
 import com.rabbitmq.stream.Address;
 import com.rabbitmq.stream.AuthenticationFailureException;
 import com.rabbitmq.stream.BackOffDelayPolicy;
+import com.rabbitmq.stream.ChannelCustomizer;
 import com.rabbitmq.stream.Constants;
 import com.rabbitmq.stream.Consumer;
 import com.rabbitmq.stream.Environment;
@@ -37,6 +38,7 @@
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfTlsNotEnabled;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.handler.ssl.SslHandler;
 import java.net.ConnectException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
@@ -137,13 +139,20 @@ void environmentCreationShouldSucceedWithUrlContainingAllCorrectInformation() {
   @DisabledIfTlsNotEnabled
   @Test
   void environmentCreationShouldSucceedWhenUsingTls() {
-    SSLParameters sslParameters = new SSLParameters();
-    sslParameters.setServerNames(Collections.singletonList(new SNIHostName("localhost")));
+    ChannelCustomizer channelCustomizer =
+        ch -> {
+          SslHandler sslHandler = ch.pipeline().get(SslHandler.class);
+          if (sslHandler != null) {
+            SSLParameters sslParameters = sslHandler.engine().getSSLParameters();
+            sslParameters.setServerNames(Collections.singletonList(new SNIHostName("localhost")));
+            sslHandler.engine().setSSLParameters(sslParameters);
+          }
+        };
     environmentBuilder
         .uri("rabbitmq-stream+tls://guest:guest@localhost:5551/%2f")
+        .channelCustomizer(channelCustomizer)
         .tls()
         .trustEverything()
-        .sslParameters(sslParameters)
         .environmentBuilder()
         .build()
         .close();

src/test/java/com/rabbitmq/stream/impl/TlsTest.java

@@ -20,6 +20,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import com.rabbitmq.stream.ChannelCustomizer;
 import com.rabbitmq.stream.ConfirmationHandler;
 import com.rabbitmq.stream.Environment;
 import com.rabbitmq.stream.OffsetSpecification;
@@ -29,6 +30,7 @@
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfTlsNotEnabled;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
 import java.io.File;
 import java.io.FileInputStream;
 import java.nio.charset.Charset;
@@ -175,10 +177,20 @@ void unverifiedConnection() {
   }
 
   @Test
-  void unverifiedConnectionWithSslParameters() {
-    SSLParameters sslParameters = new SSLParameters();
-    sslParameters.setServerNames(Collections.singletonList(new SNIHostName("localhost")));
-    cf.get(new ClientParameters().sslContext(alwaysTrustSslContext()).sslParameters(sslParameters));
+  void unverifiedConnectionWithSni() {
+    ChannelCustomizer channelCustomizer =
+        ch -> {
+          SslHandler sslHandler = ch.pipeline().get(SslHandler.class);
+          if (sslHandler != null) {
+            SSLParameters sslParameters = sslHandler.engine().getSSLParameters();
+            sslParameters.setServerNames(Collections.singletonList(new SNIHostName("localhost")));
+            sslHandler.engine().setSSLParameters(sslParameters);
+          }
+        };
+    cf.get(
+        new ClientParameters()
+            .sslContext(alwaysTrustSslContext())
+            .channelCustomizer(channelCustomizer));
   }
 
   @Test