Added Closeable

Author: artem-v

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

@@ -129,6 +129,7 @@ public Mono<Void> start() {
               .config(options -> clusterConfig)
               .handler(
                   cluster -> {
+                    //noinspection CodeBlock2Expr
                     return new ClusterMessageHandler() {
                       @Override
                       public void onMembershipEvent(MembershipEvent event) {
@@ -176,10 +177,8 @@ private void onMembershipEvent(MembershipEvent membershipEvent) {
       return;
     }
 
-    if (discoveryEvent != null) {
-      LOGGER.debug("Publish discoveryEvent: {}", discoveryEvent);
-      sink.emitNext(discoveryEvent, RETRY_NON_SERIALIZED);
-    }
+    LOGGER.debug("Publish discoveryEvent: {}", discoveryEvent);
+    sink.emitNext(discoveryEvent, RETRY_NON_SERIALIZED);
   }
 
   private ServiceDiscoveryEvent toServiceDiscoveryEvent(MembershipEvent membershipEvent) {

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

@@ -126,7 +126,7 @@
  *
  * }</pre>
  */
-public final class Microservices {
+public final class Microservices implements AutoCloseable {
 
   public static final Logger LOGGER = LoggerFactory.getLogger(Microservices.class);
 
@@ -190,12 +190,8 @@ public String toString() {
   private Mono<Microservices> start() {
     LOGGER.info("[{}][start] Starting", id);
 
-    // Create bootstrap scheduler
-    Scheduler scheduler = Schedulers.newSingle(toString(), true);
-
     return transportBootstrap
         .start(this)
-        .publishOn(scheduler)
         .flatMap(
             transportBootstrap -> {
               final ServiceCall call = call();
@@ -229,19 +225,15 @@ private Mono<Microservices> start() {
 
               return createDiscovery(
                       this, new ServiceDiscoveryOptions().serviceEndpoint(serviceEndpoint))
-                  .publishOn(scheduler)
                   .then(startGateway(new GatewayOptions().call(call)))
-                  .publishOn(scheduler)
                   .then(Mono.fromCallable(() -> Injector.inject(this, serviceInstances)))
                   .then(Mono.fromCallable(() -> JmxMonitorMBean.start(this)))
                   .then(compositeDiscovery.startListen())
-                  .publishOn(scheduler)
                   .thenReturn(this);
             })
         .onErrorResume(
             ex -> Mono.defer(this::shutdown).then(Mono.error(ex)).cast(Microservices.class))
-        .doOnSuccess(m -> LOGGER.info("[{}][start] Started", id))
-        .doOnTerminate(scheduler::dispose);
+        .doOnSuccess(m -> LOGGER.info("[{}][start] Started", id));
   }
 
   private ServiceEndpoint newServiceEndpoint(ServiceEndpoint serviceEndpoint) {
@@ -312,6 +304,18 @@ public List<ServiceEndpoint> serviceEndpoints() {
     return serviceRegistry.listServiceEndpoints();
   }
 
+  public Map<String, String> tags() {
+    return tags;
+  }
+
+  public ServiceRegistry serviceRegistry() {
+    return serviceRegistry;
+  }
+
+  public ServiceMethodRegistry methodRegistry() {
+    return methodRegistry;
+  }
+
   /**
    * Returns default service discovery context.
    *
@@ -393,6 +397,15 @@ private Mono<Void> processBeforeDestroy() {
             .collect(Collectors.toList()));
   }
 
+  @Override
+  public void close() {
+    try {
+      shutdown().toFuture().get();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   public static final class Builder {
 
     private Map<String, String> tags = new HashMap<>();