eliminates dropped error

Signed-off-by: Oleh Dokuka <[email protected]> Signed-off-by: Oleh Dokuka <[email protected]>
rsocket · Mar 28, 2022 · 571af15 · 571af15
1 parent 9a504a8
commit 571af15
Showing 1 changed file with 6 additions and 3 deletions.
diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java b/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java
@@ -192,15 +192,17 @@ public void sendErrorAndClose(RSocketErrorException rSocketErrorException) {
             t -> {
               framesSaverDisposable.dispose();
               activeReceivingSubscriber.dispose();
-              savableFramesSender.dispose();
+              savableFramesSender.onComplete();
+              savableFramesSender.cancel();
               onConnectionClosedSink.tryEmitComplete();
 
               onClose.tryEmitError(t);
             },
             () -> {
               framesSaverDisposable.dispose();
               activeReceivingSubscriber.dispose();
-              savableFramesSender.dispose();
+              savableFramesSender.onComplete();
+              savableFramesSender.cancel();
               onConnectionClosedSink.tryEmitComplete();
 
               final Throwable cause = rSocketErrorException.getCause();
@@ -253,7 +255,8 @@ void dispose(@Nullable Throwable e) {
 
     framesSaverDisposable.dispose();
     activeReceivingSubscriber.dispose();
-    savableFramesSender.dispose();
+    savableFramesSender.onComplete();
+    savableFramesSender.cancel();
     onConnectionClosedSink.tryEmitComplete();
 
     if (e != null) {