Merge pull request #53 from vinscom/upgrade-version

vinscom · web-flow · commit ca95e6c9fdb5 · 2018-12-25T15:06:00.000Z
Upgraded version
diff --git a/pom.xml b/pom.xml
@@ -96,7 +96,7 @@
         </profile>
     </profiles>
     <properties>
-        <vertx.version>3.5.4</vertx.version>
+        <vertx.version>3.6.2</vertx.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
@@ -166,7 +166,7 @@
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-transport-native-epoll</artifactId>
-            <version>4.1.19.Final</version>
+            <version>4.1.30.Final</version>
             <classifier>linux-x86_64</classifier>
         </dependency>
         <dependency>
@@ -182,7 +182,7 @@
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
-            <version>2.22.0</version>
+            <version>2.23.4</version>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -194,7 +194,7 @@
         <dependency>
             <groupId>org.hamcrest</groupId>
             <artifactId>hamcrest-core</artifactId>
-            <version>1.3</version>
+            <version>2.1</version>
             <scope>test</scope>
         </dependency>
     </dependencies>
@@ -203,7 +203,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
-                <version>2.22.0</version>
+                <version>2.22.1</version>
                 <configuration>
                     <argLine>-Dhazelcast.jmx=true -Dvertx.logger-delegate-factory-class-name=io.vertx.core.logging.Log4j2LogDelegateFactory -Dglue.layers=${project.basedir}/config-layers/common,${project.basedir}/config-layers/test</argLine>
                 </configuration>
@@ -229,7 +229,7 @@
             <plugin>
                 <groupId>org.jacoco</groupId>
                 <artifactId>jacoco-maven-plugin</artifactId>
-                <version>0.8.1</version>
+                <version>0.8.2</version>
                 <executions>
                     <execution>
                         <id>prepare-agent</id>
diff --git a/src/main/java/in/erail/model/ResponseEvent.java b/src/main/java/in/erail/model/ResponseEvent.java
@@ -24,7 +24,6 @@ public class ResponseEvent {
 
   public ResponseEvent() {
     mMultiValueHeaders = MultiMap.caseInsensitiveMultiMap();
-    mMultiValueHeaders.add(HttpHeaders.CONTENT_TYPE, MediaType.JAVASCRIPT_UTF_8.toString());
   }
 
   public boolean isIsBase64Encoded() {
@@ -83,6 +82,10 @@ public void setMultiValueHeaders(Map<String, String[]> pValue) {
                     });
   }
 
+  /**
+   * Return copy of headers map
+   * @return 
+   */
   public Map<String, String[]> getMultiValueHeaders() {
 
     Map<String, String[]> result
@@ -122,6 +125,10 @@ public void setHeaders(Map<String, String> pValue) {
                     });
   }
 
