Set back directProcessor in CompositeDiscovery, ScaldecubeServiceDsicovery

Author: artem-v

services-discovery/src/main/java/io/scalecube/services/discovery/ScalecubeServiceDiscovery.java

@@ -33,11 +33,12 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.Exceptions;
+import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Operators;
 import reactor.core.publisher.SignalType;
-import reactor.core.publisher.Sinks;
 import reactor.core.publisher.Sinks.EmitFailureHandler;
 import reactor.core.publisher.Sinks.EmitResult;
 
@@ -55,8 +56,8 @@ public final class ScalecubeServiceDiscovery implements ServiceDiscovery {
   private Cluster cluster;
 
   // Sink
-  private final Sinks.Many<ServiceDiscoveryEvent> sink =
-      Sinks.many().multicast().directBestEffort();
+  private final DirectProcessor<ServiceDiscoveryEvent> subject = DirectProcessor.create();
+  private final FluxSink<ServiceDiscoveryEvent> sink = subject.sink();
 
   /**
    * Constructor.
@@ -170,21 +171,19 @@ public void onMembershipEvent(MembershipEvent event) {
 
   @Override
   public Flux<ServiceDiscoveryEvent> listen() {
-    return sink.asFlux().onBackpressureBuffer();
+    return subject.onBackpressureBuffer();
   }
 
   @Override
   public Mono<Void> shutdown() {
     return Mono.defer(
         () -> {
           if (cluster == null) {
-            sink.emitComplete(RetryEmitFailureHandler.INSTANCE);
+            sink.complete();
             return Mono.empty();
           }
           cluster.shutdown();
-          return cluster
-              .onShutdown()
-              .doFinally(s -> sink.emitComplete(RetryEmitFailureHandler.INSTANCE));
+          return cluster.onShutdown().doFinally(s -> sink.complete());
         });
   }
 
@@ -201,7 +200,7 @@ private void onMembershipEvent(MembershipEvent membershipEvent) {
 
     if (discoveryEvent != null) {
       LOGGER.debug("Publish discoveryEvent: {}", discoveryEvent);
-      sink.emitNext(discoveryEvent, RetryEmitFailureHandler.INSTANCE);
+      sink.next(discoveryEvent);
     }
   }