moves error propagation out of the synchronise to avoid deadlock (#1060)

(cherry picked from commit 6426e45)

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java

@@ -345,15 +345,17 @@ private void terminate(Throwable e) {
       requesterLeaseTracker.dispose(e);
     }
 
+    final Collection<FrameHandler> activeStreamsCopy;
     synchronized (this) {
       final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
-      final Collection<FrameHandler> activeStreamsCopy = new ArrayList<>(activeStreams.values());
-      for (FrameHandler handler : activeStreamsCopy) {
-        if (handler != null) {
-          try {
-            handler.handleError(e);
-          } catch (Throwable ignored) {
-          }
+      activeStreamsCopy = new ArrayList<>(activeStreams.values());
+    }
+
+    for (FrameHandler handler : activeStreamsCopy) {
+      if (handler != null) {
+        try {
+          handler.handleError(e);
+        } catch (Throwable ignored) {
         }
       }
     }

rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java

@@ -185,15 +185,18 @@ final void doOnDispose() {
     requestHandler.dispose();
   }
 
-  private synchronized void cleanUpSendingSubscriptions() {
-    final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
-    final Collection<FrameHandler> activeStreamsCopy = new ArrayList<>(activeStreams.values());
+  private void cleanUpSendingSubscriptions() {
+    final Collection<FrameHandler> activeStreamsCopy;
+    synchronized (this) {
+      final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
+      activeStreamsCopy = new ArrayList<>(activeStreams.values());
+    }
+
     for (FrameHandler handler : activeStreamsCopy) {
       if (handler != null) {
         handler.handleCancel();
       }
     }
-    activeStreams.clear();
   }
 
   final void handleFrame(ByteBuf frame) {