Set back directProcessor in CompositeDiscovery, ScaldecubeServiceDsicovery

Author: artem-v

services/src/main/java/io/scalecube/services/Microservices.java

@@ -57,7 +57,9 @@
 import reactor.core.Disposable;
 import reactor.core.Disposables;
 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.SignalType;
 import reactor.core.publisher.Sinks;
@@ -555,9 +557,9 @@ private static class CompositeServiceDiscovery implements ServiceDiscovery {
     private final Map<String, ServiceDiscoveryContext> discoveryContexts =
         new ConcurrentHashMap<>();
 
-    // Sink
-    private final Sinks.Many<ServiceDiscoveryEvent> sink =
-        Sinks.many().multicast().directBestEffort();
+    // Subject
+    private final DirectProcessor<ServiceDiscoveryEvent> subject = DirectProcessor.create();
+    private final FluxSink<ServiceDiscoveryEvent> sink = subject.sink();
 
     private final Disposable.Composite disposables = Disposables.composite();
     private Scheduler scheduler;
@@ -614,7 +616,7 @@ private Mono<Void> startListen() {
     public Flux<ServiceDiscoveryEvent> listen() {
       return Flux.fromStream(microservices.serviceRegistry.listServiceEndpoints().stream())
           .map(ServiceDiscoveryEvent::newEndpointAdded)
-          .concatWith(sink.asFlux().onBackpressureBuffer())
+          .concatWith(subject)
           .subscribeOn(scheduler)
           .publishOn(scheduler);
     }
@@ -650,7 +652,7 @@ private Mono<? extends Void> start0(String id, ServiceDiscovery discovery) {
               .subscribeOn(scheduler)
               .publishOn(scheduler)
               .doOnNext(event -> onDiscoveryEvent(microservices, event))
-              .doOnNext(event -> sink.emitNext(event, RetryEmitFailureHandler.INSTANCE))
+              .doOnNext(sink::next)
               .subscribe());
 
       return Mono.deferContextual(context -> discovery.start())