fixes

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>
Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>
Signed-off-by: OlegDokuka <odokuka@vmware.com>

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java

@@ -329,6 +329,7 @@ void tryReestablishSession(Tuple2<ByteBuf, DuplexConnection> tuple2) {
       }
 
       nextDuplexConnection.dispose();
+      nextDuplexConnection.receive().subscribe();
       throw exception; // assume retryable exception
     } else {
       if (logger.isDebugEnabled()) {

rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java

@@ -96,7 +96,10 @@ public boolean connect(DuplexConnection nextConnection) {
     if (activeConnection != DisposedConnection.INSTANCE
         && ACTIVE_CONNECTION.compareAndSet(this, activeConnection, nextConnection)) {
 
-      activeConnection.dispose();
+      if (!activeConnection.isDisposed()) {
+        activeConnection.sendErrorAndClose(
+            new ConnectionErrorException("Connection unexpectedly replaced"));
+      }
 
       initConnection(nextConnection);
 
@@ -163,7 +166,7 @@ void initConnection(DuplexConnection nextConnection) {
 
   public void disconnect() {
     final DuplexConnection activeConnection = this.activeConnection;
-    if (activeConnection != DisposedConnection.INSTANCE) {
+    if (activeConnection != DisposedConnection.INSTANCE && !activeConnection.isDisposed()) {
       activeConnection.dispose();
     }
   }