From 7f667a1067a89cec0c65d29d38351beeac0c25cc Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 22 Jul 2021 19:11:12 +0200 Subject: [PATCH] Gh-709, GH-648 Add support for API v2 gateway event This is lighter version of '3666bf73f707c7969d861e2a2545a3ac93d0c713' commit since it provides a version that supports v1 event as well Basically if you have appropriate AWS library on the classpath then this version will support v2 event Refactor and simplify AWSLambdaUtils to work with AWS serialization libraries --- .../spring-cloud-function-adapter-aws/pom.xml | 8 +- .../function/adapter/aws/AWSLambdaUtils.java | 172 ++++++++---------- 2 files changed, 80 insertions(+), 100 deletions(-) diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/pom.xml b/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/pom.xml index 9e57a3366..827d1ff7f 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/pom.xml +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/pom.xml @@ -50,6 +50,12 @@ com.fasterxml.jackson.datatype jackson-datatype-joda + + com.amazonaws + aws-lambda-java-serialization + 1.0.0 + provided + com.amazonaws aws-lambda-java-log4j @@ -59,7 +65,7 @@ com.amazonaws aws-lambda-java-core - 1.2.0 + 1.2.1 provided diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/AWSLambdaUtils.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/AWSLambdaUtils.java index ea18f27a8..a8f328eda 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/AWSLambdaUtils.java +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/AWSLambdaUtils.java @@ -16,6 +16,7 @@ package org.springframework.cloud.function.adapter.aws; +import java.io.ByteArrayInputStream; import java.io.IOException; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; @@ -23,12 +24,17 @@ import java.nio.charset.StandardCharsets; import java.util.Calendar; import java.util.Date; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicReference; import com.amazonaws.services.lambda.runtime.Context; import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent; +import com.amazonaws.services.lambda.runtime.events.KinesisEvent; +import com.amazonaws.services.lambda.runtime.events.S3Event; +import com.amazonaws.services.lambda.runtime.events.SNSEvent; +import com.amazonaws.services.lambda.runtime.events.SQSEvent; +import com.amazonaws.services.lambda.runtime.serialization.PojoSerializer; +import com.amazonaws.services.lambda.runtime.serialization.events.LambdaEventSerializers; import com.fasterxml.jackson.core.JsonParser; import com.fasterxml.jackson.databind.DeserializationContext; import com.fasterxml.jackson.databind.JsonDeserializer; @@ -45,8 +51,6 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; /** * @@ -57,6 +61,8 @@ final class AWSLambdaUtils { private static Log logger = LogFactory.getLog(AWSLambdaUtils.class); + private static final String AWS_API_GATEWAY = "aws-api-gateway"; + private AWSLambdaUtils() { } @@ -66,71 +72,82 @@ final class AWSLambdaUtils { return generateMessage(payload, headers, inputType, objectMapper, null); } + private static boolean isSupportedAWSType(Type inputType) { + return "com.amazonaws.services.lambda.runtime.events.APIGatewayV2HTTPEvent".equals(FunctionTypeUtils.getRawType(inputType).getName()) + || S3Event.class.isAssignableFrom(FunctionTypeUtils.getRawType(inputType)) + || APIGatewayProxyRequestEvent.class.isAssignableFrom(FunctionTypeUtils.getRawType(inputType)) + || SNSEvent.class.isAssignableFrom(FunctionTypeUtils.getRawType(inputType)) + || SQSEvent.class.isAssignableFrom(FunctionTypeUtils.getRawType(inputType)) + || KinesisEvent.class.isAssignableFrom(FunctionTypeUtils.getRawType(inputType)); + } + @SuppressWarnings({ "unchecked", "rawtypes" }) public static Message generateMessage(byte[] payload, MessageHeaders headers, Type inputType, ObjectMapper objectMapper, @Nullable Context awsContext) { - if (!objectMapper.isEnabled(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES)) { - configureObjectMapper(objectMapper); - } - if (logger.isInfoEnabled()) { logger.info("Incoming JSON Event: " + new String(payload)); } - MessageBuilder messageBuilder = null; - Object request; - try { - request = objectMapper.readValue(payload, Object.class); - } - catch (Exception e) { - throw new IllegalStateException(e); - } if (FunctionTypeUtils.isMessage(inputType)) { inputType = FunctionTypeUtils.getImmediateGenericType(inputType, 0); } - boolean mapInputType = (inputType instanceof ParameterizedType && ((Class) ((ParameterizedType) inputType).getRawType()).isAssignableFrom(Map.class)); - if (request instanceof Map) { - Map requestMap = (Map) request; - if (requestMap.containsKey("Records")) { - List> records = (List>) requestMap.get("Records"); - Assert.notEmpty(records, "Incoming event has no records: " + requestMap); - logEvent(records); - messageBuilder = MessageBuilder.withPayload(payload); - } - else if (requestMap.containsKey("httpMethod")) { // API Gateway + + MessageBuilder messageBuilder = null; + if (inputType != null && isSupportedAWSType(inputType)) { + PojoSerializer serializer = LambdaEventSerializers.serializerFor(FunctionTypeUtils.getRawType(inputType), Thread.currentThread().getContextClassLoader()); + Object event = serializer.fromJson(new ByteArrayInputStream(payload)); + messageBuilder = MessageBuilder.withPayload(event); + if (event instanceof APIGatewayProxyRequestEvent || event.getClass().getName().equals("com.amazonaws.services.lambda.runtime.events.APIGatewayV2HTTPEvent")) { + messageBuilder.setHeader(AWS_API_GATEWAY, true); logger.info("Incoming request is API Gateway"); - if (isTypeAnApiGatewayRequest(inputType)) { - APIGatewayProxyRequestEvent gatewayEvent = objectMapper.convertValue(requestMap, APIGatewayProxyRequestEvent.class); - messageBuilder = MessageBuilder.withPayload(gatewayEvent); - } - else if (mapInputType) { - messageBuilder = MessageBuilder.withPayload(requestMap).setHeader("httpMethod", requestMap.get("httpMethod")); - } - else { - Object body = requestMap.remove("body"); - try { - body = body instanceof String - ? String.valueOf(body).getBytes(StandardCharsets.UTF_8) - : objectMapper.writeValueAsBytes(body); - } - catch (Exception e) { - throw new IllegalStateException(e); - } - - messageBuilder = MessageBuilder.withPayload(body).copyHeaders(requestMap); - } - } - - Object providedHeaders = requestMap.remove("headers"); - if (providedHeaders != null && providedHeaders instanceof Map) { - messageBuilder.removeHeader("headers"); - messageBuilder.copyHeaders((Map) providedHeaders); } } - else if (request instanceof Iterable) { - messageBuilder = MessageBuilder.withPayload(request); + else { + if (!objectMapper.isEnabled(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES)) { + configureObjectMapper(objectMapper); + } + Object request; + try { + request = objectMapper.readValue(payload, Object.class); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + + if (request instanceof Map) { + if (((Map) request).containsKey("httpMethod")) { //API Gateway + boolean mapInputType = (inputType instanceof ParameterizedType && ((Class) ((ParameterizedType) inputType).getRawType()).isAssignableFrom(Map.class)); + if (mapInputType) { + messageBuilder = MessageBuilder.withPayload(request).setHeader("httpMethod", ((Map) request).get("httpMethod")); + } + else { + Object body = ((Map) request).remove("body"); + try { + body = body instanceof String + ? String.valueOf(body).getBytes(StandardCharsets.UTF_8) + : objectMapper.writeValueAsBytes(body); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + + messageBuilder = MessageBuilder.withPayload(body).copyHeaders(((Map) request)); + } + messageBuilder.setHeader(AWS_API_GATEWAY, true); + } + Object providedHeaders = ((Map) request).remove("headers"); + if (providedHeaders != null && providedHeaders instanceof Map) { + messageBuilder.removeHeader("headers"); + messageBuilder.copyHeaders((Map) providedHeaders); + } + } + else if (request instanceof Iterable) { + messageBuilder = MessageBuilder.withPayload(request); + } } + + if (messageBuilder == null) { messageBuilder = MessageBuilder.withPayload(payload); } @@ -145,12 +162,13 @@ final class AWSLambdaUtils { @SuppressWarnings({ "rawtypes", "unchecked" }) public static byte[] generateOutput(Message requestMessage, Message responseMessage, ObjectMapper objectMapper) { + + if (!objectMapper.isEnabled(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES)) { configureObjectMapper(objectMapper); } byte[] responseBytes = responseMessage == null ? "\"OK\"".getBytes() : responseMessage.getPayload(); - if (requestMessage.getHeaders().containsKey("httpMethod") - || isPayloadAnApiGatewayRequest(requestMessage.getPayload())) { // API Gateway + if (requestMessage.getHeaders().containsKey(AWS_API_GATEWAY) && ((boolean) requestMessage.getHeaders().get(AWS_API_GATEWAY))) { Map response = new HashMap(); response.put("isBase64Encoded", false); @@ -186,7 +204,6 @@ final class AWSLambdaUtils { throw new IllegalStateException("Failed to serialize AWS Lambda output", e); } } - return responseBytes; } @@ -206,51 +223,8 @@ final class AWSLambdaUtils { objectMapper.configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true); } - private static boolean isPayloadAnApiGatewayRequest(Object payload) { - return isAPIGatewayProxyRequestEventPresent() - ? payload instanceof APIGatewayProxyRequestEvent - : false; - } - - private static boolean isTypeAnApiGatewayRequest(Type type) { - return type != null && isAPIGatewayProxyRequestEventPresent() - ? type.getTypeName().endsWith(APIGatewayProxyRequestEvent.class.getSimpleName()) - : false; - } - - private static boolean isAPIGatewayProxyRequestEventPresent() { - return ClassUtils.isPresent("com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent", - ClassUtils.getDefaultClassLoader()); - } - - private static void logEvent(List> records) { - if (isKinesisEvent(records.get(0))) { - logger.info("Incoming request is Kinesis Event"); - } - else if (isS3Event(records.get(0))) { - logger.info("Incoming request is S3 Event"); - } - else if (isSNSEvent(records.get(0))) { - logger.info("Incoming request is SNS Event"); - } - else { - logger.info("Incoming request is SQS Event"); - } - } private static boolean isRequestKinesis(Message requestMessage) { return requestMessage.getHeaders().containsKey("Records"); } - - private static boolean isSNSEvent(Map record) { - return record.containsKey("Sns"); - } - - private static boolean isS3Event(Map record) { - return record.containsKey("s3"); - } - - private static boolean isKinesisEvent(Map record) { - return record.containsKey("kinesis"); - } }