Polishing

Eagerly remove Conversation on complete to avoid cancellation if the conversation is already completed (e.g. concatMap(close())).
Remove error logging in FluxDiscardOnCancel to avoid unnecessary noise.

[#231]

Author: mp911de

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

@@ -182,9 +182,10 @@ public Mono<Void> close() {
 
             drainError(EXPECTED);
 
+            boolean connected = isConnected();
             if (this.isClosed.compareAndSet(false, true)) {
 
-                if (!isConnected() || this.processId == null) {
+                if (!connected || this.processId == null) {
                     this.connection.dispose();
                     return this.connection.onDispose();
                 }
@@ -452,7 +453,7 @@ private void drainError(Supplier<? extends Throwable> supplier) {
         Conversation receiver;
 
         while ((receiver = this.messageSubscriber.conversations.poll()) != null) {
-            receiver.sink.error(supplier.get());
+            receiver.onError(supplier.get());
         }
 
         if (!this.notificationProcessor.isTerminated()) {
@@ -642,27 +643,55 @@ private long decrementDemand() {
         }
 
         /**
-         * Emit a {@link BackendMessage}. Returns whether the emission should be continued by returning {@literal true} or whether the conversation is complete by returning {@literal false}.
+         * Check whether the {@link BackendMessage} can complete the conversation.
          *
          * @param item
          * @return
          */
-        public boolean emit(BackendMessage item) {
+        public boolean canComplete(BackendMessage item) {
+            return this.takeUntil.test(item);
+        }
 
-            if (this.sink.isCancelled()) {
-                ReferenceCountUtil.release(item);
+        /**
+         * Complete the conversation.
+         *
+         * @param item
+         * @return
+         */
+        public void complete(BackendMessage item) {
+
+            ReferenceCountUtil.release(item);
+            if (!this.sink.isCancelled()) {
+                this.sink.complete();
             }
+        }
 
-            if (this.takeUntil.test(item)) {
+        /**
+         * Emit a {@link BackendMessage}.
+         *
+         * @param item
+         * @return
+         */
+        public void emit(BackendMessage item) {
+
+            if (this.sink.isCancelled()) {
                 ReferenceCountUtil.release(item);
-                this.sink.complete();
-                return false;
             }
 
             decrementDemand();
             this.sink.next(item);
+        }
 
-            return true;
+        /**
+         * Notify the conversation about an error. Drops errors silently if the conversation is finished.
+         *
+         * @param throwable
+         */
+        public void onError(Throwable throwable) {
+
+            if (!this.sink.isCancelled()) {
+                this.sink.error(throwable);
+            }
         }
 
         public boolean hasDemand() {
@@ -789,8 +818,11 @@ private void drainLoop() {
                                 break;
                             }
 
-                            if (!conversation.emit(item)) {
+                            if (conversation.canComplete(item)) {
                                 this.conversations.poll();
+                                conversation.complete(item);
+                            } else {
+                                conversation.emit(item);
                             }
                         }
                     }

src/main/java/io/r2dbc/postgresql/util/FluxDiscardOnCancel.java

@@ -88,9 +88,7 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
-            if (this.get()) {
-                Operators.onErrorDropped(t, this.ctx);
-            } else {
+            if (!this.get()) {
                 this.actual.onError(t);
             }
         }