fixes possible racing on direct processor in tests

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

Author: OlegDokuka

Diff for: rsocket-core/src/test/java/io/rsocket/test/util/LocalDuplexConnection.java

@@ -21,19 +21,20 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
 
 public class LocalDuplexConnection implements DuplexConnection {
-  private final DirectProcessor<Frame> send;
+  private final FluxSink<Frame> sendFluxSink;
   private final DirectProcessor<Frame> receive;
   private final MonoProcessor<Void> onClose;
   private final String name;
 
   public LocalDuplexConnection(
       String name, DirectProcessor<Frame> send, DirectProcessor<Frame> receive) {
     this.name = name;
-    this.send = send;
+    this.sendFluxSink = send.serialize().sink();
     this.receive = receive;
     onClose = MonoProcessor.create();
   }
@@ -42,8 +43,8 @@ public LocalDuplexConnection(
   public Mono<Void> send(Publisher<Frame> frame) {
     return Flux.from(frame)
         .doOnNext(f -> System.out.println(name + " - " + f.toString()))
-        .doOnNext(send::onNext)
-        .doOnError(send::onError)
+        .doOnNext(sendFluxSink::next)
+        .doOnError(sendFluxSink::error)
         .then();
   }
 

Diff for: rsocket-core/src/test/java/io/rsocket/test/util/TestDuplexConnection.java

@@ -27,6 +27,7 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
 
@@ -40,7 +41,9 @@ public class TestDuplexConnection implements DuplexConnection {
 
   private final LinkedBlockingQueue<Frame> sent;
   private final DirectProcessor<Frame> sentPublisher;
+  private final FluxSink<Frame> sentFluxSink;
   private final DirectProcessor<Frame> received;
+  private final FluxSink<Frame> receivedFluxSink;
   private final MonoProcessor<Void> onClose;
   private final ConcurrentLinkedQueue<Subscriber<Frame>> sendSubscribers;
   private volatile double availability = 1;
@@ -49,7 +52,9 @@ public class TestDuplexConnection implements DuplexConnection {
   public TestDuplexConnection() {
     sent = new LinkedBlockingQueue<>();
     received = DirectProcessor.create();
+    receivedFluxSink = received.serialize().sink();
     sentPublisher = DirectProcessor.create();
+    sentFluxSink = sentPublisher.serialize().sink();
     sendSubscribers = new ConcurrentLinkedQueue<>();
     onClose = MonoProcessor.create();
   }
@@ -65,7 +70,7 @@ public Mono<Void> send(Publisher<Frame> frames) {
         .doOnNext(
             frame -> {
               sent.offer(frame);
-              sentPublisher.onNext(frame);
+              sentFluxSink.next(frame);
             })
         .doOnError(throwable -> logger.error("Error in send stream on test connection.", throwable))
         .subscribe(subscriber);
@@ -116,7 +121,7 @@ public Publisher<Frame> getSentAsPublisher() {
 
   public void addToReceivedBuffer(Frame... received) {
     for (Frame frame : received) {
-      this.received.onNext(frame);
+      this.receivedFluxSink.next(frame);
     }
   }