+  /**
+   * Return copy of Header Map
+   * @return 
+   */
   public Map<String, String> getHeaders() {
 
     Map<String, String> result
@@ -157,4 +164,8 @@ public void addHeader(String pHeaderName, String pMediaType) {
   public void addHeader(String pHeaderName, MediaType pMediaType) {
     addHeader(HttpHeaders.CONTENT_TYPE, pMediaType.toString());
   }
+
+  public String headerValue(String pHeaderName) {
+    return mMultiValueHeaders.get(pHeaderName);
+  }
 }
diff --git a/src/main/java/in/erail/route/OpenAPI3RouteBuilder.java b/src/main/java/in/erail/route/OpenAPI3RouteBuilder.java
@@ -90,7 +90,8 @@ public void process(RoutingContext pRequestContext, String pServiceUniqueId) {
                     (reply) -> {
                       if (reply.succeeded()) {
                         JsonObject response = (JsonObject) reply.result().body();
-                        buildResponseFromReply(response, pRequestContext).end();
+                        HttpServerResponse resp = buildResponseFromReply(response, pRequestContext);
+                        resp.end();
                       } else {
                         ((Meter) getMetrics().get(pServiceUniqueId + FAIL_SUFFIX)).mark();
                         getLog().error(() -> "Error in reply:" + reply.cause().toString());
@@ -144,8 +145,10 @@ public HttpServerResponse buildResponseFromReply(JsonObject pReplyResponse, Rout
 
     ResponseEvent response = pReplyResponse.mapTo(ResponseEvent.class);
 
-    if (!response.getHeaders().containsKey(HttpHeaders.CONTENT_TYPE)) {
-      response.getHeaders().put(HttpHeaders.CONTENT_TYPE, MediaType.OCTET_STREAM.toString());
+    Optional<String> contentType = Optional.ofNullable(response.headerValue(HttpHeaders.CONTENT_TYPE));
+
+    if (contentType.isPresent()) {
+      response.addHeader(HttpHeaders.CONTENT_TYPE, MediaType.OCTET_STREAM);
     }
 
     response
@@ -159,7 +162,7 @@ public HttpServerResponse buildResponseFromReply(JsonObject pReplyResponse, Rout
     pContext.response().setStatusCode(response.getStatusCode());
 
     @SuppressWarnings("unchecked")
-		Map<String,String>[] cookies = Optional.ofNullable(response.getCookies()).orElse(new Map[0]);
+    Map<String, String>[] cookies = Optional.ofNullable(response.getCookies()).orElse(new Map[0]);
 
     Arrays
             .stream(cookies)
@@ -184,9 +187,8 @@ public HttpServerResponse buildResponseFromReply(JsonObject pReplyResponse, Rout
 
     body.ifPresent((t) -> {
       pContext.response().putHeader(HttpHeaderNames.CONTENT_LENGTH.toString(), Integer.toString(t.length));
-      pContext.response().write(Buffer.newInstance(io.vertx.core.buffer.Buffer.buffer(t)));
+      pContext.response().write(Buffer.buffer(t));
     });
-
     return pContext.response();
   }
 
@@ -210,7 +212,7 @@ public void setDeliveryOptions(DeliveryOptions pDeliveryOptions) {
   public Router getRouter(Router pRouter) {
 
     OpenAPI3RouterFactory apiFactory = OpenAPI3RouterFactory
-            .rxCreateRouterFactoryFromFile(getVertx(), getOpenAPI3File().getAbsolutePath())
+            .rxCreate(getVertx(), getOpenAPI3File().getAbsolutePath())
             .blockingGet();
 
     Optional
diff --git a/src/main/java/in/erail/security/SecurityTools.java b/src/main/java/in/erail/security/SecurityTools.java
@@ -7,16 +7,12 @@
 import in.erail.glue.annotation.StartService;
 import io.reactivex.Single;
 import io.vertx.reactivex.core.Vertx;
-import io.vertx.reactivex.core.shareddata.AsyncMap;
-import io.vertx.reactivex.core.shareddata.Lock;
 import java.security.InvalidAlgorithmParameterException;
 import java.security.InvalidKeyException;
 import java.security.NoSuchAlgorithmException;
 import java.security.SecureRandom;
 import java.util.Arrays;
 import java.util.Base64;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import javax.crypto.BadPaddingException;
@@ -47,63 +43,8 @@ public void startup() {
       return;
     }
 
-    Map<String, Object> cryptCtx = new HashMap<>();
-
-    Single<Lock> lock = getVertx()
-            .sharedData()
-            .rxGetLockWithTimeout("_in.erail.security", 5000);
-
-    getVertx()
-            .sharedData()
-            .<String, byte[]>rxGetClusterWideMap("_in.erail.security")
-            .flatMap((m) -> {
-              cryptCtx.put("map", m);
-              return m.rxGet("key");
-            })
-            .map((k) -> {
-              cryptCtx.put("key", k);
-              return cryptCtx;
-            })
-            .flatMap((ctx) -> {
-              if (ctx.get("key") == null) {
-                return lock
-                        .map((l) -> {
-                          ctx.put("lock", l);
-                          return ctx;
-                        });
-              }
-              return Single.just(ctx);
-            })
-            .flatMap(ctx -> {
-              if (ctx.get("lock") != null) {
-                return ((AsyncMap<String, Object>) (ctx.get("map")))
-                        .rxGet("key")
-                        .map((k) -> {
-                          ctx.put("key", k);
-                          return ctx;
-                        });
-              }
-              return Single.just(ctx);
-            })
-            .flatMap((ctx) -> {
-              if (ctx.get("key") == null) {
-                KeyGenerator keygen = KeyGenerator.getInstance("AES");
-                keygen.init(128);
-                byte[] key = keygen.generateKey().getEncoded();
-                return ((AsyncMap<String, Object>) (ctx.get("map")))
-                        .rxPut("key", key)
-                        .doOnComplete(() -> ctx.put("key", key))
-                        .toSingleDefault(ctx);
-              }
-              return Single.just(ctx);
-            })
-            .map(ctx -> (byte[]) ctx.get("key"))
-            .doFinally(() -> {
-              if (cryptCtx.containsKey("lock")) {
-                Lock l = (Lock) cryptCtx.get("lock");
-                l.release();
-              }
-            })
+    generateKey()
+            .flatMap(v -> addValueToClusterMap("key", v))
             .subscribe((key) -> {
               mKeySpec.complete(new SecretKeySpec(key, "AES"));
               String unique = Base64.getEncoder().encodeToString(Arrays.copyOfRange(key, 0, 5));
@@ -112,6 +53,25 @@ public void startup() {
             });
   }
 
+  protected Single<byte[]> addValueToClusterMap(String pKey, byte[] pValue) {
+    return getVertx()
+            .sharedData()
+            .<String, byte[]>rxGetClusterWideMap("_in.erail.security")
+            .flatMapMaybe(m -> m.rxPutIfAbsent(pKey, pValue))
+            .toSingle(pValue);
+  }
+
+  protected Single<byte[]> generateKey() {
+    KeyGenerator keygen;
+    try {
+      keygen = KeyGenerator.getInstance("AES");
+    } catch (NoSuchAlgorithmException ex) {
+      return Single.error(ex);
+    }
+    keygen.init(128);
+    return Single.just(keygen.generateKey().getEncoded());
+  }
+
   /**
    * Unique string across cluster. Changes on each restart of cluster.
    *
diff --git a/src/main/java/in/erail/service/SingletonServiceImpl.java b/src/main/java/in/erail/service/SingletonServiceImpl.java
@@ -2,6 +2,7 @@
 
 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;
@@ -28,17 +29,8 @@ public void start() {
       return;
     }
 
-    getVertx()
-            .sharedData()
-            .<String, String>rxGetClusterWideMap(getServiceMapName())
+    allowServiceToStart()
             .subscribeOn(Schedulers.io())
-            .flatMap((m) -> m.rxPutIfAbsent(getServiceName(), getClusterManager().getNodeID()))
-            .map((ownerNodeId) -> {
-              if (ownerNodeId == null) {
-                return true;
-              }
-              return getClusterManager().getNodeID().equals(ownerNodeId);
-            })
             .flatMapCompletable((success) -> {
               if (success) {
                 getLog().info(String.format("Starting Service:[%s]", getServiceName()));
@@ -50,6 +42,20 @@ public void start() {
             .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
   public void nodeAdded(String pNodeID) {
   }
diff --git a/src/main/java/io/vertx/reactivex/ext/web/handler/sockjs/processor/SetSubscriberCountHeaderProcessor.java b/src/main/java/io/vertx/reactivex/ext/web/handler/sockjs/processor/SetSubscriberCountHeaderProcessor.java
@@ -54,8 +54,8 @@ public Single<BridgeEventContext> process(Single<BridgeEventContext> pContext) {
                       .map((count) -> setHeader(count, ctx))
                       .doOnError((err) -> {
                         getLog().error(String.format("[%s] Error getting value for Key[%s] from redis", ctx.getId(), ctx.getAddressKey()), err);
-                      });
-
+                      })
+                      .toSingle(ctx);
             });
   }
 
diff --git a/src/test/java/in/erail/service/SingletonServiceImplTest.java b/src/test/java/in/erail/service/SingletonServiceImplTest.java
@@ -44,7 +44,7 @@ public void testStart(TestContext context) throws InterruptedException, Executio
                         vertx
                                 .sharedData()
                                 .<String, String>rxGetClusterWideMap("__in.erail.services")
-                                .flatMap((m) -> {
+                                .flatMapMaybe((m) -> {
                                   return m.rxGet("DummySingletonService");
                                 })
                                 .doOnSuccess((nodeId) -> {
@@ -59,7 +59,7 @@ public void testStart(TestContext context) throws InterruptedException, Executio
                                             vertx
                                                     .sharedData()
                                                     .<String, String>rxGetClusterWideMap("__in.erail.services")
-                                                    .flatMap(m2 -> {
+                                                    .flatMapMaybe(m2 -> {
                                                       return m2.rxGet("DummySingletonService");
                                                     })
                                                     .subscribe((updatedNodeId) -> {
