Remove doOnNext(…) hook to close the sink if the connection is closed…

… while emitting requests.

Once the conversation is accepted, we no longer need to check on a new backend message whether the connection is closed as a channelInactive()/connection.close() signal terminates conversations anyway.

[#492]

Signed-off-by: Mark Paluch <mpaluch@vmware.com>

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

@@ -55,6 +55,7 @@
 import reactor.core.publisher.Operators;
 import reactor.core.scheduler.Schedulers;
 import reactor.netty.Connection;
+import reactor.netty.channel.AbortedException;
 import reactor.netty.resources.ConnectionProvider;
 import reactor.netty.resources.LoopResources;
 import reactor.netty.tcp.TcpClient;
@@ -513,6 +514,11 @@ private void handleClose() {
     }
 
     private void handleConnectionError(Throwable error) {
+
+        if (AbortedException.isConnectionReset(error) && !isConnected()) {
+            drainError(() -> this.messageSubscriber.createClientClosedException(error));
+        }
+
         drainError(() -> new PostgresConnectionException(error));
     }
 
@@ -554,6 +560,10 @@ public PostgresConnectionClosedException(String reason) {
             super(reason);
         }
 
+        public PostgresConnectionClosedException(String reason, @Nullable Throwable cause) {
+            super(reason, cause);
+        }
+
     }
 
     static class PostgresConnectionException extends R2dbcNonTransientResourceException {
@@ -689,7 +699,7 @@ private class BackendMessageSubscriber implements CoreSubscriber<BackendMessage>
 
         private Subscription upstream;
 
-        public Flux<BackendMessage> addConversation(Predicate<BackendMessage> takeUntil, Publisher<FrontendMessage> requests, Consumer<Flux<FrontendMessage>> sender,
+        public Flux<BackendMessage> addConversation(Predicate<BackendMessage> takeUntil, Publisher<FrontendMessage> requests, Consumer<Publisher<FrontendMessage>> sender,
                                                     Supplier<Boolean> isConnected) {
 
             return Flux.create(sink -> {
@@ -707,13 +717,7 @@ public Flux<BackendMessage> addConversation(Predicate<BackendMessage> takeUntil,
                             return;
                         }
 
-                        Flux<FrontendMessage> requestMessages = Flux.from(requests).doOnNext(m -> {
-                            if (!isConnected.get()) {
-                                sink.error(new PostgresConnectionClosedException("Cannot exchange messages because the connection is closed"));
-                            }
-                        });
-
-                        sender.accept(requestMessages);
+                        sender.accept(requests);
                     } else {
                         sink.error(new RequestQueueException("Cannot exchange messages because the request queue limit is exceeded"));
                     }
@@ -722,7 +726,11 @@ public Flux<BackendMessage> addConversation(Predicate<BackendMessage> takeUntil,
         }
 
         PostgresConnectionClosedException createClientClosedException() {
-            return new PostgresConnectionClosedException("Cannot exchange messages because the connection is closed");
+            return createClientClosedException(null);
+        }
+
+        PostgresConnectionClosedException createClientClosedException(@Nullable Throwable cause) {
+            return new PostgresConnectionClosedException("Cannot exchange messages because the connection is closed", cause);
         }
 
         /**