SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1052 Fix collection/array processing for AWS invocations with Publisher input type functions

Browse Source 
Resolves #1052
This commit is contained in:
Oleg Zhurakousky
2023-07-13 15:20:22 +02:00
parent 351e73b7f4
commit aff82e1457
7 changed files with 293 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/AWSLambdaUtils.java
View File

@@ -74,14 +74,7 @@ public final class AWSLambdaUtils {
if (FunctionTypeUtils.isMessage(inputType) || FunctionTypeUtils.isPublisher(inputType)) {
inputType = FunctionTypeUtils.getImmediateGenericType(inputType, 0);
}
String typeName = inputType.getTypeName();
return typeName.equals("com.amazonaws.services.lambda.runtime.events.APIGatewayV2HTTPEvent")
|| typeName.equals("com.amazonaws.services.lambda.runtime.events.S3Event")
|| typeName.equals("com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent")
|| typeName.equals("com.amazonaws.services.lambda.runtime.events.SNSEvent")
|| typeName.equals("com.amazonaws.services.lambda.runtime.events.SQSEvent")
|| typeName.equals("com.amazonaws.services.lambda.runtime.events.APIGatewayCustomAuthorizerEvent")
|| typeName.equals("com.amazonaws.services.lambda.runtime.events.KinesisEvent");
return FunctionTypeUtils.getRawType(inputType).getPackage().getName().startsWith("com.amazonaws.services.lambda.runtime.events");
}
@SuppressWarnings("rawtypes")
@@ -121,6 +114,9 @@ public final class AWSLambdaUtils {
MessageBuilder<byte[]> builder = MessageBuilder.withPayload(payload);
if (isApiGateway) {
builder.setHeader(AWSLambdaUtils.AWS_API_GATEWAY, true);
if (JsonMapper.isJsonStringRepresentsCollection(((Map) structMessage).get("body"))) {
builder.setHeader("payload", ((Map) structMessage).get("body"));
}
}
if (!isSupplier && AWSLambdaUtils.isSupportedAWSType(inputType)) {
builder.setHeader(AWSLambdaUtils.AWS_EVENT, true);
@@ -145,35 +141,50 @@ public final class AWSLambdaUtils {
}
}
private static Object convertFromJsonIfNecessary(Object value, JsonMapper objectMapper) {
if (JsonMapper.isJsonString(value)) {
return objectMapper.fromJson(value, Object.class);
}
return value;
}
@SuppressWarnings("unchecked")
public static byte[] generateOutputFromObject(Message<?> requestMessage, Object output, JsonMapper objectMapper, Type functionOutputType) {
Message<byte[]> responseMessage = null;
if (output instanceof Publisher<?>) {
List<Object> result = new ArrayList<>();
for (Object value : Flux.from((Publisher<?>) output).toIterable()) {
Message<?> lastMessage = null;
for (Object item : Flux.from((Publisher<?>) output).toIterable()) {
if (logger.isDebugEnabled()) {
logger.debug("Response value: " + value);
logger.debug("Response value: " + item);
}
if (item instanceof Message<?> message) {
result.add(convertFromJsonIfNecessary(message.getPayload(), objectMapper));
lastMessage = message;
}
else {
result.add(convertFromJsonIfNecessary(item, objectMapper));
}
result.add(value);
}
if (result.size() > 1) {
output = result;
byte[] resultPayload;
if (result.size() == 1) {
resultPayload = objectMapper.toJson(result.get(0));
}
else if (result.size() == 1) {
output = result.get(0);
else if (result.size() > 1) {
resultPayload = objectMapper.toJson(result);
}
else {
output = null;
resultPayload = null;
}
if (output instanceof Message<?> && ((Message<?>) output).getPayload() instanceof byte[]) {
responseMessage = (Message<byte[]>) output;
}
else if (output != null) {
if (logger.isDebugEnabled()) {
logger.debug("OUTPUT: " + output + " - " + output.getClass().getName());
if (resultPayload != null) {
System.out.println(new String(resultPayload));
MessageBuilder<byte[]> messageBuilder = MessageBuilder.withPayload(resultPayload);
if (lastMessage != null) {
messageBuilder.copyHeaders(lastMessage.getHeaders());
}
byte[] payload = objectMapper.toJson(output);
responseMessage = MessageBuilder.withPayload(payload).build();
responseMessage = messageBuilder.build();
}
}
else {

8
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/AWSTypesMessageConverter.java
View File

@@ -86,7 +86,13 @@ class AWSTypesMessageConverter extends JsonMessageConverter {
return structMessage;
}
else {
Object body = structMessage.get("body");
Object body;
if (message.getHeaders().containsKey("payload")) {
body = message.getPayload();
}
else {
body = structMessage.get("body");
}
Object convertedResult = this.jsonMapper.fromJson(body, targetClass);
return convertedResult;
}

30
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/CustomRuntimeEventLoop.java
View File

@@ -16,6 +16,7 @@
package org.springframework.cloud.function.adapter.aws;
import java.io.ByteArrayInputStream;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.SocketException;
@@ -44,11 +45,8 @@ import org.springframework.http.RequestEntity;
import org.springframework.http.ResponseEntity;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;
import static org.apache.http.HttpHeaders.USER_AGENT;
/**
@@ -107,7 +105,6 @@ public final class CustomRuntimeEventLoop implements SmartLifecycle {
return this.running;
}
@SuppressWarnings("unchecked")
private void eventLoop(ConfigurableApplicationContext context) {
Environment environment = context.getEnvironment();
logger.info("Starting spring-cloud-function CustomRuntimeEventLoop");
@@ -140,33 +137,18 @@ public final class CustomRuntimeEventLoop implements SmartLifecycle {
try {
FunctionInvocationWrapper function = locateFunction(environment, functionCatalog, response.getHeaders());
Message<byte[]> eventMessage = AWSLambdaUtils
.generateMessage(response.getBody().getBytes(StandardCharsets.UTF_8), function.getInputType(), function.isSupplier(), mapper);
ByteArrayInputStream is = new ByteArrayInputStream(response.getBody().getBytes(StandardCharsets.UTF_8));
Message<?> requestMessage = AWSLambdaUtils.generateMessage(is, function.getInputType(), function.isSupplier(), mapper, null);
if (logger.isDebugEnabled()) {
logger.debug("Event message: " + eventMessage);
}
Object functionResponse = function.apply(requestMessage);
byte[] responseBytes = AWSLambdaUtils.generateOutputFromObject(requestMessage, functionResponse, mapper, function.getOutputType());
String invocationUrl = MessageFormat
.format(LAMBDA_INVOCATION_URL_TEMPLATE, runtimeApi, LAMBDA_VERSION_DATE, requestId);
String traceId = response.getHeaders().getFirst("Lambda-Runtime-Trace-Id");
if (StringUtils.hasText(traceId)) {
if (logger.isDebugEnabled()) {
logger.debug("Lambda-Runtime-Trace-Id: " + traceId);
}
System.setProperty("com.amazonaws.xray.traceHeader", traceId);
}
Object responseObject = function.apply(eventMessage);
if (responseObject != null && logger.isDebugEnabled()) {
logger.debug("Reply from function: " + responseObject);
}
byte[] outputBody = AWSLambdaUtils.generateOutputFromObject(eventMessage, responseObject, mapper, function.getOutputType());
ResponseEntity<Object> result = rest.exchange(RequestEntity.post(URI.create(invocationUrl))
.header(USER_AGENT, USER_AGENT_VALUE)
.body(outputBody), Object.class);
.body(responseBytes), Object.class);
if (logger.isInfoEnabled()) {
logger.info("Result POST status: " + result);