Polishing.

Refine exception propagation by collecting exceptions as list instead of using an atomic reference.
Also, introduce exception aggregator for easier exception surfacing.

[resolves #678][#677]

Author: mp911de

README.md

@@ -607,6 +607,9 @@ Logging facilities:
 
 * Driver Logging (`io.r2dbc.postgresql`)
 * Query Logging (`io.r2dbc.postgresql.QUERY` on `DEBUG` level)
+* Connection Context (`io.r2dbc.postgresql.client.ConnectionContext`)
+    * `DEBUG` level enables connection and process identifiers in log messages and exceptions (`[cid: 0x1][pid: 109]`)
+    * `TRACE` level enables socket information (remote and local addresses) to the connection context (`[cid: 0x1][pid: 109][id: 0x79dfc4d4, L:/127.0.0.1:49391 - R:localhost/127.0.0.1:49366]`)
 * Parameters' values Logging (`io.r2dbc.postgresql.PARAM` on `DEBUG` level)
 * Transport Logging (`io.r2dbc.postgresql.client`)
     * `DEBUG` enables `Message` exchange logging

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

@@ -35,7 +35,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 import static io.r2dbc.postgresql.MultiHostConnectionStrategy.TargetServerType.ANY;
@@ -73,7 +74,7 @@ public final class MultiHostConnectionStrategy implements ConnectionStrategy {
 
     @Override
     public Mono<Client> connect() {
-        return Mono.defer(() -> connect(this.multiHostConfiguration.getTargetServerType()));
+        return connect(this.multiHostConfiguration.getTargetServerType());
     }
 
     @Override
@@ -83,40 +84,37 @@ public String toString() {
     }
 
     public Mono<Client> connect(TargetServerType targetServerType) {
-        AtomicReference<Throwable> exceptionRef = new AtomicReference<>();
+        List<Throwable> errors = new CopyOnWriteArrayList<>();
 
-        return Mono.defer(() -> attemptConnection(targetServerType))
+        return attemptConnection(targetServerType, errors::add)
             .onErrorResume(e -> {
-                if (!exceptionRef.compareAndSet(null, e)) {
-                    exceptionRef.get().addSuppressed(e);
-                }
+                errors.add(e);
                 return Mono.empty();
             })
-            .switchIfEmpty(Mono.defer(() -> targetServerType == PREFER_SECONDARY ? attemptConnection(PRIMARY) : Mono.empty()))
+            .switchIfEmpty(Mono.defer(() -> targetServerType == PREFER_SECONDARY ? attemptConnection(PRIMARY, errors::add) : Mono.empty()))
             .switchIfEmpty(Mono.error(() -> {
-                Throwable error = exceptionRef.get();
-                if (error == null) {
-                    return new PostgresqlConnectionFactory.PostgresConnectionException(String.format("No server matches target type '%s'", targetServerType), null);
+                if (errors.isEmpty()) {
+                    return new ExceptionAggregator(String.format("No server matches target type '%s'", targetServerType), null);
                 } else {
-                    return new PostgresqlConnectionFactory.PostgresConnectionException(String.format("Cannot connect to a host of %s", this.addresses), error);
+                    RuntimeException exception = new ExceptionAggregator(null, errors.size() == 1 ?
+                        errors.get(0) : null);
+
+                    if (errors.size() > 1) {
+                        errors.forEach(exception::addSuppressed);
+                    }
+                    return exception;
                 }
             }));
     }
 
-    private Mono<Client> attemptConnection(TargetServerType targetServerType) {
-        AtomicReference<Throwable> exceptionRef = new AtomicReference<>();
+    private Mono<Client> attemptConnection(TargetServerType targetServerType, Consumer<Throwable> errorHandler) {
         return getCandidates(targetServerType).concatMap(candidate -> this.attemptConnection(targetServerType, candidate)
                 .onErrorResume(e -> {
-                    if (!exceptionRef.compareAndSet(null, e)) {
-                        exceptionRef.get().addSuppressed(e);
-                    }
+                    errorHandler.accept(e);
                     this.statusMap.put(candidate, HostConnectOutcome.fail(candidate));
                     return Mono.empty();
                 }))
-            .next()
-            .switchIfEmpty(Mono.defer(() -> exceptionRef.get() != null
-                ? Mono.error(exceptionRef.get())
-                : Mono.empty()));
+            .next();
     }
 
     private Mono<Client> attemptConnection(TargetServerType targetServerType, SocketAddress candidate) {
@@ -229,6 +227,13 @@ public interface HostSelector {
 
     }
 
+    static class ExceptionAggregator extends RuntimeException {
+
+        public ExceptionAggregator(@Nullable String message, @Nullable Throwable cause) {
+            super(message, cause);
+        }
+    }
+
     private static class HostConnectOutcome {
 
         static final Clock DEFAULT_CLOCK = Clock.systemDefaultZone();

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

@@ -181,6 +181,16 @@ private Mono<PostgresqlConnection> closeWithError(Client client, Throwable throw
 
     private Throwable cannotConnect(Throwable throwable, ConnectionStrategy strategy) {
 
+        // Rewrite ExceptionAggregator but preserve suppressed exceptions to enrich the exception context
+        if (throwable instanceof MultiHostConnectionStrategy.ExceptionAggregator) {
+            String message = throwable.getMessage() != null ? String.format("Cannot connect to %s: %s", strategy, throwable.getMessage()) : String.format("Cannot connect to %s", strategy);
+            PostgresConnectionException exception = new PostgresConnectionException(message, throwable.getCause());
+            for (Throwable t : throwable.getSuppressed()) {
+                exception.addSuppressed(t);
+            }
+            return exception;
+        }
+
         if (throwable instanceof R2dbcException) {
             return throwable;
         }
@@ -226,6 +236,7 @@ static class PostgresConnectionException extends R2dbcNonTransientResourceExcept
 
         private final ErrorDetails errorDetails;
 
+
         public PostgresConnectionException(String reason, @Nullable Throwable cause) {
             super(reason, CONNECTION_DOES_NOT_EXIST, 0, null, cause);
             this.errorDetails = ErrorDetails.fromCodeAndMessage(CONNECTION_DOES_NOT_EXIST, reason);

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

@@ -100,10 +100,6 @@ public final class ReactorNettyClient implements Client {
 
     private static final boolean DEBUG_ENABLED = logger.isDebugEnabled();
 
-    private static final Supplier<PostgresConnectionClosedException> UNEXPECTED = () -> new PostgresConnectionClosedException("Connection unexpectedly closed");
-
-    private static final Supplier<PostgresConnectionClosedException> EXPECTED = () -> new PostgresConnectionClosedException("Connection closed");
-
     private final ByteBufAllocator byteBufAllocator;
 
     private final ConnectionSettings settings;
@@ -112,6 +108,10 @@ public final class ReactorNettyClient implements Client {
 
     private final Scheduler scheduler;
 
+    private final Supplier<PostgresConnectionClosedException> unexpected;
+
+    private final Supplier<PostgresConnectionClosedException> expected;
+
     private ConnectionContext context;
 
     private final Sinks.Many<Publisher<FrontendMessage>> requestSink = Sinks.many().unicast().onBackpressureBuffer();
@@ -149,12 +149,16 @@ private ReactorNettyClient(Connection connection, ConnectionSettings settings) {
         Assert.requireNonNull(connection, "Connection must not be null");
         this.settings = Assert.requireNonNull(settings, "ConnectionSettings must not be null");
 
+
         connection.addHandlerLast(new EnsureSubscribersCompleteChannelHandler(this.requestSink));
         connection.addHandlerLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE - 5, 1, 4, -4, 0));
         this.connection = connection;
         this.byteBufAllocator = connection.outbound().alloc();
 
         ConnectionContext connectionContext = new ConnectionContext().withChannelId(connection.channel().toString());
+        this.unexpected = () -> new PostgresConnectionClosedException(this.context.getMessage("Connection unexpectedly closed"));
+        this.expected = () -> new PostgresConnectionClosedException(this.context.getMessage("Connection closed"));
+
         SslHandler sslHandler = this.connection.channel().pipeline().get(SslHandler.class);
 
         if (sslHandler == null) {
@@ -211,7 +215,7 @@ public Mono<Void> close() {
 
             this.notificationProcessor.tryEmitComplete();
 
-            drainError(EXPECTED);
+            drainError(expected);
 
             boolean connected = isConnected();
             if (this.isClosed.compareAndSet(false, true)) {
@@ -536,9 +540,9 @@ private static String toString(List<Field> fields) {
 
     private void handleClose() {
         if (this.isClosed.compareAndSet(false, true)) {
-            drainError(UNEXPECTED);
+            drainError(unexpected);
         } else {
-            drainError(EXPECTED);
+            drainError(expected);
         }
     }
 
@@ -548,7 +552,7 @@ private void handleConnectionError(Throwable error) {
             drainError(() -> this.messageSubscriber.createClientClosedException(error));
         }
 
-        drainError(() -> new PostgresConnectionException(error));
+        drainError(() -> new PostgresConnectionException(this.context, error));
     }
 
     private void drainError(Supplier<? extends Throwable> supplier) {
@@ -606,8 +610,8 @@ static class PostgresConnectionException extends R2dbcNonTransientResourceExcept
 
         private final static ErrorDetails ERROR_DETAILS = ErrorDetails.fromCodeAndMessage(CONNECTION_FAILURE, "An I/O error occurred while sending to the backend or receiving from the backend");
 
-        public PostgresConnectionException(Throwable cause) {
-            super(ERROR_DETAILS.getMessage(), ERROR_DETAILS.getCode(), 0, null, cause);
+        public PostgresConnectionException(ConnectionContext context, Throwable cause) {
+            super(context.getMessage(ERROR_DETAILS.getMessage()), ERROR_DETAILS.getCode(), 0, null, cause);
         }
 
         @Override
@@ -789,7 +793,7 @@ PostgresConnectionClosedException createClientClosedException() {
         }
 
         PostgresConnectionClosedException createClientClosedException(@Nullable Throwable cause) {
-            return new PostgresConnectionClosedException("Cannot exchange messages because the connection is closed", cause);
+            return new PostgresConnectionClosedException(ReactorNettyClient.this.context.getMessage("Cannot exchange messages because the connection is closed"), cause);
         }
 
         /**

src/test/resources/logback-test.xml

@@ -24,6 +24,7 @@
     </appender>
 
     <logger name="io.r2dbc.postgresql" level="INFO"/>
+    <logger name="io.r2dbc.postgresql.client.ConnectionContext" level="INFO"/>
     <logger name="org.testcontainers"  level="INFO"/>
     <logger name="reactor.netty"       level="WARN"/>
     <logger name="stream"              level="INFO"/>