Save iteration to cancel subscriptions

Closes gh-914 Signed-off-by: Rossen Stoyanchev <[email protected]>
rsocket · Oct 6, 2020 · 1ca3b99 · 1ca3b99
1 parent 7af1848
commit 1ca3b99
Showing 1 changed file with 11 additions and 13 deletions.
diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java
@@ -128,18 +128,7 @@ class RSocketResponder implements RSocket {
   }
 
   private void handleSendProcessorError(Throwable t) {
-    sendingSubscriptions
-        .values()
-        .forEach(
-            subscription -> {
-              try {
-                subscription.cancel();
-              } catch (Throwable e) {
-                if (LOGGER.isDebugEnabled()) {
-                  LOGGER.debug("Dropped exception", t);
-                }
-              }
-            });
+    cleanUpSendingSubscriptions();
 
     channelProcessors
         .values()
@@ -275,7 +264,16 @@ private void cleanup(Throwable e) {
   }
 
   private synchronized void cleanUpSendingSubscriptions() {
-    sendingSubscriptions.values().forEach(Subscription::cancel);
+    // Iterate explicitly to handle collisions with concurrent removals
+    for (IntObjectMap.PrimitiveEntry<Subscription> entry : sendingSubscriptions.entries()) {
+      try {
+        entry.value().cancel();
+      } catch (Throwable ex) {
+        if (LOGGER.isDebugEnabled()) {
+          LOGGER.debug("Dropped exception", ex);
+        }
+      }
+    }
     sendingSubscriptions.clear();
   }