Replace IntObjectHashMap with NonBlockingHashMapLong (#459)

Author: Ryland Degnan

build.gradle

@@ -80,8 +80,8 @@ subprojects {
     }
 
     dependencies {
-        compile "io.projectreactor:reactor-core:3.1.1.RELEASE"
-        compile "io.netty:netty-buffer:4.1.16.Final"
+        compile "io.projectreactor:reactor-core:3.1.2.RELEASE"
+        compile "io.netty:netty-buffer:4.1.17.Final"
         compile "org.reactivestreams:reactive-streams:1.0.1"
         compile "org.slf4j:slf4j-api:1.7.25"
         compile "com.google.code.findbugs:jsr305:3.0.2"
@@ -90,7 +90,7 @@ subprojects {
         testCompile "org.mockito:mockito-core:2.10.0"
         testCompile "org.hamcrest:hamcrest-library:1.3"
         testCompile "org.slf4j:slf4j-log4j12:1.7.25"
-        testCompile "io.projectreactor:reactor-test:3.1.1.RELEASE"
+        testCompile "io.projectreactor:reactor-test:3.1.2.RELEASE"
     }
 
     publishing {

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

@@ -500,6 +500,7 @@ public static int initialRequestN(final Frame frame) {
           result = 1;
           break;
         case FIRE_AND_FORGET:
+        case METADATA_PUSH:
           result = 0;
           break;
         default:

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

@@ -16,41 +16,36 @@
 
 package io.rsocket;
 
-import static io.rsocket.util.ExceptionUtil.noStacktrace;
-
 import io.netty.buffer.Unpooled;
-import io.netty.util.collection.IntObjectHashMap;
 import io.rsocket.exceptions.ConnectionException;
 import io.rsocket.exceptions.Exceptions;
 import io.rsocket.internal.LimitableRequestPublisher;
 import io.rsocket.internal.UnboundedProcessor;
-import java.nio.channels.ClosedChannelException;
+import io.rsocket.util.NonBlockingHashMapLong;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import reactor.core.Disposable;
+import reactor.core.publisher.*;
+
+import javax.annotation.Nullable;
 import java.time.Duration;
 import java.util.Collection;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
-import javax.annotation.Nullable;
-import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
-import reactor.core.Disposable;
-import reactor.core.publisher.*;
 
 /** Client Side of a RSocket socket. Sends {@link Frame}s to a {@link RSocketServer} */
 class RSocketClient implements RSocket {
 
-  private static final ClosedChannelException CLOSED_CHANNEL_EXCEPTION =
-      noStacktrace(new ClosedChannelException());
-
   private final DuplexConnection connection;
   private final Function<Frame, ? extends Payload> frameDecoder;
   private final Consumer<Throwable> errorConsumer;
   private final StreamIdSupplier streamIdSupplier;
   private final MonoProcessor<Void> started;
-  private final IntObjectHashMap<LimitableRequestPublisher> senders;
-  private final IntObjectHashMap<Subscriber<Payload>> receivers;
+  private final NonBlockingHashMapLong<LimitableRequestPublisher> senders;
+  private final NonBlockingHashMapLong<UnicastProcessor<Payload>> receivers;
   private final AtomicInteger missedAckCounter;
 
   private final UnboundedProcessor<Frame> sendProcessor;
@@ -80,8 +75,8 @@ class RSocketClient implements RSocket {
     this.errorConsumer = errorConsumer;
     this.streamIdSupplier = streamIdSupplier;
     this.started = MonoProcessor.create();
-    this.senders = new IntObjectHashMap<>(256, 0.9f);
-    this.receivers = new IntObjectHashMap<>(256, 0.9f);
+    this.senders = new NonBlockingHashMapLong<>(256);
+    this.receivers = new NonBlockingHashMapLong<>(256);
     this.missedAckCounter = new AtomicInteger();
 
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
@@ -127,7 +122,7 @@ class RSocketClient implements RSocket {
   }
 
   private void handleSendProcessorError(Throwable t) {
-    Collection<Subscriber<Payload>> values;
+    Collection<UnicastProcessor<Payload>> values;
     Collection<LimitableRequestPublisher> values1;
     synchronized (RSocketClient.this) {
       values = receivers.values();
@@ -151,7 +146,7 @@ private void handleSendProcessorCancel(SignalType t) {
     if (SignalType.ON_ERROR == t) {
       return;
     }
-    Collection<Subscriber<Payload>> values;
+    Collection<UnicastProcessor<Payload>> values;
     Collection<LimitableRequestPublisher> values1;
     synchronized (RSocketClient.this) {
       values = receivers.values();
@@ -222,10 +217,15 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
 
   @Override
   public Mono<Void> metadataPush(Payload payload) {
-    final Frame requestFrame = Frame.Request.from(0, FrameType.METADATA_PUSH, payload, 1);
-    payload.release();
-    sendProcessor.onNext(requestFrame);
-    return Mono.empty();
+    Mono<Void> defer =
+        Mono.fromRunnable(
+            () -> {
+              final Frame requestFrame = Frame.Request.from(0, FrameType.METADATA_PUSH, payload, 1);
+              payload.release();
+              sendProcessor.onNext(requestFrame);
+            });
+
+    return started.then(defer);
   }
 
   @Override
@@ -303,7 +303,7 @@ private Mono<Payload> handleRequestResponse(final Payload payload) {
                   Frame.Request.from(streamId, FrameType.REQUEST_RESPONSE, payload, 1);
               payload.release();
 
-              MonoProcessor<Payload> receiver = MonoProcessor.create();
+              UnicastProcessor<Payload> receiver = UnicastProcessor.create();
 
               synchronized (this) {
                 receivers.put(streamId, receiver);
@@ -312,6 +312,7 @@ private Mono<Payload> handleRequestResponse(final Payload payload) {
               sendProcessor.onNext(requestFrame);
 
               return receiver
+                  .singleOrEmpty()
                   .doOnError(t -> sendProcessor.onNext(Frame.Error.from(streamId, t)))
                   .doOnCancel(() -> sendProcessor.onNext(Frame.Cancel.from(streamId)))
                   .doFinally(
@@ -438,7 +439,7 @@ private boolean contains(int streamId) {
 
   protected void cleanup() {
     try {
-      Collection<Subscriber<Payload>> subscribers;
+      Collection<UnicastProcessor<Payload>> subscribers;
       Collection<LimitableRequestPublisher> publishers;
       synchronized (RSocketClient.this) {
         subscribers = receivers.values();
@@ -468,9 +469,9 @@ private synchronized void cleanUpLimitableRequestPublisher(
     }
   }
 
-  private synchronized void cleanUpSubscriber(Subscriber<?> subscriber) {
+  private synchronized void cleanUpSubscriber(UnicastProcessor<?> subscriber) {
     try {
-      subscriber.onError(CLOSED_CHANNEL_EXCEPTION);
+      subscriber.cancel();
     } catch (Throwable t) {
       errorConsumer.accept(t);
     }

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

@@ -16,20 +16,12 @@
 
 package io.rsocket;
 
-import static io.rsocket.Frame.Request.initialRequestN;
-import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_C;
-import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M;
-
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
-import io.netty.util.collection.IntObjectHashMap;
 import io.rsocket.exceptions.ApplicationException;
 import io.rsocket.internal.LimitableRequestPublisher;
 import io.rsocket.internal.UnboundedProcessor;
-import java.util.Collection;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import javax.annotation.Nullable;
+import io.rsocket.util.NonBlockingHashMapLong;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
@@ -39,6 +31,15 @@
 import reactor.core.publisher.SignalType;
 import reactor.core.publisher.UnicastProcessor;
 
+import javax.annotation.Nullable;
+import java.util.Collection;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static io.rsocket.Frame.Request.initialRequestN;
+import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_C;
+import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M;
+
 /** Server side RSocket. Receives {@link Frame}s from a {@link RSocketClient} */
 class RSocketServer implements RSocket {
 
@@ -47,8 +48,8 @@ class RSocketServer implements RSocket {
   private final Function<Frame, ? extends Payload> frameDecoder;
   private final Consumer<Throwable> errorConsumer;
 
-  private final IntObjectHashMap<Subscription> sendingSubscriptions;
-  private final IntObjectHashMap<UnicastProcessor<Payload>> channelProcessors;
+  private final NonBlockingHashMapLong<Subscription> sendingSubscriptions;
+  private final NonBlockingHashMapLong<UnicastProcessor<Payload>> channelProcessors;
 
   private final UnboundedProcessor<Frame> sendProcessor;
   private Disposable receiveDisposable;
@@ -62,8 +63,8 @@ class RSocketServer implements RSocket {
     this.requestHandler = requestHandler;
     this.frameDecoder = frameDecoder;
     this.errorConsumer = errorConsumer;
-    this.sendingSubscriptions = new IntObjectHashMap<>();
-    this.channelProcessors = new IntObjectHashMap<>();
+    this.sendingSubscriptions = new NonBlockingHashMapLong<>();
+    this.channelProcessors = new NonBlockingHashMapLong<>();
 
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
     // connections