Lambda working

Author: vinscom

src/main/java/in/erail/amazon/lambda/AWSLambda.java

@@ -2,6 +2,7 @@
 
 import com.amazonaws.services.lambda.runtime.Context;
 import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.io.ByteStreams;
 import in.erail.glue.Glue;
@@ -10,88 +11,99 @@
 import io.reactivex.schedulers.Schedulers;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.json.JsonObject;
-import io.vertx.reactivex.core.eventbus.Message;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.OutputStreamWriter;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-import static in.erail.common.FrameworkConstants.RoutingContext.Json.*;
+import in.erail.model.RequestEvent;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 /**
  *
  * @author vinay
  */
 public class AWSLambda implements RequestStreamHandler {
 
+  protected Logger log = LogManager.getLogger(AWSLambda.class.getCanonicalName());
   private static final String SERVICE_ENV = "SERVICE";
   private static final String SERVICE_SYS_PROP = "service";
   private final RESTService mService;
+  private final List<String> allowedFields = new ArrayList<>();
 
   public AWSLambda() {
+
+    allowedFields.add("isBase64Encoded");
+    allowedFields.add("statusCode");
+    allowedFields.add("headers");
+    allowedFields.add("multiValueHeaders");
+    allowedFields.add("body");
+
     String component = System.getenv(SERVICE_ENV);
     if (Strings.isNullOrEmpty(component)) {
       component = System.getProperty(SERVICE_SYS_PROP);
     }
+
     mService = Glue.instance().resolve(component);
   }
 
   @Override
   public void handleRequest(InputStream inputStream, OutputStream outputStream, Context context) throws IOException {
-
-    JsonObject requestJson = new JsonObject(Buffer.buffer(ByteStreams.toByteArray(inputStream)));
-    JsonObject responseJson = handleMessage(requestJson);
-
     try (OutputStreamWriter writer = new OutputStreamWriter(outputStream, "UTF-8")) {
-      writer.write(responseJson.toString());
+      JsonObject requestJson = new JsonObject(Buffer.buffer(ByteStreams.toByteArray(inputStream)));
+      log.debug(() -> requestJson.toString());
+      String resp = handleMessage(requestJson).blockingGet();
+      log.debug(() -> resp);
+      writer.write(resp);
     }
   }
 
-  public JsonObject handleMessage(JsonObject pJsonObject) {
-
-    JsonObject msg;
+  public Single<String> handleMessage(JsonObject pRequest) {
+    return Single
+            .just(pRequest)
+            .subscribeOn(Schedulers.computation())
+            .map(this::convertBodyToBase64)
+            .map(reqJson -> reqJson.mapTo(RequestEvent.class))
+            .map(req -> getService().process(req))
+            .map(resp -> JsonObject.mapFrom(resp))
+            .map(this::sanatizeResponse)
+            .map(respJson -> respJson.toString());
+  }
 
-    try {
-      CompletableFuture<JsonObject> result = new CompletableFuture<>();
-      MessageAWSLambda<JsonObject> message = new MessageAWSLambda<>(result, pJsonObject);
+  protected JsonObject sanatizeResponse(JsonObject pResp) {
+    Preconditions.checkNotNull(pResp);
 
-      Single
-              .just(message)
-              .map(m -> new Message<JsonObject>(message))
-              .subscribeOn(Schedulers.computation())
-              .subscribe(m -> getService().process(m));
+    Set<String> keys = new HashSet(pResp.fieldNames());
 
-      msg = result.get();
-    } catch (InterruptedException | ExecutionException ex) {
-      msg = new JsonObject();
-      Logger.getLogger(AWSLambda.class.getName()).log(Level.SEVERE, null, ex);
-    }
+    keys
+            .stream()
+            .filter(key -> !allowedFields.contains(key))
+            .forEach(key -> pResp.remove(key));
 
-    return transform(msg);
+    return pResp;
   }
 
-  public JsonObject transform(JsonObject pMsg) {
-
-    pMsg.put(IS_BASE64_ENCODED, Boolean.TRUE);
+  protected JsonObject convertBodyToBase64(JsonObject pRequest) {
 
-    if (!pMsg.containsKey(STATUS_CODE)) {
-      pMsg.put(STATUS_CODE, "200");
-    }
-
-    if (!pMsg.containsKey(HEADERS)) {
-      pMsg.put(HEADERS, new JsonObject());
-    }
+    Boolean isBase64Encoded
+            = Optional
+                    .ofNullable(pRequest.getBoolean("isBase64Encoded"))
+                    .orElse(Boolean.FALSE);
 
-    if (pMsg.containsKey(BODY)) {
-      pMsg.put(BODY, pMsg.getString(BODY));
-    } else {
-      pMsg.put(BODY, "");
+    if (isBase64Encoded == false && pRequest.containsKey("body")) {
+      Optional<String> body = Optional.ofNullable(pRequest.getString("body"));
+      body.ifPresent((t) -> {
+        pRequest.remove("body");
+        pRequest.put("body", body.get().getBytes());
+      });
     }
 
-    return pMsg;
+    return pRequest;
   }
 
   public RESTService getService() {