Added AWS response wrappers

2026-09-22 21:55:39 +00:00 · 2020-12-12 09:03:24 +01:00
parent 830f0cb9ba
commit 841c3e62e6
8 changed files with 216 additions and 35 deletions
@@ -1 +1 @@
-{"lastDeploymentHandler":"com.sap.sailing.ingestion.FixIngestionLambda","lastInvokeHandler":null,"handlerMetadata":{"com.amazonaws.lambda.demo.LambdaFunctionHandler":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"IngestionTest","awsIamRoleName":"axel-test-role-y52w2i9t","awsS3BucketName":"sapsailing-access-logs-eu-west-2","memory":0,"timeout":0},"invoke":null},"com.sap.sailing.ingestion.EndpointRegistration":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"EndpointRegistration","awsIamRoleName":"fixstorageendpoint-lambda-role","awsS3BucketName":"sapsailing-lambda-functions-bucket-eu-west-2","memory":0,"timeout":0},"invoke":null},"com.sap.sailing.ingestion.EndpointRegistrationLambda":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"EndpointRegistration","awsIamRoleName":"fixstorageendpoint-lambda-role","awsS3BucketName":"sapsailing-lambda-functions-bucket-eu-west-2","memory":0,"timeout":0},"invoke":null},"com.sap.sailing.ingestion.FixIngestionLambda":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"FixIngestion","awsIamRoleName":"fixstorageendpoint-lambda-role","awsS3BucketName":"sapsailing-lambda-functions-bucket-eu-west-2","memory":0,"timeout":0},"invoke":null}},"lastInvokedHandlerDeployed":false}
+{"lastDeploymentHandler":"com.sap.sailing.ingestion.PingLambda","lastInvokeHandler":null,"handlerMetadata":{"com.amazonaws.lambda.demo.LambdaFunctionHandler":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"IngestionTest","awsIamRoleName":"axel-test-role-y52w2i9t","awsS3BucketName":"sapsailing-access-logs-eu-west-2","memory":0,"timeout":0},"invoke":null},"com.sap.sailing.ingestion.EndpointRegistration":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"EndpointRegistration","awsIamRoleName":"fixstorageendpoint-lambda-role","awsS3BucketName":"sapsailing-lambda-functions-bucket-eu-west-2","memory":0,"timeout":0},"invoke":null},"com.sap.sailing.ingestion.EndpointRegistrationLambda":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"EndpointRegistration","awsIamRoleName":"fixstorageendpoint-lambda-role","awsS3BucketName":"sapsailing-lambda-functions-bucket-eu-west-2","memory":0,"timeout":0},"invoke":null},"com.sap.sailing.ingestion.FixIngestionLambda":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"Ping","awsIamRoleName":"fixstorageendpoint-lambda-role","awsS3BucketName":"sapsailing-lambda-functions-bucket-eu-west-2","memory":0,"timeout":0},"invoke":null},"com.sap.sailing.ingestion.PingLambda":{"deployment":{"regionId":"eu-west-2","awsLambdaFunctionName":"Ping","awsIamRoleName":"fixstorageendpoint-lambda-role","awsS3BucketName":"sapsailing-access-logs-eu-west-2","memory":0,"timeout":0},"invoke":null}},"lastInvokedHandlerDeployed":false}
@@ -7,17 +7,16 @@ import org.redisson.api.RMap;

 import com.amazonaws.services.lambda.runtime.Context;
 import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.google.gson.Gson;
+import com.sap.sailing.ingestion.dto.AWSIngestionWrapper;
+import com.sap.sailing.ingestion.dto.AWSResponseWrapper;
 import com.sap.sailing.ingestion.dto.EndpointDTO;

 /**
 * <p>
 * This lambda handles the registration of HTTP endpoints that need to receive GPS fixes for one or more device UUIDs.
 * Once registered, an endpoint will receive fixes that are being transmitted in almost real-time. The transmission will
- * have a timeout of 2 seconds in order to not unnecessarily block the lambda execution.
- * </p>
- * 
- * <p>
- * Registration needs to be refreshed every 24 hours. Registrations older than this timeframe will be dropped.
+ * have a timeout of 4 seconds in order to not unnecessarily block the lambda execution.
 * </p>
 * 
 * <p>
@@ -31,32 +30,41 @@ import com.sap.sailing.ingestion.dto.EndpointDTO;
 * 
 * <pre>
 * {
- *     endpointUuid: "UUID of sailing server"
- *     endpointCallbackUrl: "Callback URL including the hostname, in most cases this should be /v1/gps_fixes"
- *     deviceUuids: [
+ *     body: {
+ *      endpointUuid: "UUID of sailing server"
+ *      endpointCallbackUrl: "Callback URL including the hostname, in most cases this should be /v1/gps_fixes"
+ *      action: "register|unregister",
+ *      deviceUuids: [
 *         "List of one or more device UUIDs"
- *     ]
+ *      ]
+ *     }
 * }
 * </pre>
 *
 */
