Using Vertx thread for RxJava

Author: vinscom

src/main/java/in/erail/service/RESTServiceImpl.java

@@ -50,7 +50,7 @@ public void start() throws InstantiationException, IllegalAccessException {
               .eventBus()
               .<JsonObject>consumer(getServiceUniqueId())
               .toFlowable()
-              .subscribeOn(getScheduler())
+              .observeOn(getScheduler())
               .flatMapSingle(this::handleRequest)
               .doOnSubscribe((s) -> getLog().info(() -> String.format("%s[%s] service started", getServiceUniqueId(), Thread.currentThread().getName())))
               .doOnTerminate(() -> getLog().info(() -> String.format("%s[%s] service stopped", getServiceUniqueId(), Thread.currentThread().getName())))

src/main/java/in/erail/service/SingletonServiceImpl.java

@@ -3,10 +3,10 @@
 import in.erail.glue.annotation.StartService;
 import io.reactivex.Completable;
 import io.reactivex.Single;
-import io.reactivex.schedulers.Schedulers;
 import io.vertx.core.spi.cluster.ClusterManager;
 import io.vertx.core.spi.cluster.NodeListener;
 import io.vertx.reactivex.core.Vertx;
+import java.util.Optional;
 import org.apache.logging.log4j.Logger;
 
 /**
@@ -29,31 +29,41 @@ public void start() {
       return;
     }
 
-    allowServiceToStart()
-            .subscribeOn(Schedulers.io())
+    Single
+            .just(Optional.<String>empty())
+            .flatMapCompletable(this::init)
+            .subscribe();
+  }
+
+  protected Completable init(Optional<String> pOldNodeID) {
+    final String serviceName = getServiceName();
+    final String thisNodeId = getClusterManager().getNodeID();
+
+    return Single
+            .just(getServiceMapName())
+            .flatMap(name -> getVertx().sharedData().<String, String>rxGetClusterWideMap(name))
+            .flatMap(m -> {
+              if (pOldNodeID.isPresent()) {
+                return m
+                        .rxReplaceIfPresent(getServiceName(), pOldNodeID.get(), thisNodeId)
+                        .map(v -> v ? thisNodeId : "");
+              }
+              return m
+                      .rxPutIfAbsent(serviceName, thisNodeId)
+                      .switchIfEmpty(Single.just(thisNodeId));
+            })
+            .doOnSuccess(serviceOwnerId -> getLog().debug(() -> "Service Owner ID:" + serviceOwnerId + ", This Node ID:" + thisNodeId))
+            .map(serviceOwnerId -> thisNodeId.equals(serviceOwnerId))
+            .doOnSuccess(t -> getLog().debug(() -> "Service Start Decision:" + getServiceName() + ":" + t))
             .flatMapCompletable((success) -> {
               if (success) {
                 getLog().info(String.format("Starting Service:[%s]", getServiceName()));
                 return startService()
                         .doOnComplete(() -> getLog().info(String.format("Service:[%s] started", getServiceName())));
               }
               return Completable.complete();
-            })
-            .blockingAwait();
-  }
+            });
 
-  protected Single<Boolean> allowServiceToStart() {
-    final String serviceName = getServiceName();
-    final String value = getClusterManager().getNodeID();
-
-    return getVertx()
-            .sharedData()
-            .<String, String>rxGetClusterWideMap(getServiceMapName())
-            .flatMapMaybe(m -> m.rxPutIfAbsent(serviceName, value))
-            .toSingle(value)
-            .doOnSuccess(serviceOwnerId -> getLog().debug(() -> "Service Owner ID:" + serviceOwnerId + ", This Node ID:" + value))
-            .map(serviceOwnerId -> getClusterManager().getNodeID().equals(serviceOwnerId))
-            .doOnSuccess(t -> getLog().debug(() -> "Service Start Decision:" + getServiceName() + ":" + t));
   }
 
   @Override
@@ -67,20 +77,10 @@ public void nodeLeft(String pNodeID) {
       return;
     }
 
-    getVertx()
-            .sharedData()
-            .<String, String>rxGetClusterWideMap(getServiceMapName())
-            .subscribeOn(Schedulers.io())
-            .flatMap((m) -> m.rxReplaceIfPresent(getServiceName(), pNodeID, getClusterManager().getNodeID()))
-            .flatMapCompletable((success) -> {
-              if (success) {
-                getLog().info(String.format("Starting Service:[%s] becuase of cluster state update", getServiceName()));
-                return startService()
-                        .doOnComplete(() -> getLog().info(String.format("Service:[%s] start complete because of cluster state update", getServiceName())));
-              }
-              return Completable.complete();
-            })
-            .blockingAwait();
+    Single
+            .just(Optional.of(pNodeID))
+            .flatMapCompletable(this::init)
+            .subscribe();
   }
 
   public Vertx getVertx() {

src/main/java/io/vertx/core/VertxInstance.java

@@ -4,12 +4,14 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import in.erail.glue.annotation.StartService;
+import io.reactivex.plugins.RxJavaPlugins;
+import io.vertx.reactivex.core.RxHelper;
 
 public class VertxInstance {
 
   private VertxOptions mVertxOptions;
   private boolean mClusterEnable = true;
-  private CompletableFuture<Vertx> mVertx = new CompletableFuture<>();
+  private final CompletableFuture<Vertx> mVertx = new CompletableFuture<>();
 
   @StartService
   public void start() {
@@ -22,7 +24,11 @@ public void start() {
   }
 
   public Vertx create() throws InterruptedException, ExecutionException {
-      return mVertx.get();
+    Vertx v = mVertx.get();
+    RxJavaPlugins.setComputationSchedulerHandler(s -> RxHelper.scheduler(v));
+    RxJavaPlugins.setIoSchedulerHandler(s -> RxHelper.blockingScheduler(v));
+    RxJavaPlugins.setNewThreadSchedulerHandler(s -> RxHelper.scheduler(v));
+    return v;
   }
 
   public VertxOptions getVertxOptions() {