Merge pull request #55 from vinscom/develop

Minor refactoring

Author: vinscom

src/main/java/in/erail/security/SecurityTools.java

@@ -32,7 +32,6 @@ public class SecurityTools {
   private final CompletableFuture<SecretKeySpec> mKeySpec = new CompletableFuture<>();
   private final CompletableFuture<String> mGlobalUniqueString = new CompletableFuture<>();
 
-  @SuppressWarnings("unchecked")
   @StartService
   public void startup() {
 

src/main/java/in/erail/server/Server.java

@@ -48,7 +48,7 @@ public void start() {
     }
 
     mHttpServer = server
-            .requestHandler(router::accept)
+            .requestHandler(router)
             .rxListen()
             .blockingGet();
 

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

@@ -3,6 +3,7 @@
 
 import in.erail.model.RequestEvent;
 import in.erail.model.ResponseEvent;
+import io.reactivex.Maybe;
 
 /**
  *
@@ -11,5 +12,5 @@
 public interface RESTService {
   String getOperationId();
   String getServiceUniqueId();
-  ResponseEvent process(RequestEvent pRequest);
+  Maybe<ResponseEvent> process(RequestEvent pRequest);
 }

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

@@ -6,9 +6,11 @@
 import org.apache.logging.log4j.Logger;
 import in.erail.glue.annotation.StartService;
 import in.erail.model.RequestEvent;
+import in.erail.model.ResponseEvent;
 import io.reactivex.Scheduler;
 import io.reactivex.Single;
 import io.vertx.reactivex.core.eventbus.Message;
+import org.apache.commons.lang3.exception.ExceptionUtils;
 
 /**
  *
@@ -34,15 +36,19 @@ public void start() {
               .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())))
               .flatMapSingle(this::handleRequest)
-              .subscribe((resp) -> getLog().trace(() -> resp.toString()), err -> getLog().error(() -> String.format("Process exception:[%s],Error:[%s]", getServiceUniqueId(), err)));
+              .subscribe(
+                      resp -> getLog().trace(() -> resp.toString()), 
+                      err -> getLog().error(() -> String.format("Process exception:[%s],Error:[%s]", getServiceUniqueId(), ExceptionUtils.getStackTrace(err)))
+              );
     }
   }
 
   public Single<JsonObject> handleRequest(Message<JsonObject> pMessage) {
     return Single
             .just(pMessage)
             .map(m -> pMessage.body().mapTo(RequestEvent.class))
-            .map(req -> process(req))
+            .flatMapMaybe(req -> process(req))
+            .toSingle(new ResponseEvent())
             .map(resp -> JsonObject.mapFrom(resp))
             .doOnSuccess(resp -> pMessage.reply(resp));
   }

src/test/java/in/erail/route/CORSRouteBuilderTest.java

@@ -23,6 +23,7 @@ public class CORSRouteBuilderTest {
   @Rule
   public Timeout rule = Timeout.seconds(2000);
 
+  @SuppressWarnings("deprecation")
   @Test
   public void testProcess(TestContext context) {
 

src/test/java/in/erail/service/BinaryBodyService.java

@@ -8,9 +8,9 @@
 import in.erail.model.ResponseEvent;
 import in.erail.test.TestConstants;
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.reactivex.Maybe;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.json.JsonObject;
-import io.vertx.reactivex.core.eventbus.Message;
 
 /**
  *
@@ -19,12 +19,12 @@
 public class BinaryBodyService extends RESTServiceImpl {
 
   @Override
-  public ResponseEvent process(RequestEvent pRequest) {
+  public Maybe<ResponseEvent> process(RequestEvent pRequest) {
     String topicName = pRequest.getPathParameters().get(TestConstants.Service.Broadcast.APIMessage.PARAM_TOPIC_NAME);
 
     if (Strings.isNullOrEmpty(topicName)) {
-      return new ResponseEvent()
-              .setStatusCode(HttpResponseStatus.BAD_REQUEST.code());
+      return Maybe.just(new ResponseEvent()
+              .setStatusCode(HttpResponseStatus.BAD_REQUEST.code()));
     }
 
     ResponseEvent response = new ResponseEvent();
@@ -34,7 +34,7 @@ public ResponseEvent process(RequestEvent pRequest) {
 
     String bodyContent = jsonBody.getString("data");
     response.setBody(bodyContent.getBytes());
-    return response;
+    return Maybe.just(response);
   }
 
 }

src/test/java/in/erail/service/BinaryBodyServiceTest.java

@@ -25,6 +25,7 @@ public class BinaryBodyServiceTest {
   @Rule
   public Timeout rule = Timeout.seconds(2000);
 
+  @SuppressWarnings("deprecation")
   @Test
   public void testProcess(TestContext context) {
 
@@ -44,7 +45,7 @@ public void testProcess(TestContext context) {
             .putHeader(HttpHeaders.AUTHORIZATION, TestConstants.ACCESS_TOKEN)
             .handler(response -> {
               context.assertEquals(response.statusCode(), 200, response.statusMessage());
-              context.assertEquals(response.getHeader(HttpHeaders.ACCESS_CONTROL_ALLOW_ORIGIN),"*");
+              context.assertEquals(response.getHeader(HttpHeaders.ACCESS_CONTROL_ALLOW_ORIGIN), "*");
               context.assertTrue(MediaType.parse(response.getHeader(HttpHeaders.CONTENT_TYPE)).equals(MediaType.PLAIN_TEXT_UTF_8));
               response.bodyHandler((event) -> {
                 context.assertEquals(event.toString(), "testdata");

src/test/java/in/erail/service/BroadcastService.java

@@ -5,10 +5,10 @@
 import com.google.common.net.MediaType;
 import in.erail.test.TestConstants;
 import io.vertx.core.json.JsonObject;
-import io.vertx.reactivex.core.eventbus.Message;
 import in.erail.model.RequestEvent;
 import in.erail.model.ResponseEvent;
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.reactivex.Maybe;
 
 /**
  *
@@ -17,12 +17,12 @@
 public class BroadcastService extends RESTServiceImpl {
 
   @Override
-  public ResponseEvent process(RequestEvent pRequest) {
-    
+  public Maybe<ResponseEvent> process(RequestEvent pRequest) {
+
     String topicName = pRequest.getPathParameters().get(TestConstants.Service.Broadcast.APIMessage.PARAM_TOPIC_NAME);
 
     if (Strings.isNullOrEmpty(topicName)) {
-      return new ResponseEvent().setStatusCode(HttpResponseStatus.BAD_REQUEST.code());
+      return Maybe.just(new ResponseEvent().setStatusCode(HttpResponseStatus.BAD_REQUEST.code()));
     }
 
     JsonObject bodyJson = new JsonObject(pRequest.bodyAsString());
@@ -36,8 +36,8 @@ public ResponseEvent process(RequestEvent pRequest) {
     ResponseEvent response = new ResponseEvent();
     response.setBody(TestConstants.Service.Message.successMessage().toString().getBytes());
     response.addHeader(HttpHeaders.CONTENT_TYPE, MediaType.JSON_UTF_8);
-    
-    return response;
+
+    return Maybe.just(response);
   }
- 
+
 }

src/test/java/in/erail/service/BroadcastServiceTest.java

@@ -25,6 +25,7 @@ public class BroadcastServiceTest {
   @Rule
   public Timeout rule = Timeout.seconds(2000);
 
+  @SuppressWarnings("deprecation")
   @Test
   public void testProcess(TestContext context) {