Added Option to set reactor.netty.resources.ConnectionProvider

Author: paul-kraftlauget

README.md

@@ -88,6 +88,7 @@ Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
 | `trustStoreType`  | Type of the TrustStore. Defaults to `KeyStore.getDefaultType()`. _(Optional)_
 | `trustStore`      | Path to the certificate TrustStore file. _(Optional)_
 | `trustStorePassword` | Password used to check the integrity of the TrustStore data. _(Optional)_
+| `connectionProvider` | Set the `reactor.netty.resources.ConnectionProvider` to be used when creating the connection. Defaults to `ConnectionProvider.newConnection()`. _(Optional)_                                                                                                              |
 
 
 **Programmatic Configuration**

src/main/java/io/r2dbc/mssql/MssqlConnectionConfiguration.java

@@ -119,13 +119,16 @@ public final class MssqlConnectionConfiguration {
 
     private final String username;
 
+    @Nullable
+    private final ConnectionProvider connectionProvider;
+
     private MssqlConnectionConfiguration(@Nullable String applicationName, @Nullable UUID connectionId, Duration connectTimeout, @Nullable String database, String host, String hostNameInCertificate,
                                          @Nullable Duration lockWaitTimeout, CharSequence password, Predicate<String> preferCursoredExecution, int port, boolean sendStringParametersAsUnicode,
                                          boolean ssl,
                                          Function<SslContextBuilder, SslContextBuilder> sslContextBuilderCustomizer,
                                          @Nullable Function<SslContextBuilder, SslContextBuilder> sslTunnelSslContextBuilderCustomizer, boolean tcpKeepAlive, boolean tcpNoDelay,
                                          boolean trustServerCertificate, @Nullable File trustStore, @Nullable String trustStoreType,
-                                         @Nullable char[] trustStorePassword, String username) {
+                                         @Nullable char[] trustStorePassword, String username, @Nullable ConnectionProvider connectionProvider) {
 
         this.applicationName = applicationName;
         this.connectionId = connectionId;
@@ -148,6 +151,7 @@ private MssqlConnectionConfiguration(@Nullable String applicationName, @Nullable
         this.trustStoreType = trustStoreType;
         this.trustStorePassword = trustStorePassword;
         this.username = Assert.requireNonNull(username, "username must not be null");
+        this.connectionProvider = connectionProvider;
     }
 
     /**
@@ -185,12 +189,14 @@ MssqlConnectionConfiguration withRedirect(Redirect redirect) {
         return new MssqlConnectionConfiguration(this.applicationName, this.connectionId, this.connectTimeout, this.database, redirectServerName, hostNameInCertificate, this.lockWaitTimeout,
                 this.password,
                 this.preferCursoredExecution, redirect.getPort(), this.sendStringParametersAsUnicode, this.ssl, this.sslContextBuilderCustomizer,
-                this.sslTunnelSslContextBuilderCustomizer, this.tcpKeepAlive, this.tcpNoDelay, this.trustServerCertificate, this.trustStore, this.trustStoreType, this.trustStorePassword, this.username);
+                this.sslTunnelSslContextBuilderCustomizer, this.tcpKeepAlive, this.tcpNoDelay, this.trustServerCertificate, this.trustStore, this.trustStoreType, this.trustStorePassword, this.username,
+                this.connectionProvider);
     }
 
     ClientConfiguration toClientConfiguration() {
         return new DefaultClientConfiguration(this.connectTimeout, this.host, this.hostNameInCertificate, this.port, this.ssl, this.sslContextBuilderCustomizer,
-                this.sslTunnelSslContextBuilderCustomizer, this.tcpKeepAlive, this.tcpNoDelay, this.trustServerCertificate, this.trustStore, this.trustStoreType, this.trustStorePassword);
+                this.sslTunnelSslContextBuilderCustomizer, this.tcpKeepAlive, this.tcpNoDelay, this.trustServerCertificate, this.trustStore, this.trustStoreType, this.trustStorePassword,
+                this.connectionProvider);
     }
 
     ConnectionOptions toConnectionOptions() {
@@ -387,6 +393,9 @@ public static final class Builder {
         @Nullable
         private char[] trustStorePassword;
 
+        @Nullable
+        private ConnectionProvider connectionProvider;
+
         private Builder() {
         }
 
@@ -703,6 +712,18 @@ public Builder username(String username) {
             return this;
         }
 
+        /**
+         * Configure the {@link ConnectionProvider} to be used with Netty
+         *
+         * @param connectionProvider the connection provider
+         * @return this {@link Builder}
+         * @since 1.1.0
+         */
+        public Builder connectionProvider(ConnectionProvider connectionProvider) {
+            this.connectionProvider = connectionProvider;
+            return this;
+        }
+
         /**
          * Returns a configured {@link MssqlConnectionConfiguration}.
          *
@@ -719,7 +740,7 @@ public MssqlConnectionConfiguration build() {
                     this.preferCursoredExecution, this.port, this.sendStringParametersAsUnicode, this.ssl, this.sslContextBuilderCustomizer,
                     this.sslTunnelSslContextBuilderCustomizer, this.tcpKeepAlive,
                     this.tcpNoDelay, this.trustServerCertificate, this.trustStore,
-                    this.trustStoreType, this.trustStorePassword, this.username);
+                    this.trustStoreType, this.trustStorePassword, this.username, this.connectionProvider);
         }
 
     }
@@ -756,10 +777,14 @@ static class DefaultClientConfiguration implements ClientConfiguration {
         @Nullable
         private final char[] trustStorePassword;
 
+        @Nullable
+        private final ConnectionProvider connectionProvider;
+
         DefaultClientConfiguration(Duration connectTimeout, String host, String hostNameInCertificate, int port, boolean ssl,
                                    Function<SslContextBuilder, SslContextBuilder> sslContextBuilderCustomizer,
                                    @Nullable Function<SslContextBuilder, SslContextBuilder> sslTunnelSslContextBuilderCustomizer, boolean tcpKeepAlive, boolean tcpNoDelay,
-                                   boolean trustServerCertificate, @Nullable File trustStore, @Nullable String trustStoreType, @Nullable char[] trustStorePassword) {
+                                   boolean trustServerCertificate, @Nullable File trustStore, @Nullable String trustStoreType, @Nullable char[] trustStorePassword,
+                                   ConnectionProvider connectionProvider) {
 
             this.connectTimeout = connectTimeout;
             this.host = host;
@@ -774,6 +799,7 @@ static class DefaultClientConfiguration implements ClientConfiguration {
             this.trustStore = trustStore;
             this.trustStoreType = trustStoreType;
             this.trustStorePassword = trustStorePassword;
+            this.connectionProvider = connectionProvider;
         }
 
         @Override
@@ -803,7 +829,8 @@ public boolean isTcpNoDelay() {
 
         @Override
         public ConnectionProvider getConnectionProvider() {
-            return ConnectionProvider.newConnection();
+            return Optional.ofNullable(connectionProvider)
+                    .orElseGet(ConnectionProvider::newConnection);
         }
 
         @Override

src/main/java/io/r2dbc/mssql/MssqlConnectionFactoryProvider.java

@@ -21,6 +21,7 @@
 import io.r2dbc.spi.ConnectionFactoryOptions;
 import io.r2dbc.spi.ConnectionFactoryProvider;
 import io.r2dbc.spi.Option;
+import reactor.netty.resources.ConnectionProvider;
 import reactor.util.Logger;
 import reactor.util.Loggers;
 
@@ -134,6 +135,13 @@ public final class MssqlConnectionFactoryProvider implements ConnectionFactoryPr
      */
     public static final Option<char[]> TRUST_STORE_PASSWORD = Option.valueOf("trustStorePassword");
 
+    /**
+     * Optional {@link reactor.netty.resources.ConnectionProvider} to control Netty configuration directly
+     *
+     * @since 1.1.0
+     */
+    public static final Option<ConnectionProvider> CONNECTION_PROVIDER = Option.valueOf("connectionProvider");
+
     /**
      * Driver option value.
      */
@@ -194,6 +202,7 @@ public MssqlConnectionFactory create(ConnectionFactoryOptions connectionFactoryO
         mapper.from(TRUST_STORE).map(OptionMapper::toFile).to(builder::trustStore);
         mapper.fromTyped(TRUST_STORE_TYPE).to(builder::trustStoreType);
         mapper.from(TRUST_STORE_PASSWORD).map(it -> it instanceof String ? ((String) it).toCharArray() : (char[]) it).to(builder::trustStorePassword);
+        mapper.fromTyped(CONNECTION_PROVIDER).to(builder::connectionProvider);
 
         builder.host(connectionFactoryOptions.getRequiredValue(HOST).toString());
         builder.password((CharSequence) connectionFactoryOptions.getRequiredValue(PASSWORD));

src/test/java/io/r2dbc/mssql/MssqlConnectionConfigurationUnitTests.java

@@ -28,6 +28,7 @@
 import org.testcontainers.shaded.org.bouncycastle.cert.jcajce.JcaX509CertificateConverter;
 import org.testcontainers.shaded.org.bouncycastle.operator.ContentSigner;
 import org.testcontainers.shaded.org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;
+import reactor.netty.resources.ConnectionProvider;
 
 import java.io.File;
 import java.io.FileOutputStream;
@@ -86,6 +87,7 @@ void builderNoUsername() {
     void configuration() {
         UUID connectionId = UUID.randomUUID();
         Predicate<String> TRUE = s -> true;
+        ConnectionProvider connectionProvider = ConnectionProvider.create("test");
         MssqlConnectionConfiguration configuration = MssqlConnectionConfiguration.builder()
                 .connectionId(connectionId)
                 .database("test-database")
@@ -95,6 +97,7 @@ void configuration() {
                 .port(100)
                 .username("test-username")
                 .sendStringParametersAsUnicode(false)
+                .connectionProvider(connectionProvider)
                 .build();
 
         assertThat(configuration)
@@ -105,7 +108,8 @@ void configuration() {
                 .hasFieldOrPropertyWithValue("preferCursoredExecution", TRUE)
                 .hasFieldOrPropertyWithValue("port", 100)
                 .hasFieldOrPropertyWithValue("username", "test-username")
-                .hasFieldOrPropertyWithValue("sendStringParametersAsUnicode", false);
+                .hasFieldOrPropertyWithValue("sendStringParametersAsUnicode", false)
+                .hasFieldOrPropertyWithValue("connectionProvider", connectionProvider);
     }
 
     @Test