Avoid queueing in UnicastProcessor receivers

Closes gh-887

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

Author: rstoyanchev

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2018 the original author or authors.
+ * Copyright 2015-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -348,7 +348,7 @@ private Flux<Payload> handleRequestStream(final Payload payload) {
     }
 
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
-    final UnicastProcessor<Payload> receiver = UnicastProcessor.create();
+    final UnicastProcessor<Payload> receiver = UnicastProcessor.create(Queues.<Payload>one().get());
     final AtomicBoolean once = new AtomicBoolean();
 
     return Flux.defer(
@@ -456,7 +456,7 @@ private Flux<Payload> handleChannel(Flux<Payload> request) {
   private Flux<? extends Payload> handleChannel(Payload initialPayload, Flux<Payload> inboundFlux) {
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
 
-    final UnicastProcessor<Payload> receiver = UnicastProcessor.create();
+    final UnicastProcessor<Payload> receiver = UnicastProcessor.create(Queues.<Payload>one().get());
 
     return receiver
         .transform(

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2018 the original author or authors.
+ * Copyright 2015-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -48,6 +48,7 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.*;
 import reactor.util.annotation.Nullable;
+import reactor.util.concurrent.Queues;
 
 /** Responder side of RSocket. Receives {@link ByteBuf}s from a peer's {@link RSocketRequester} */
 class RSocketResponder implements RSocket {
@@ -537,7 +538,7 @@ protected void hookOnError(Throwable throwable) {
   }
 
   private void handleChannel(int streamId, Payload payload, long initialRequestN) {
-    UnicastProcessor<Payload> frames = UnicastProcessor.create();
+    UnicastProcessor<Payload> frames = UnicastProcessor.create(Queues.<Payload>one().get());
     channelProcessors.put(streamId, frames);
 
     Flux<Payload> payloads =

rsocket-core/src/test/java/io/rsocket/core/RSocketTest.java

@@ -49,6 +49,7 @@
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.test.publisher.TestPublisher;
 
@@ -158,13 +159,13 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
   }
 
   @Test(timeout = 2000)
-  public void testStream() throws Exception {
+  public void testStream() {
     Flux<Payload> responses = rule.crs.requestStream(DefaultPayload.create("Payload In"));
     StepVerifier.create(responses).expectNextCount(10).expectComplete().verify();
   }
 
   @Test(timeout = 2000)
-  public void testChannel() throws Exception {
+  public void testChannel() {
     Flux<Payload> requests =
         Flux.range(0, 10).map(i -> DefaultPayload.create("streaming in -> " + i));
     Flux<Payload> responses = rule.crs.requestChannel(requests);
@@ -543,6 +544,8 @@ public Mono<Payload> requestResponse(Payload payload) {
                 @Override
                 public Flux<Payload> requestStream(Payload payload) {
                   return Flux.range(1, 10)
+                      .subscribeOn(Schedulers.boundedElastic())
+                      .doOnNext(i -> System.out.println("Creating payload " + i))
                       .map(
                           i -> DefaultPayload.create("server got -> [" + payload.toString() + "]"));
                 }
@@ -556,6 +559,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
                       .subscribe();
 
                   return Flux.range(1, 10)
+                      .subscribeOn(Schedulers.boundedElastic())
                       .map(
                           payload ->
                               DefaultPayload.create("server got -> [" + payload.toString() + "]"));

rsocket-examples/src/test/java/io/rsocket/integration/TestingStreaming.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
 
 public class TestingStreaming {
   LocalServerTransport serverTransport = LocalServerTransport.create("test");
@@ -49,6 +50,7 @@ public void testRangeButThrowException() {
                                     }
                                   })
                               .map(l -> DefaultPayload.create("l -> " + l))
+                              .subscribeOn(Schedulers.boundedElastic())
                               .cast(Payload.class)))
               .bind(serverTransport)
               .block();
@@ -71,6 +73,7 @@ public void testRangeOfConsumers() {
                       payload ->
                           Flux.range(1, 1000)
                               .map(l -> DefaultPayload.create("l -> " + l))
+                              .subscribeOn(Schedulers.boundedElastic())
                               .cast(Payload.class)))
               .bind(serverTransport)
               .block();
@@ -104,6 +107,7 @@ public void testSingleConsumer() {
                       payload ->
                           Flux.range(1, 10_000)
                               .map(l -> DefaultPayload.create("l -> " + l))
+                              .subscribeOn(Schedulers.boundedElastic())
                               .cast(Payload.class)))
               .bind(serverTransport)
               .block();