Fixing CI

Author: artem-v

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

@@ -33,31 +33,25 @@
 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;
 
 public final class ScalecubeServiceDiscovery implements ServiceDiscovery {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(ServiceDiscovery.class);
 
-  static {
-    Operators.enableOnDiscard(null, o -> LOGGER.warn("[onDiscard] element = {}", o));
-  }
-
   private final ServiceEndpoint serviceEndpoint;
 
   private ClusterConfig clusterConfig;
   private Cluster cluster;
 
   // Sink
-  private final DirectProcessor<ServiceDiscoveryEvent> subject = DirectProcessor.create();
-  private final FluxSink<ServiceDiscoveryEvent> sink = subject.sink();
+  private final Sinks.Many<ServiceDiscoveryEvent> sink =
+      Sinks.many().multicast().directBestEffort();
 
   /**
    * Constructor.
@@ -171,19 +165,21 @@ public void onMembershipEvent(MembershipEvent event) {
 
   @Override
   public Flux<ServiceDiscoveryEvent> listen() {
-    return subject.onBackpressureBuffer();
+    return sink.asFlux().onBackpressureBuffer();
   }
 
   @Override
   public Mono<Void> shutdown() {
     return Mono.defer(
         () -> {
           if (cluster == null) {
-            sink.complete();
+            sink.emitComplete(RetryEmitFailureHandler.INSTANCE);
             return Mono.empty();
           }
           cluster.shutdown();
-          return cluster.onShutdown().doFinally(s -> sink.complete());
+          return cluster
+              .onShutdown()
+              .doFinally(s -> sink.emitComplete(RetryEmitFailureHandler.INSTANCE));
         });
   }
 
@@ -200,7 +196,7 @@ private void onMembershipEvent(MembershipEvent membershipEvent) {
 
     if (discoveryEvent != null) {
       LOGGER.debug("Publish discoveryEvent: {}", discoveryEvent);
-      sink.next(discoveryEvent);
+      sink.emitNext(discoveryEvent, RetryEmitFailureHandler.INSTANCE);
     }
   }
 
@@ -300,7 +296,6 @@ private static class RetryEmitFailureHandler implements EmitFailureHandler {
 
     @Override
     public boolean onEmitFailure(SignalType signalType, EmitResult emitResult) {
-      LOGGER.warn("[onEmitFailure] signalType={}, emitResult={}", signalType, emitResult);
       return emitResult == FAIL_NON_SERIALIZED;
     }
   }

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

@@ -57,9 +57,7 @@
 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;
@@ -558,8 +556,8 @@ private static class CompositeServiceDiscovery implements ServiceDiscovery {
         new ConcurrentHashMap<>();
 
     // Subject
-    private final DirectProcessor<ServiceDiscoveryEvent> subject = DirectProcessor.create();
-    private final FluxSink<ServiceDiscoveryEvent> sink = subject.sink();
+    private final Sinks.Many<ServiceDiscoveryEvent> sink =
+        Sinks.many().multicast().directBestEffort();
 
     private final Disposable.Composite disposables = Disposables.composite();
     private Scheduler scheduler;
@@ -616,7 +614,7 @@ private Mono<Void> startListen() {
     public Flux<ServiceDiscoveryEvent> listen() {
       return Flux.fromStream(microservices.serviceRegistry.listServiceEndpoints().stream())
           .map(ServiceDiscoveryEvent::newEndpointAdded)
-          .concatWith(subject)
+          .concatWith(sink.asFlux().onBackpressureBuffer())
           .subscribeOn(scheduler)
           .publishOn(scheduler);
     }
@@ -652,7 +650,7 @@ private Mono<? extends Void> start0(String id, ServiceDiscovery discovery) {
               .subscribeOn(scheduler)
               .publishOn(scheduler)
               .doOnNext(event -> onDiscoveryEvent(microservices, event))
-              .doOnNext(sink::next)
+              .doOnNext(event -> sink.emitNext(event, RetryEmitFailureHandler.INSTANCE))
               .subscribe());
 
       return Mono.deferContextual(context -> discovery.start())
@@ -676,6 +674,7 @@ public Mono<Void> shutdown() {
       return Mono.defer(
           () -> {
             disposables.dispose();
+            sink.emitComplete(RetryEmitFailureHandler.INSTANCE);
             return Mono.whenDelayError(
                     discoveryInstances.values().stream()
                         .map(ServiceDiscovery::shutdown)