Add Option<LoopResources> tcpLoopResources

Add that to configure the `LoopResources` for tcp sockets.
Unix domain socket still use `SocketLoopResources`.

[resolves #319]

Author: harry-hao

README.md

@@ -67,13 +67,13 @@ Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
 
 | Option            | Description
 | ----------------- | -----------
-| `ssl`             | Enables SSL usage (`SSLMode.VERIFY_FULL`)
+| `ssl`             | Enables SSL usage (`SSLMode.VERIFY_FULL`).
 | `driver`          | Must be `postgresql`.
-| `host`            | Server hostname to connect to
+| `host`            | Server hostname to connect to.
 | `port`            | Server port to connect to.  Defaults to `5432`. _(Optional)_
 | `socket`          | Unix Domain Socket path to connect to as alternative to TCP. _(Optional)_
-| `username`        | Login username
-| `password`        | Login password _(Optional when using TLS Certificate authentication)_
+| `username`        | Login username.
+| `password`        | Login password. _(Optional when using TLS Certificate authentication)_
 | `database`        | Database to select. _(Optional)_
 | `applicationName` | The name of the application connecting to the database.  Defaults to `r2dbc-postgresql`. _(Optional)_
 | `autodetectExtensions` | Whether to auto-detect and register `Extension`s from the class path.  Defaults to `true`. _(Optional)_
@@ -89,7 +89,8 @@ Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
 | `sslPassword`     | Key password to decrypt SSL key. _(Optional)_
 | `sslHostnameVerifier` | `javax.net.ssl.HostnameVerifier` implementation. _(Optional)_
 | `tcpNoDelay`      | Enabled/disable TCP NoDelay. Disabled by default. _(Optional)_
-| `tcpKeepAlive`    | Enabled/disable TCP KeepAlive. Disabled by default _(Optional)_
+| `tcpKeepAlive`    | Enabled/disable TCP KeepAlive. Disabled by default. _(Optional)_
+| `tcpLoopResources`| TCP LoopResources. _(Optional)_
 
 **Programmatic Configuration**
 

src/main/java/io/r2dbc/postgresql/PostgresqlConnectionConfiguration.java

@@ -25,6 +25,7 @@
 import io.r2dbc.postgresql.extension.CodecRegistrar;
 import io.r2dbc.postgresql.extension.Extension;
 import io.r2dbc.postgresql.util.Assert;
+import reactor.netty.resources.LoopResources;
 import reactor.netty.tcp.SslProvider;
 import reactor.util.annotation.Nullable;
 
@@ -89,10 +90,12 @@ public final class PostgresqlConnectionConfiguration {
 
     private final boolean tcpNoDelay;
 
+    private final LoopResources tcpLoopResources;
+
     private PostgresqlConnectionConfiguration(String applicationName, boolean autodetectExtensions, @Nullable Duration connectTimeout, @Nullable String database, List<Extension> extensions,
                                               ToIntFunction<String> fetchSize, boolean forceBinary, @Nullable String host, @Nullable Map<String, String> options, @Nullable CharSequence password,
                                               int port, int preparedStatementCacheQueries, @Nullable String schema, @Nullable String socket, boolean tcpKeepAlive, boolean tcpNoDelay,
-                                              String username, SSLConfig sslConfig) {
+                                              String username, SSLConfig sslConfig, LoopResources tcpLoopResources) {
         this.applicationName = Assert.requireNonNull(applicationName, "applicationName must not be null");
         this.autodetectExtensions = autodetectExtensions;
         this.connectTimeout = connectTimeout;
@@ -115,6 +118,7 @@ private PostgresqlConnectionConfiguration(String applicationName, boolean autode
         this.sslConfig = sslConfig;
         this.tcpKeepAlive = tcpKeepAlive;
         this.tcpNoDelay = tcpNoDelay;
+        this.tcpLoopResources = tcpLoopResources;
     }
 
     /**
@@ -244,10 +248,15 @@ boolean isTcpNoDelay() {
     boolean isUseSocket() {
         return getSocket() != null;
     }
+
     SSLConfig getSslConfig() {
         return this.sslConfig;
     }
 
+    LoopResources getTcpLoopResources() {
+        return this.tcpLoopResources;
+    }
+
     private static String obfuscate(int length) {
 
         StringBuilder builder = new StringBuilder();
@@ -318,9 +327,12 @@ public static final class Builder {
 
         private Function<SslContextBuilder, SslContextBuilder> sslContextBuilderCustomizer = Function.identity();
 
-        private boolean tcpKeepAlive;
+        private boolean tcpKeepAlive = false;
 
-        private boolean tcpNoDelay;
+        private boolean tcpNoDelay = false;
+
+        @Nullable
+        private LoopResources tcpLoopResources = null;
 
         @Nullable
         private String username;
@@ -373,7 +385,7 @@ public PostgresqlConnectionConfiguration build() {
             return new PostgresqlConnectionConfiguration(this.applicationName, this.autodetectExtensions, this.connectTimeout, this.database, this.extensions, this.fetchSize, this.forceBinary,
                 this.host, this.options, this.password, this.port, this.preparedStatementCacheQueries, this.schema, this.socket, this.tcpKeepAlive, this.tcpNoDelay, this.username,
                 this.createSslConfig()
-            );
+            , this.tcpLoopResources);
         }
 
         /**
@@ -680,6 +692,18 @@ public Builder username(String username) {
             return this;
         }
 
+        /**
+         * Configure TCP {@link LoopResources}.
+         *
+         * @param loopResources the {@link LoopResources}
+         * @return this {@link Builder}
+         * @since 1.0.0
+         */
+        public Builder tcpLoopResources(LoopResources loopResources) {
+            this.tcpLoopResources = Assert.requireNonNull(loopResources, "tcpLoopResources must not be null");
+            return this;
+        }
+
         @Override
         public String toString() {
             return "Builder{" +

src/main/java/io/r2dbc/postgresql/PostgresqlConnectionFactoryProvider.java

@@ -23,6 +23,7 @@
 import io.r2dbc.spi.ConnectionFactoryOptions;
 import io.r2dbc.spi.ConnectionFactoryProvider;
 import io.r2dbc.spi.Option;
+import reactor.netty.resources.LoopResources;
 
 import javax.net.ssl.HostnameVerifier;
 import java.util.LinkedHashMap;
@@ -139,6 +140,13 @@ public final class PostgresqlConnectionFactoryProvider implements ConnectionFact
      */
     public static final Option<Boolean> TCP_NODELAY = Option.valueOf("tcpNoDelay");
 
+    /**
+     * TCP {@link LoopResources}.
+     *
+     * @since 1.0.0
+     */
+    public static final Option<LoopResources> TCP_LOOP_RESOURCES = Option.valueOf("tcpLoopResources");
+
     /**
      * Determine the number of queries that are cached in each connection.
      * The default is {@code -1}, meaning there's no limit. The value of {@code 0} disables the cache. Any other value specifies the cache size.
@@ -211,6 +219,7 @@ private static PostgresqlConnectionConfiguration.Builder fromConnectionFactoryOp
         });
         mapper.from(TCP_KEEPALIVE).map(OptionMapper::toBoolean).to(builder::tcpKeepAlive);
         mapper.from(TCP_NODELAY).map(OptionMapper::toBoolean).to(builder::tcpNoDelay);
+        mapper.from(TCP_LOOP_RESOURCES).to(builder::tcpLoopResources);
         builder.username(options.getRequiredValue(USER));
 
         return builder;

src/main/java/io/r2dbc/postgresql/client/ConnectionSettings.java

@@ -16,6 +16,7 @@
 
 package io.r2dbc.postgresql.client;
 
+import reactor.netty.resources.LoopResources;
 import reactor.util.annotation.Nullable;
 
 import java.time.Duration;
@@ -34,10 +35,14 @@ public final class ConnectionSettings {
 
     private final boolean tcpNoDelay;
 
-    public ConnectionSettings(@Nullable Duration connectTimeout, boolean tcpKeepAlive, boolean tcpNoDelay) {
+    @Nullable
+    private final LoopResources tcpLoopResources;
+
+    public ConnectionSettings(@Nullable Duration connectTimeout, boolean tcpKeepAlive, boolean tcpNoDelay, @Nullable LoopResources tcpLoopResources) {
         this.tcpKeepAlive = tcpKeepAlive;
         this.tcpNoDelay = tcpNoDelay;
         this.connectTimeout = connectTimeout;
+        this.tcpLoopResources = tcpLoopResources;
     }
 
     @Nullable
@@ -53,4 +58,12 @@ boolean isTcpNoDelay() {
         return this.tcpNoDelay;
     }
 
+    public boolean hasTcpLoopResources() {
+        return this.tcpLoopResources != null;
+    }
+    
+    LoopResources getTcpLoopResources() {
+        return this.tcpLoopResources;
+    }
+
 }

src/main/java/io/r2dbc/postgresql/client/ReactorNettyClient.java

@@ -334,9 +334,30 @@ public static Mono<ReactorNettyClient> connect(String host, int port) {
      * @param connectTimeout connect timeout
      * @param sslConfig      SSL configuration
      * @throws IllegalArgumentException if {@code host} is {@code null}
+     * @throws IllegalArgumentException if {@code sslConfig} is {@code null}
      */
     public static Mono<ReactorNettyClient> connect(String host, int port, @Nullable Duration connectTimeout, SSLConfig sslConfig) {
-        return connect(ConnectionProvider.newConnection(), InetSocketAddress.createUnresolved(host, port), new ConnectionSettings(connectTimeout, false, false), sslConfig);
+        Assert.requireNonNull(host, "host must not be null");
+        Assert.requireNonNull(sslConfig, "sslConfig must not be null");
+
+        return connect(host, port, connectTimeout, sslConfig, null);
+    }
+
+    /**
+     * Create a new frame processor connected to a given host.
+     *
+     * @param host             the host to connect to
+     * @param port             the port to connect to
+     * @param connectTimeout   connect timeout
+     * @param sslConfig        SSL configuration
+     * @param tcpLoopResources tcp loop resources
+     * @throws IllegalArgumentException if {@code host} is {@code null}
+     */
+    public static Mono<ReactorNettyClient> connect(String host, int port, @Nullable Duration connectTimeout, @Nullable SSLConfig sslConfig, @Nullable LoopResources tcpLoopResources) {
+        Assert.requireNonNull(host, "host must not be null");
+
+        ConnectionSettings settings = new ConnectionSettings(connectTimeout, false, false, tcpLoopResources);
+        return connect(ConnectionProvider.newConnection(), InetSocketAddress.createUnresolved(host, port), settings, sslConfig);
     }
 
     /**
@@ -358,6 +379,9 @@ public static Mono<ReactorNettyClient> connect(ConnectionProvider connectionProv
         if (!(socketAddress instanceof InetSocketAddress)) {
             tcpClient = tcpClient.runOn(new SocketLoopResources(), true);
         } else {
+            if (connectionSettings.hasTcpLoopResources()) {
+                tcpClient = tcpClient.runOn(connectionSettings.getTcpLoopResources());
+            }
             tcpClient = tcpClient.option(ChannelOption.SO_KEEPALIVE, connectionSettings.isTcpKeepAlive());
             tcpClient = tcpClient.option(ChannelOption.TCP_NODELAY, connectionSettings.isTcpNoDelay());
         }

src/test/java/io/r2dbc/postgresql/PostgresqlConnectionConfigurationUnitTests.java

@@ -18,13 +18,15 @@
 
 import io.r2dbc.postgresql.client.SSLMode;
 import org.junit.jupiter.api.Test;
+import reactor.netty.resources.LoopResources;
 
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.mockito.Mockito.mock;
 
 /**
  * Unit tests for {@link PostgresqlConnectionConfiguration}.
@@ -61,11 +63,18 @@ void builderNegativeFetchSize() {
             .withMessage("fetch size must be greater or equal zero");
     }
 
+    @Test
+    void builderNoTcpLoopResources() {
+        assertThatIllegalArgumentException().isThrownBy(() -> PostgresqlConnectionConfiguration.builder().tcpLoopResources(null))
+            .withMessage("tcpLoopResources must not be null");
+    }
+
     @Test
     void configuration() {
         Map<String, String> options = new HashMap<>();
         options.put("lock_timeout", "10s");
         options.put("statement_timeout", "60000"); // [ms]
+        LoopResources loopResources = mock(LoopResources.class);
 
         PostgresqlConnectionConfiguration configuration = PostgresqlConnectionConfiguration.builder()
             .applicationName("test-application-name")
@@ -77,6 +86,9 @@ void configuration() {
             .schema("test-schema")
             .username("test-username")
             .sslMode(SSLMode.ALLOW)
+            .tcpKeepAlive(true)
+            .tcpNoDelay(true)
+            .tcpLoopResources(loopResources)
             .build();
 
         assertThat(configuration)
@@ -88,7 +100,10 @@ void configuration() {
             .hasFieldOrPropertyWithValue("password", null)
             .hasFieldOrPropertyWithValue("port", 100)
             .hasFieldOrPropertyWithValue("username", "test-username")
-            .hasFieldOrProperty("sslConfig");
+            .hasFieldOrProperty("sslConfig")
+            .hasFieldOrPropertyWithValue("tcpKeepAlive", true)
+            .hasFieldOrPropertyWithValue("tcpNoDelay", true)
+            .hasFieldOrPropertyWithValue("tcpLoopResources", loopResources);
 
         assertThat(configuration.getOptions())
             .containsEntry("lock_timeout", "10s")
@@ -114,7 +129,10 @@ void configurationDefaults() {
             .hasFieldOrPropertyWithValue("port", 5432)
             .hasFieldOrProperty("options")
             .hasFieldOrPropertyWithValue("username", "test-username")
-            .hasFieldOrProperty("sslConfig");
+            .hasFieldOrProperty("sslConfig")
+            .hasFieldOrPropertyWithValue("tcpKeepAlive", false)
+            .hasFieldOrPropertyWithValue("tcpNoDelay", false)
+            .hasFieldOrPropertyWithValue("tcpLoopResources", null);
 
         assertThat(configuration.getOptions())
             .containsEntry("search_path", "test-schema");