Additional cases of map iteration and removal

A follow-up fix that extends the changes in
1ca3b99

See gh-914

Signed-off-by: Rossen Stoyanchev <[email protected]>

Author: rstoyanchev

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

@@ -771,30 +771,28 @@ private void terminate(Throwable e) {
     connection.dispose();
     leaseHandler.dispose();
 
-    receivers
-        .values()
-        .forEach(
-            receiver -> {
-              try {
-                receiver.onError(e);
-              } catch (Throwable t) {
-                if (LOGGER.isDebugEnabled()) {
-                  LOGGER.debug("Dropped exception", t);
-                }
-              }
-            });
-    senders
-        .values()
-        .forEach(
-            sender -> {
-              try {
-                sender.cancel();
-              } catch (Throwable t) {
-                if (LOGGER.isDebugEnabled()) {
-                  LOGGER.debug("Dropped exception", t);
-                }
-              }
-            });
+    // Iterate explicitly to handle collisions with concurrent removals
+    for (IntObjectMap.PrimitiveEntry<Processor<Payload, Payload>> entry : receivers.entries()) {
+      try {
+        entry.value().onError(e);
+      } catch (Throwable ex) {
+        if (LOGGER.isDebugEnabled()) {
+          LOGGER.debug("Dropped exception", ex);
+        }
+      }
+    }
+
+    // Iterate explicitly to handle collisions with concurrent removals
+    for (IntObjectMap.PrimitiveEntry<Subscription> entry : senders.entries()) {
+      try {
+        entry.value().cancel();
+      } catch (Throwable ex) {
+        if (LOGGER.isDebugEnabled()) {
+          LOGGER.debug("Dropped exception", ex);
+        }
+      }
+    }
+
     senders.clear();
     receivers.clear();
     sendProcessor.dispose();

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

@@ -27,7 +27,14 @@
 import io.rsocket.DuplexConnection;
 import io.rsocket.Payload;
 import io.rsocket.RSocket;
-import io.rsocket.frame.*;
+import io.rsocket.frame.CancelFrameCodec;
+import io.rsocket.frame.ErrorFrameCodec;
+import io.rsocket.frame.FrameHeaderCodec;
+import io.rsocket.frame.FrameType;
+import io.rsocket.frame.PayloadFrameCodec;
+import io.rsocket.frame.RequestChannelFrameCodec;
+import io.rsocket.frame.RequestNFrameCodec;
+import io.rsocket.frame.RequestStreamFrameCodec;
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.internal.SynchronizedIntObjectHashMap;
 import io.rsocket.internal.UnboundedProcessor;
@@ -46,7 +53,11 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.Exceptions;
-import reactor.core.publisher.*;
+import reactor.core.publisher.BaseSubscriber;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
+import reactor.core.publisher.UnicastProcessor;
 import reactor.util.annotation.Nullable;
 import reactor.util.concurrent.Queues;
 
@@ -129,19 +140,7 @@ class RSocketResponder implements RSocket {
 
   private void handleSendProcessorError(Throwable t) {
     cleanUpSendingSubscriptions();
-
-    channelProcessors
-        .values()
-        .forEach(
-            subscription -> {
-              try {
-                subscription.onError(t);
-              } catch (Throwable e) {
-                if (LOGGER.isDebugEnabled()) {
-                  LOGGER.debug("Dropped exception", t);
-                }
-              }
-            });
+    cleanUpChannelProcessors(t);
   }
 
   private void tryTerminateOnConnectionError(Throwable e) {
@@ -278,16 +277,15 @@ private synchronized void cleanUpSendingSubscriptions() {
   }
 
   private synchronized void cleanUpChannelProcessors(Throwable e) {
-    channelProcessors
-        .values()
-        .forEach(
-            payloadPayloadProcessor -> {
-              try {
-                payloadPayloadProcessor.onError(e);
-              } catch (Throwable t) {
-                // noops
-              }
-            });
+    // Iterate explicitly to handle collisions with concurrent removals
+    for (IntObjectMap.PrimitiveEntry<Processor<Payload, Payload>> entry :
+        channelProcessors.entries()) {
+      try {
+        entry.value().onError(e);
+      } catch (Throwable ex) {
+        // noops
+      }
+    }
     channelProcessors.clear();
   }