migrates from Reactor UnicastProcessor to new Sinks API (rsocket#931)

JWood48 · jakobs · web-flow · commit 29638da7e7ce · 2020-09-26T16:29:55.000+03:00
Co-authored-by: jakobs &lt;jakobs@novotek.com&gt;
diff --git a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/fnf/TaskProcessingWithServerSideNotificationsExample.java b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/fnf/TaskProcessingWithServerSideNotificationsExample.java
@@ -35,7 +35,7 @@
 import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.UnicastProcessor;
+import reactor.core.publisher.Sinks;
 import reactor.util.concurrent.Queues;
 
 /**
@@ -48,12 +48,12 @@
 public class TaskProcessingWithServerSideNotificationsExample {
 
   public static void main(String[] args) throws InterruptedException {
-    UnicastProcessor<Task> tasksProcessor =
-        UnicastProcessor.create(Queues.<Task>unboundedMultiproducer().get());
+    Sinks.Many<Task> tasksProcessor =
+        Sinks.many().unicast().onBackpressureBuffer(Queues.<Task>unboundedMultiproducer().get());
     ConcurrentMap<String, BlockingQueue<Task>> idToCompletedTasksMap = new ConcurrentHashMap<>();
     ConcurrentMap<String, RSocket> idToRSocketMap = new ConcurrentHashMap<>();
     BackgroundWorker backgroundWorker =
-        new BackgroundWorker(tasksProcessor, idToCompletedTasksMap, idToRSocketMap);
+        new BackgroundWorker(tasksProcessor.asFlux(), idToCompletedTasksMap, idToRSocketMap);
 
     RSocketServer.create(new TasksAcceptor(tasksProcessor, idToCompletedTasksMap, idToRSocketMap))
         .bindNow(TcpServerTransport.create(9991));
@@ -132,12 +132,12 @@ static class TasksAcceptor implements SocketAcceptor {
 
     static final Logger logger = LoggerFactory.getLogger(TasksAcceptor.class);
 
-    final UnicastProcessor<Task> tasksToProcess;
+    final Sinks.Many<Task> tasksToProcess;
     final ConcurrentMap<String, BlockingQueue<Task>> idToCompletedTasksMap;
     final ConcurrentMap<String, RSocket> idToRSocketMap;
 
     TasksAcceptor(
-        UnicastProcessor<Task> tasksToProcess,
+        Sinks.Many<Task> tasksToProcess,
         ConcurrentMap<String, BlockingQueue<Task>> idToCompletedTasksMap,
         ConcurrentMap<String, RSocket> idToRSocketMap) {
       this.tasksToProcess = tasksToProcess;
@@ -197,11 +197,11 @@ private static class RSocketTaskHandler implements RSocket {
       private final String id;
       private final RSocket sendingSocket;
       private ConcurrentMap<String, RSocket> idToRSocketMap;
-      private UnicastProcessor<Task> tasksToProcess;
+      private Sinks.Many<Task> tasksToProcess;
 
       public RSocketTaskHandler(
           ConcurrentMap<String, RSocket> idToRSocketMap,
-          UnicastProcessor<Task> tasksToProcess,
+          Sinks.Many<Task> tasksToProcess,
           String id,
           RSocket sendingSocket) {
         this.id = id;
@@ -213,9 +213,11 @@ public RSocketTaskHandler(
       @Override
       public Mono<Void> fireAndForget(Payload payload) {
         logger.info("Received a Task[{}] from Client.ID[{}]", payload.getDataUtf8(), id);
-        tasksToProcess.onNext(new Task(id, payload.getDataUtf8()));
+        Sinks.Emission emission = tasksToProcess.tryEmitNext(new Task(id, payload.getDataUtf8()));
         payload.release();
-        return Mono.empty();
+        return emission.hasFailed()
+            ? Mono.error(new Sinks.EmissionException(emission))
+            : Mono.empty();
       }
 
       @Override