-public class EndpointRegistrationLambda implements RequestHandler<EndpointDTO, String> {
+public class EndpointRegistrationLambda implements RequestHandler<AWSIngestionWrapper<EndpointDTO>, String> {
    @SuppressWarnings("unchecked")
    @Override
-    public String handleRequest(EndpointDTO input, Context context) {
+    public String handleRequest(AWSIngestionWrapper<EndpointDTO> awsWrappedInput, Context context) {
+        final EndpointDTO input = awsWrappedInput.getBody();
        if (input != null && input.getDevicesUuid() != null && input.getDevicesUuid().size() > 0) {
            context.getLogger().log("Getting cache instance...");
-            RMap<String, List<EndpointDTO>> cacheMap = Utils.getCacheMap();
+            final RMap<String, List<EndpointDTO>> cacheMap = Utils.getCacheMap();
            context.getLogger().log("Got cache instance");
            for (final String deviceUuid : input.getDevicesUuid()) {
                final Object memObject = cacheMap.get(deviceUuid);
-                final List<EndpointDTO> endpoints = memObject == null ? new ArrayList<>() : (List<EndpointDTO>) memObject;
-                endpoints.add(input);
+                final List<EndpointDTO> endpoints = memObject == null ? new ArrayList<>()
+                        : (List<EndpointDTO>) memObject;
+                if (input.isRegisterAction()) {
+                    endpoints.add(input);
+                    context.getLogger().log("Added endpoint for device UUID " + deviceUuid + " with url "
+                            + input.getEndpointCallbackUrl());
+                } else if (input.isUnRegisterAction()) {
+                    endpoints.remove(input);
+                }
                cacheMap.put(deviceUuid, endpoints);
-                context.getLogger().log("Added endpoint for device UUID " + deviceUuid + " with url "
-                        + input.getEndpointCallbackUrl());
            }
        }
-        return input.getEndpointUuid();
+        return new Gson().toJson(AWSResponseWrapper.successResponseAsJson(input.getEndpointUuid()));
    }
 }
@@ -8,6 +8,8 @@ import java.net.URL;
 import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;

 import org.redisson.api.RMap;

@@ -15,6 +17,7 @@ import com.amazonaws.services.lambda.runtime.Context;
 import com.amazonaws.services.lambda.runtime.LambdaLogger;
 import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
 import com.google.gson.Gson;
+import com.sap.sailing.ingestion.dto.AWSResponseWrapper;
 import com.sap.sailing.ingestion.dto.EndpointDTO;
 import com.sap.sailing.ingestion.dto.FixHeaderDTO;

@@ -23,6 +26,9 @@ import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.utils.IoUtils;

+/**
+ * 
+ */
 public class FixIngestionLambda implements RequestStreamHandler {
    @Override
    public void handleRequest(InputStream input, OutputStream output, Context context) {
@@ -30,35 +36,47 @@ public class FixIngestionLambda implements RequestStreamHandler {
            final byte[] streamAsBytes = IoUtils.toByteArray(input);
            FixHeaderDTO dto = new Gson().fromJson(new String(streamAsBytes), FixHeaderDTO.class);
            storeFixFileToS3(dto.getDeviceUuid(), streamAsBytes, context.getLogger());
-            context.getLogger().log("Getting data for device uuid " + dto.getDeviceUuid());
            RMap<String, List<EndpointDTO>> cacheMap = Utils.getCacheMap();
            final List<EndpointDTO> listOfEndpointsToTrigger = cacheMap.get(dto.getDeviceUuid());
            if (listOfEndpointsToTrigger != null) {
-                context.getLogger().log("Connecting to endpoints");
                final List<EndpointDTO> endpointsToTrigger = listOfEndpointsToTrigger;
+                final ForkJoinPool dispatchToSubscribersTask = ForkJoinPool.commonPool();
                for (final EndpointDTO endpoint : endpointsToTrigger) {
-                    context.getLogger().log("Connecting to endpoint " + endpoint.getEndpointCallbackUrl());
-                    final URL endpointUrl = new URL(endpoint.getEndpointCallbackUrl());
-                    final HttpURLConnection connectionToEndpoint = (HttpURLConnection) endpointUrl.openConnection();
-                    connectionToEndpoint.setRequestMethod("POST");
-                    connectionToEndpoint.setRequestProperty("Content-Type", "application/json; utf-8");
-                    connectionToEndpoint.setRequestProperty("Accept", "application/json");
-                    connectionToEndpoint.setDoOutput(true);
-                    connectionToEndpoint.setConnectTimeout((int) Duration.ofSeconds(3).toMillis());
-                    final byte[] jsonAsBytes = new Gson().toJson(input).getBytes();
-                    try (final OutputStream os = connectionToEndpoint.getOutputStream()) {
-                        os.write(jsonAsBytes);
-                    }
+                    dispatchToSubscribersTask.submit(() -> {
+                        dispatchToSubscribers(context, endpoint, streamAsBytes);
+                    });
                }
+                // wait for tasks to complete for <number of end-points>*<timeout for connection>+<ramp-up time>
+                dispatchToSubscribersTask.awaitQuiescence((endpointsToTrigger.size() * 3) + 2, TimeUnit.SECONDS);
            } else {
                context.getLogger().log("No endpoint has been configured for UUID " + dto.getDeviceUuid());
            }
+            output.write(new Gson().toJson(AWSResponseWrapper.successResponseAsJson(dto.getDeviceUuid())).getBytes());
        } catch (IOException e) {
            context.getLogger().log(e.getMessage());
        }
    }

-    private void storeFixFileToS3(String deviceUuid, byte[] jsonAsBytes, LambdaLogger logger) throws IOException {
+    private void dispatchToSubscribers(final Context context, final EndpointDTO endpoint, final byte[] jsonAsBytes) {
+        context.getLogger().log("Connecting to endpoint " + endpoint.getEndpointCallbackUrl());
+        try {
+            final URL endpointUrl = new URL(endpoint.getEndpointCallbackUrl());
+            final HttpURLConnection connectionToEndpoint = (HttpURLConnection) endpointUrl.openConnection();
+            connectionToEndpoint.setRequestMethod("POST");
+            connectionToEndpoint.setRequestProperty("Content-Type", "application/json; utf-8");
+            connectionToEndpoint.setRequestProperty("Accept", "application/json");
+            connectionToEndpoint.setDoOutput(true);
+            connectionToEndpoint.setConnectTimeout((int) Duration.ofSeconds(3).toMillis());
+            try (final OutputStream os = connectionToEndpoint.getOutputStream()) {
+                os.write(jsonAsBytes);
+            }
+        } catch (IOException ex) {
+            context.getLogger().log(ex.getMessage());
+        }
+    }
+
+    private void storeFixFileToS3(final String deviceUuid, final byte[] jsonAsBytes, final LambdaLogger logger)
+            throws IOException {
        try (final S3Client s3Client = S3Client.builder().region(Configuration.S3_REGION).build()) {
            final String destinationKey = getDestinationKey(deviceUuid);
            final PutObjectRequest putObjectRequest = PutObjectRequest.builder().bucket(Configuration.S3_BUCKET_NAME)
@@ -68,11 +86,11 @@ public class FixIngestionLambda implements RequestStreamHandler {
        }
    }

-    private String getDestinationKey(String deviceUuid) {
+    private String getDestinationKey(final String deviceUuid) {
        return getUuidSplitIntoS3Prefixes(deviceUuid) + "/" + LocalDateTime.now().toString() + ".json";
    }

-    private String getUuidSplitIntoS3Prefixes(String uuid) {
+    private String getUuidSplitIntoS3Prefixes(final String uuid) {
        return String.join("/", uuid.split("-"));
    }
 }
@@ -0,0 +1,15 @@
+package com.sap.sailing.ingestion;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.google.gson.Gson;
+import com.sap.sailing.ingestion.dto.FixHeaderDTO;
+
+public class PingLambda implements RequestHandler<FixHeaderDTO, String> {
+    @Override
+    public String handleRequest(FixHeaderDTO input, Context context) {
+        final String jsonAsString = new Gson().toJson(input);
+        context.getLogger().log(jsonAsString);
+        return jsonAsString;
+    }
+}
@@ -0,0 +1,19 @@
+package com.sap.sailing.ingestion.dto;
+
+import java.io.Serializable;
+
+public class AWSIngestionWrapper<T> implements Serializable {
+
+    private static final long serialVersionUID = 6640992447269375968L;
+
+    private T body;
+
+    public T getBody() {
+        return body;
+    }
+
+    public void setBody(T body) {
+        this.body = body;
+    }
+
+}
@@ -0,0 +1,24 @@
+package com.sap.sailing.ingestion.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class AWSResponseHeader {
+
+    private String contentType;
+
+    public static AWSResponseHeader jsonResponse() {
+        final AWSResponseHeader result = new AWSResponseHeader();
+        result.setContentType("application/json");
+        return result;
+    }
+
+    @JsonProperty("Content-Type")
+    public String getContentType() {
+        return contentType;
+    }
+
+    public void setContentType(String contentType) {
+        this.contentType = contentType;
+    }
+
+}
@@ -0,0 +1,59 @@
+package com.sap.sailing.ingestion.dto;
+
+import java.io.Serializable;
+
+public class AWSResponseWrapper<T> implements Serializable {
+
+    private static final long serialVersionUID = -3550784751905818148L;
+
+    private int statusCode;
+    private String statusDescription;
+    private AWSResponseHeader headers;
+
+    /**
+     * Returns a response that signals callers that the request has been successful
+     */
+    public static <C> AWSResponseWrapper<C> successResponseAsJson(C input) {
+        final AWSResponseWrapper<C> result = new AWSResponseWrapper<C>();
+        result.setStatusCode(200);
+        result.setStatusDescription("200 OK");
+        result.setHeaders(AWSResponseHeader.jsonResponse());
+        result.setBody(input);
+        return result;
+    }
+
+    private T body;
+
+    public T getBody() {
+        return body;
+    }
+
+    public void setBody(T body) {
+        this.body = body;
+    }
+
+    public int getStatusCode() {
+        return statusCode;
+    }
+
+    public void setStatusCode(int statusCode) {
+        this.statusCode = statusCode;
+    }
+
+    public String getStatusDescription() {
+        return statusDescription;
+    }
+
+    public void setStatusDescription(String statusDescription) {
+        this.statusDescription = statusDescription;
+    }
+
+    public AWSResponseHeader getHeaders() {
+        return headers;
+    }
+
+    public void setHeaders(AWSResponseHeader headers) {
+        this.headers = headers;
+    }
+
+}
@@ -5,8 +5,15 @@ import java.util.List;

 public class EndpointDTO implements Serializable {
    private static final long serialVersionUID = 3115461658787136449L;
+    
+    public final static String REGISTER_ACTION = "register";
+    public final static String UNREGISTER_ACTION = "unregister";

    private String endpointUuid;
+    /**
+     * One of register or unregister
+     */
+    private String action;
    private String endpointCallbackUrl;
    private List<String> devicesUuid;

@@ -33,4 +40,35 @@ public class EndpointDTO implements Serializable {
    public void setDevicesUuid(List<String> devicesUuid) {
        this.devicesUuid = devicesUuid;
    }
+
+    public String getAction() {
+        return action;
+    }
+
+    public void setAction(String action) {
+        this.action = action;
+    }
+    
+    public boolean isRegisterAction() {
+        return getAction().equalsIgnoreCase(REGISTER_ACTION);
+    }
+
+    public boolean isUnRegisterAction() {
+        return getAction().equalsIgnoreCase(UNREGISTER_ACTION);
+    }
+    
+    @Override
+    public boolean equals(Object o) {
+        if (o == this)
+            return true;
+        if (!(o instanceof EndpointDTO))
+            return false;
+        EndpointDTO other = (EndpointDTO)o;
+        return getEndpointUuid().equals(other.getEndpointUuid());
+    }
+    
+    @Override
+    public int hashCode() {
+        return getEndpointUuid().hashCode();
+    }
 }