Support unix domain socket connections

We now support connections using unix domain sockets
for local IPC transport instead of TCP usage.

[resolves #181]

Author: mp911de

README.md

@@ -6,6 +6,7 @@ This driver provides the following features:
 
 * Login with username/password (MD5, SASL/SCRAM) or implicit trust
 * SCRAM authentication
+* Unix Domain Socket transport
 * TLS
 * Explicit transactions
 * Notifications
@@ -70,6 +71,7 @@ Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
 | `driver`          | Must be `postgresql`.
 | `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)_
 | `database`        | Database to select. _(Optional)_

pom.xml

@@ -37,6 +37,7 @@
         <jsr305.version>3.0.2</jsr305.version>
         <junit.version>5.5.2</junit.version>
         <jmh.version>1.21</jmh.version>
+        <netty.version>4.1.42.Final</netty.version>
         <mbr.version>0.1.0.RELEASE</mbr.version>
         <logback.version>1.2.3</logback.version>
         <mockito.version>3.0.0</mockito.version>
@@ -60,6 +61,13 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
+            <dependency>
+                <groupId>io.netty</groupId>
+                <artifactId>netty-bom</artifactId>
+                <version>${netty.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
             <dependency>
                 <groupId>org.junit</groupId>
                 <artifactId>junit-bom</artifactId>
@@ -171,6 +179,12 @@
             <version>${hikari-cp.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport-native-kqueue</artifactId>
+            <classifier>osx-x86_64</classifier>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>postgresql</artifactId>

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

@@ -71,25 +71,28 @@ public final class PostgresqlConnectionConfiguration {
 
     private final String schema;
 
+    private final String socket;
+
     private final String username;
 
     private final SSLConfig sslConfig;
 
     private PostgresqlConnectionConfiguration(String applicationName, boolean autodetectExtensions,
-                                              @Nullable Duration connectTimeout, @Nullable String database, List<Extension> extensions, boolean forceBinary, String host,
-                                              @Nullable Map<String, String> options, @Nullable CharSequence password, int port, @Nullable String schema, String username,
+                                              @Nullable Duration connectTimeout, @Nullable String database, List<Extension> extensions, boolean forceBinary, @Nullable String host,
+                                              @Nullable Map<String, String> options, @Nullable CharSequence password, int port, @Nullable String schema, @Nullable String socket, String username,
                                               SSLConfig sslConfig) {
         this.applicationName = Assert.requireNonNull(applicationName, "applicationName must not be null");
         this.autodetectExtensions = autodetectExtensions;
         this.connectTimeout = connectTimeout;
         this.extensions = Assert.requireNonNull(extensions, "extensions must not be null");
         this.database = database;
         this.forceBinary = forceBinary;
-        this.host = Assert.requireNonNull(host, "host must not be null");
+        this.host = host;
         this.options = options;
         this.password = password;
         this.port = port;
         this.schema = schema;
+        this.socket = socket;
         this.username = Assert.requireNonNull(username, "username must not be null");
         this.sslConfig = sslConfig;
     }
@@ -150,10 +153,22 @@ List<Extension> getExtensions() {
         return this.extensions;
     }
 
+    @Nullable
     String getHost() {
         return this.host;
     }
 
+    String getRequiredHost() {
+
+        String host = getHost();
+
+        if (host == null || host.isEmpty()) {
+            throw new IllegalStateException("Connection is configured for socket connections and not for host usage");
+        }
+
+        return host;
+    }
+
     @Nullable
     Map<String, String> getOptions() {
         return this.options;
@@ -173,6 +188,22 @@ String getSchema() {
         return this.schema;
     }
 
+    @Nullable
+    String getSocket() {
+        return this.socket;
+    }
+
+    String getRequiredSocket() {
+
+        String socket = getSocket();
+
+        if (socket == null || socket.isEmpty()) {
+            throw new IllegalStateException("Connection is configured to use host and port connections and not for socket usage");
+        }
+
+        return socket;
+    }
+
     String getUsername() {
         return this.username;
     }
@@ -185,6 +216,10 @@ boolean isForceBinary() {
         return this.forceBinary;
     }
 
+    boolean isUseSocket() {
+        return getSocket() != null;
+    }
+
     SSLConfig getSslConfig() {
         return this.sslConfig;
     }
@@ -223,6 +258,9 @@ public static final class Builder {
         @Nullable
         private String schema;
 
+        @Nullable
+        private String socket;
+
         @Nullable
         private String sslCert = null;
 
@@ -275,16 +313,20 @@ public Builder autodetectExtensions(boolean autodetectExtensions) {
          */
         public PostgresqlConnectionConfiguration build() {
 
-            if (this.host == null) {
-                throw new IllegalArgumentException("host must not be null");
+            if (this.host == null && this.socket == null) {
+                throw new IllegalArgumentException("host or socket must not be null");
+            }
+
+            if (this.host != null && this.socket != null) {
+                throw new IllegalArgumentException("Connection must be configured for either host/port or socket usage but not both");
             }
 
             if (this.username == null) {
                 throw new IllegalArgumentException("username must not be null");
             }
 
             return new PostgresqlConnectionConfiguration(this.applicationName, this.autodetectExtensions, this.connectTimeout, this.database, this.extensions, this.forceBinary, this.host,
-                this.options, this.password, this.port, this.schema, this.username, this.createSslConfig());
+                this.options, this.password, this.port, this.schema, this.socket, this.username, this.createSslConfig());
         }
 
         /**
@@ -418,6 +460,19 @@ public Builder schema(@Nullable String schema) {
             return this;
         }
 
+        /**
+         * Configure the unix domain socket to connect to.
+         *
+         * @param socket the socket path
+         * @return this {@link Builder}
+         * @throws IllegalArgumentException if {@code socket} is {@code null}
+         */
+        public Builder socket(String socket) {
+            this.socket = Assert.requireNonNull(socket, "host must not be null");
+            sslMode(SSLMode.DISABLE);
+            return this;
+        }
+
         /**
          * Configure ssl cert for client certificate authentication.
          *
@@ -488,6 +543,7 @@ public String toString() {
                 ", port=" + this.port +
                 ", schema='" + this.schema + '\'' +
                 ", username='" + this.username + '\'' +
+                ", socket='" + this.socket + '\'' +
                 ", sslMode='" + this.sslMode + '\'' +
                 ", sslRootCert='" + this.sslRootCert + '\'' +
                 ", sslCert='" + this.sslCert + '\'' +
@@ -520,7 +576,7 @@ public Builder username(String username) {
         }
 
         private SSLConfig createSslConfig() {
-            if (this.sslMode == SSLMode.DISABLE) {
+            if (this.socket != null || this.sslMode == SSLMode.DISABLE) {
                 return new SSLConfig(SSLMode.DISABLE, null, (hostname, session) -> true);
             }
             HostnameVerifier hostnameVerifier = this.sslHostnameVerifier;

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

@@ -17,6 +17,7 @@
 package io.r2dbc.postgresql;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.unix.DomainSocketAddress;
 import io.r2dbc.postgresql.authentication.AuthenticationHandler;
 import io.r2dbc.postgresql.authentication.PasswordAuthenticationHandler;
 import io.r2dbc.postgresql.authentication.SASLAuthenticationHandler;
@@ -37,8 +38,11 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.netty.resources.ConnectionProvider;
 import reactor.util.annotation.Nullable;
 
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
@@ -50,10 +54,12 @@
  */
 public final class PostgresqlConnectionFactory implements ConnectionFactory {
 
-    private final Function<SSLConfig, Mono<? extends Client>> clientFactory;
-
     private final PostgresqlConnectionConfiguration configuration;
 
+    private final SocketAddress endpoint;
+
+    private final Function<SSLConfig, Mono<? extends Client>> clientFactory;
+
     private final Extensions extensions;
 
     /**
@@ -63,17 +69,32 @@ public final class PostgresqlConnectionFactory implements ConnectionFactory {
      * @throws IllegalArgumentException if {@code configuration} is {@code null}
      */
     public PostgresqlConnectionFactory(PostgresqlConnectionConfiguration configuration) {
-        this.clientFactory = sslConfig -> ReactorNettyClient.connect(configuration.getHost(), configuration.getPort(), configuration.getConnectTimeout(), sslConfig).cast(Client.class);
         this.configuration = Assert.requireNonNull(configuration, "configuration must not be null");
+        this.endpoint = createSocketAddress(configuration);
+        this.clientFactory = sslConfig -> ReactorNettyClient.connect(ConnectionProvider.newConnection(), this.endpoint, configuration.getConnectTimeout(), sslConfig).cast(Client.class);
         this.extensions = getExtensions(configuration);
     }
 
     PostgresqlConnectionFactory(Function<SSLConfig, Mono<? extends Client>> clientFactory, PostgresqlConnectionConfiguration configuration) {
-        this.clientFactory = Assert.requireNonNull(clientFactory, "clientFactory must not be null");
         this.configuration = Assert.requireNonNull(configuration, "configuration must not be null");
+        this.endpoint = createSocketAddress(configuration);
+        this.clientFactory = Assert.requireNonNull(clientFactory, "clientFactory must not be null");
         this.extensions = getExtensions(configuration);
     }
 
+    private static SocketAddress createSocketAddress(PostgresqlConnectionConfiguration configuration) {
+
+        if (!configuration.isUseSocket()) {
+            return InetSocketAddress.createUnresolved(configuration.getRequiredHost(), configuration.getPort());
+        }
+
+        if (configuration.isUseSocket()) {
+            return new DomainSocketAddress(configuration.getRequiredSocket());
+        }
+
+        throw new IllegalArgumentException("Cannot create SocketAddress for " + configuration);
+    }
+
     private static Extensions getExtensions(PostgresqlConnectionConfiguration configuration) {
         Extensions extensions = Extensions.from(configuration.getExtensions());
 
@@ -149,7 +170,7 @@ private Throwable cannotConnect(Throwable throwable) {
         }
 
         return new PostgresConnectionException(
-            String.format("Cannot connect to %s:%d", this.configuration.getHost(), this.configuration.getPort()), throwable
+            String.format("Cannot connect to %s", this.endpoint), throwable
         );
     }