Merge 1.0.x into master

Signed-off-by: Oleh Dokuka <[email protected]>
Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

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

@@ -35,6 +35,8 @@
 import io.rsocket.keepalive.KeepAliveSupport;
 import io.rsocket.plugins.RequestInterceptor;
 import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -344,8 +346,9 @@ private void terminate(Throwable e) {
     }
 
     synchronized (this) {
-      for (IntObjectMap.PrimitiveEntry<FrameHandler> entry : activeStreams.entries()) {
-        FrameHandler handler = entry.value();
+      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);

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

@@ -33,6 +33,8 @@
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.plugins.RequestInterceptor;
 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.Function;
@@ -184,8 +186,9 @@ final void doOnDispose() {
   }
 
   private synchronized void cleanUpSendingSubscriptions() {
-    for (IntObjectMap.PrimitiveEntry<FrameHandler> entry : activeStreams.entries()) {
-      FrameHandler handler = entry.value();
+    final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
+    final Collection<FrameHandler> activeStreamsCopy = new ArrayList<>(activeStreams.values());
+    for (FrameHandler handler : activeStreamsCopy) {
       if (handler != null) {
         handler.handleCancel();
       }