Replace client factories with composable connection strategies

Author: kressi

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

@@ -0,0 +1,21 @@
+package io.r2dbc.postgresql;
+
+import io.r2dbc.postgresql.client.Client;
+import io.r2dbc.postgresql.client.ConnectionSettings;
+import reactor.core.publisher.Mono;
+
+import java.net.SocketAddress;
+import java.util.Map;
+import java.util.function.Function;
+
+public interface ConnectionStrategy {
+
+    default ConnectionStrategy andThen(boolean guard, Function<ConnectionStrategy, ConnectionStrategy> nextStrategyProvider) {
+        return guard ? nextStrategyProvider.apply(this) : this;
+    }
+
+    Mono<Client> connect();
+    ConnectionStrategy withConnectionSettings(ConnectionSettings connectionSettings);
+    ConnectionStrategy withAddress(SocketAddress address);
+    ConnectionStrategy withOptions(Map<String, String> options);
+}

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

@@ -0,0 +1,47 @@
+package io.r2dbc.postgresql;
+
+import io.netty.channel.unix.DomainSocketAddress;
+import io.r2dbc.postgresql.client.MultiHostConfiguration;
+import io.r2dbc.postgresql.client.SSLConfig;
+import io.r2dbc.postgresql.client.SSLMode;
+import io.r2dbc.postgresql.client.SingleHostConfiguration;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+
+public class ConnectionStrategyFactory {
+
+    public static ConnectionStrategy getConnectionStrategy(ClientSupplier clientSupplier, PostgresqlConnectionConfiguration configuration) {
+        SingleHostConfiguration singleHostConfiguration = configuration.getSingleHostConfiguration();
+        MultiHostConfiguration multiHostConfiguration = configuration.getMultiHostConfiguration();
+        SSLConfig sslConfig = configuration.getSslConfig();
+        SocketAddress address = singleHostConfiguration != null ? createSocketAddress(singleHostConfiguration) : null;
+        return new DefaultConnectionStrategy(clientSupplier, address, configuration, configuration.getConnectionSettings(), configuration.getOptions())
+            .andThen(!SSLMode.DISABLE.equals(sslConfig.getSslMode()), strategy -> new SslFallbackConnectionStrategy(strategy, configuration, configuration.getOptions()))
+            .andThen(multiHostConfiguration != null, strategy -> new MultiHostConnectionStrategy(strategy, createSocketAddress(multiHostConfiguration), configuration));
+    }
+
+    private static SocketAddress createSocketAddress(SingleHostConfiguration configuration) {
+        if (!configuration.isUseSocket()) {
+            return InetSocketAddress.createUnresolved(configuration.getRequiredHost(), configuration.getPort());
+        }
+        return DomainSocketFactory.getDomainSocketAddress(configuration);
+    }
+
+    static class DomainSocketFactory {
+        private static SocketAddress getDomainSocketAddress(SingleHostConfiguration configuration) {
+            return new DomainSocketAddress(configuration.getRequiredSocket());
+        }
+    }
+
+    private static List<SocketAddress> createSocketAddress(MultiHostConfiguration configuration) {
+        List<SocketAddress> addressList = new ArrayList<>(configuration.getHosts().size());
+        for (MultiHostConfiguration.ServerHost host : configuration.getHosts()) {
+            addressList.add(InetSocketAddress.createUnresolved(host.getHost(), host.getPort()));
+        }
+        return addressList;
+    }
+
+}

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

@@ -0,0 +1,75 @@
+package io.r2dbc.postgresql;
+
+import io.r2dbc.postgresql.authentication.AuthenticationHandler;
+import io.r2dbc.postgresql.authentication.PasswordAuthenticationHandler;
+import io.r2dbc.postgresql.authentication.SASLAuthenticationHandler;
+import io.r2dbc.postgresql.client.Client;
+import io.r2dbc.postgresql.client.ConnectionSettings;
+import io.r2dbc.postgresql.client.StartupMessageFlow;
+import io.r2dbc.postgresql.message.backend.AuthenticationMessage;
+import io.r2dbc.postgresql.util.Assert;
+import reactor.core.publisher.Mono;
+
+import javax.annotation.Nullable;
+import java.net.SocketAddress;
+import java.util.Map;
+
+public class DefaultConnectionStrategy implements ConnectionStrategy {
+
+    private final ClientSupplier clientSupplier;
+    private final SocketAddress address;
+    private final PostgresqlConnectionConfiguration configuration;
+    private final ConnectionSettings connectionSettings;
+    private final Map<String, String> options;
+
+    DefaultConnectionStrategy(
+        ClientSupplier clientSupplier,
+        @Nullable SocketAddress address,
+        PostgresqlConnectionConfiguration configuration,
+        ConnectionSettings connectionSettings,
+        @Nullable Map<String, String> options
+    ) {
+        this.clientSupplier = clientSupplier;
+        this.address = address;
+        this.configuration = configuration;
+        this.connectionSettings = connectionSettings;
+        this.options = options;
+    }
+
+    @Override
+    public Mono<Client> connect() {
+        Assert.requireNonNull(this.address, "address must not be null");
+        return this.clientSupplier.connect(this.address, this.connectionSettings)
+            .delayUntil(client -> StartupMessageFlow
+                .exchange(this.configuration.getApplicationName(), this::getAuthenticationHandler, client, this.configuration.getDatabase(), this.configuration.getUsername(), this.options)
+                .handle(ExceptionFactory.INSTANCE::handleErrorResponse));
+    }
+
+    @Override
+    public ConnectionStrategy withConnectionSettings(ConnectionSettings connectionSettings) {
+        return new DefaultConnectionStrategy(this.clientSupplier, this.address, this.configuration, connectionSettings, this.options);
+    }
+
+    @Override
+    public ConnectionStrategy withAddress(SocketAddress address) {
+        return new DefaultConnectionStrategy(this.clientSupplier, address, this.configuration, this.connectionSettings, this.options);
+    }
+
+    @Override
+    public ConnectionStrategy withOptions(Map<String, String> options) {
+        return new DefaultConnectionStrategy(this.clientSupplier, this.address, this.configuration, this.connectionSettings, options);
+    }
+
+    protected AuthenticationHandler getAuthenticationHandler(AuthenticationMessage message) {
+        if (PasswordAuthenticationHandler.supports(message)) {
+            CharSequence password = Assert.requireNonNull(this.configuration.getPassword(), "Password must not be null");
+            return new PasswordAuthenticationHandler(password, this.configuration.getUsername());
+        } else if (SASLAuthenticationHandler.supports(message)) {
+            CharSequence password = Assert.requireNonNull(this.configuration.getPassword(), "Password must not be null");
+            return new SASLAuthenticationHandler(password, this.configuration.getUsername());
+        } else {
+            throw new IllegalStateException(String.format("Unable to provide AuthenticationHandler capable of handling %s", message));
+        }
+    }
+
+}