From b8c7c2e36c2c16c8be112b67e70073cd7b70e8df Mon Sep 17 00:00:00 2001
From: Oleh Dokuka <oleh.dokuka@icloud.com>
Date: Wed, 2 Jun 2021 23:04:02 +0300
Subject: [PATCH] ensures no modifications during iteration

since Processor/Subscription termination with `cancel` or `onError` leads to the following self-removal logic, it can happen that collection concurrent modification exception may appear. To avoid so we can copy all the entries and by doing so avoid any subsequent problems

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>
Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>
---
 .../java/io/rsocket/core/RSocketRequester.java | 18 +++++++++++++-----
 .../java/io/rsocket/core/RSocketResponder.java |  9 +++++++--
 2 files changed, 20 insertions(+), 7 deletions(-)

diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java
index ae9bf6e97..d1a37e7e8 100644
--- a/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java
+++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java
@@ -50,6 +50,8 @@
 import io.rsocket.keepalive.KeepAliveSupport;
 import io.rsocket.lease.RequesterLeaseHandler;
 import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Consumer;
@@ -772,20 +774,26 @@ private void terminate(Throwable e) {
     leaseHandler.dispose();
 
     // Iterate explicitly to handle collisions with concurrent removals
-    for (IntObjectMap.PrimitiveEntry<Processor<Payload, Payload>> entry : receivers.entries()) {
+    final IntObjectMap<Processor<Payload, Payload>> receivers = this.receivers;
+    // copy to avoid collection modification from the foreach loop
+    final Collection<Processor<Payload, Payload>> receiversCopy =
+        new ArrayList<>(receivers.values());
+    for (Processor<Payload, Payload> handler : receiversCopy) {
       try {
-        entry.value().onError(e);
+        handler.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()) {
+    final IntObjectMap<Subscription> senders = this.senders;
+    // copy to avoid collection modification from the foreach loop
+    final Collection<Subscription> sendersCopy = new ArrayList<>(senders.values());
+    for (Subscription subscription : sendersCopy) {
       try {
-        entry.value().cancel();
+        subscription.cancel();
       } catch (Throwable ex) {
         if (LOGGER.isDebugEnabled()) {
           LOGGER.debug("Dropped exception", ex);
diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java
index 54f339c12..edb01ba16 100644
--- a/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java
+++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java
@@ -40,6 +40,8 @@
 import io.rsocket.internal.UnboundedProcessor;
 import io.rsocket.lease.ResponderLeaseHandler;
 import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Consumer;
@@ -264,9 +266,12 @@ private void cleanup(Throwable e) {
 
   private synchronized void cleanUpSendingSubscriptions() {
     // Iterate explicitly to handle collisions with concurrent removals
-    for (IntObjectMap.PrimitiveEntry<Subscription> entry : sendingSubscriptions.entries()) {
+    final IntObjectMap<Subscription> sendingSubscriptions = this.sendingSubscriptions;
+    final Collection<Subscription> sendingSubscriptionsCopy =
+        new ArrayList<>(sendingSubscriptions.values());
+    for (Subscription subscription : sendingSubscriptionsCopy) {
       try {
-        entry.value().cancel();
+        subscription.cancel();
       } catch (Throwable ex) {
         if (LOGGER.isDebugEnabled()) {
           LOGGER.debug("Dropped exception", ex);