removes UnboundProcessor optimizations. Not possible today

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

Author: OlegDokuka

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

@@ -41,7 +41,6 @@
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.SignalType;
 import reactor.core.publisher.UnicastProcessor;
-import reactor.util.concurrent.Queues;
 
 /** Client Side of a RSocket socket. Sends {@link ByteBuf}s to a {@link RSocketServer} */
 class RSocketClient implements RSocket {
@@ -226,8 +225,7 @@ private Flux<Payload> handleRequestStream(final Payload payload) {
           int streamId = streamIdSupplier.nextStreamId();
 
           final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
-          final UnicastProcessor<Payload> receiver =
-              UnicastProcessor.create(Queues.<Payload>one().get());
+          final UnicastProcessor<Payload> receiver = UnicastProcessor.create();
 
           receivers.put(streamId, receiver);
 
@@ -277,8 +275,7 @@ private Flux<Payload> handleChannel(Flux<Payload> request) {
     return lifecycle.activeFlux(
         () -> {
           final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
-          final UnicastProcessor<Payload> receiver =
-              UnicastProcessor.create(Queues.<Payload>one().get());
+          final UnicastProcessor<Payload> receiver = UnicastProcessor.create();
           final int streamId = streamIdSupplier.nextStreamId();
 
           return receiver

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

@@ -84,6 +84,12 @@ public Mono<Payload> requestResponse(Payload payload) {
     rule.assertServerError("java.lang.NullPointerException: Deliberate exception.");
   }
 
+  @Test(timeout = 2000)
+  public void testStream() throws Exception {
+    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 {
     Flux<Payload> requests =
@@ -136,7 +142,9 @@ public Mono<Payload> requestResponse(Payload payload) {
 
                 @Override
                 public Flux<Payload> requestStream(Payload payload) {
-                  return Flux.never();
+                  return Flux.range(1, 10)
+                      .map(
+                          i -> DefaultPayload.create("server got -> [" + payload.toString() + "]"));
                 }
 
                 @Override