Merge pull request #813 from scalecube/update-dependencies-and-add-failurehandler

Fixed Sinks, added RetryEmitFailureHandler

Author: artem-v

pom.xml

@@ -57,7 +57,7 @@
   </scm>
 
   <properties>
-    <scalecube-cluster.version>2.6.7</scalecube-cluster.version>
+    <scalecube-cluster.version>2.6.8</scalecube-cluster.version>
     <scalecube-commons.version>1.0.13</scalecube-commons.version>
     <scalecube-security.version>1.0.19</scalecube-security.version>
 

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

@@ -3,6 +3,7 @@
 import static io.scalecube.services.discovery.api.ServiceDiscoveryEvent.newEndpointAdded;
 import static io.scalecube.services.discovery.api.ServiceDiscoveryEvent.newEndpointLeaving;
 import static io.scalecube.services.discovery.api.ServiceDiscoveryEvent.newEndpointRemoved;
+import static reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED;
 
 import io.scalecube.cluster.Cluster;
 import io.scalecube.cluster.ClusterConfig;
@@ -34,7 +35,10 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+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 {
 
@@ -169,11 +173,13 @@ public Mono<Void> shutdown() {
     return Mono.defer(
         () -> {
           if (cluster == null) {
-            sink.tryEmitComplete();
+            sink.emitComplete(RetryEmitFailureHandler.INSTANCE);
             return Mono.empty();
           }
           cluster.shutdown();
-          return cluster.onShutdown().doFinally(s -> sink.tryEmitComplete());
+          return cluster
+              .onShutdown()
+              .doFinally(s -> sink.emitComplete(RetryEmitFailureHandler.INSTANCE));
         });
   }
 
@@ -190,7 +196,7 @@ private void onMembershipEvent(MembershipEvent membershipEvent) {
 
     if (discoveryEvent != null) {
       LOGGER.debug("Publish discoveryEvent: {}", discoveryEvent);
-      sink.tryEmitNext(discoveryEvent);
+      sink.emitNext(discoveryEvent, RetryEmitFailureHandler.INSTANCE);
     }
   }
 
@@ -283,4 +289,14 @@ private void onDiscoveryEvent(ServiceDiscoveryEvent event) {
       }
     }
   }
+
+  private static class RetryEmitFailureHandler implements EmitFailureHandler {
+
+    private static final RetryEmitFailureHandler INSTANCE = new RetryEmitFailureHandler();
+
+    @Override
+    public boolean onEmitFailure(SignalType signalType, EmitResult emitResult) {
+      return emitResult == FAIL_NON_SERIALIZED;
+    }
+  }
 }

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

@@ -1,5 +1,7 @@
 package io.scalecube.services;
 
+import static reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED;
+
 import io.scalecube.net.Address;
 import io.scalecube.services.api.ServiceMessage;
 import io.scalecube.services.auth.Authenticator;
@@ -57,7 +59,10 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
 import reactor.core.publisher.Sinks;
+import reactor.core.publisher.Sinks.EmitFailureHandler;
+import reactor.core.publisher.Sinks.EmitResult;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
@@ -167,7 +172,7 @@ private Microservices(Builder builder) {
     shutdown
         .asMono()
         .then(doShutdown())
-        .doFinally(s -> onShutdown.tryEmitEmpty())
+        .doFinally(s -> onShutdown.emitEmpty(RetryEmitFailureHandler.INSTANCE))
         .subscribe(
             null, ex -> LOGGER.warn("[{}][doShutdown] Exception occurred: {}", id, ex.toString()));
   }
@@ -344,7 +349,11 @@ public Flux<ServiceDiscoveryEvent> listenDiscovery() {
    * @return result of shutdown
    */
   public Mono<Void> shutdown() {
-    return Mono.fromRunnable(shutdown::tryEmitEmpty).then(onShutdown.asMono());
+    return Mono.defer(
+        () -> {
+          shutdown.emitEmpty(RetryEmitFailureHandler.INSTANCE);
+          return onShutdown.asMono();
+        });
   }
 
   /**
@@ -900,4 +909,14 @@ private static String asString(ServiceInfo serviceInfo) {
           .toString();
     }
   }
+
+  private static class RetryEmitFailureHandler implements EmitFailureHandler {
+
+    private static final RetryEmitFailureHandler INSTANCE = new RetryEmitFailureHandler();
+
+    @Override
+    public boolean onEmitFailure(SignalType signalType, EmitResult emitResult) {
+      return emitResult == FAIL_NON_SERIALIZED;
+    }
+  }
 }