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

Merge branch '4.1.x' of github.com:spring-cloud/spring-cloud-function into 4.1.x

Browse Source
This commit is contained in:
Ryan Baxter
2025-04-13 09:35:42 -04:00
parent f780ce2519 8954784106
commit 811e8172bb
71 changed files with 307 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/pom.xml
View File

@@ -13,13 +13,13 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-adapter-parent</artifactId>
<version>4.1.4-SNAPSHOT</version>
<version>4.1.6-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<aws-lambda-events.version>3.11.4</aws-lambda-events.version>
<aws-lambda-events.version>3.14.0</aws-lambda-events.version>
<aws-java-sdk.version>1.12.29</aws-java-sdk.version>
<aws-lambda-java-log4j.version>1.0.1</aws-lambda-java-log4j.version>
<aws-lambda-java-serialization.version>1.1.5</aws-lambda-java-serialization.version>

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

@@ -17,11 +17,14 @@
package org.springframework.cloud.function.adapter.aws;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import com.amazonaws.services.lambda.runtime.serialization.PojoSerializer;
import com.amazonaws.services.lambda.runtime.serialization.events.LambdaEventSerializers;
import com.amazonaws.services.lambda.runtime.serialization.events.serializers.S3EventSerializer;
import org.springframework.cloud.function.cloudevent.CloudEventMessageUtils;
import org.springframework.cloud.function.context.config.JsonMessageConverter;
@@ -30,6 +33,7 @@ import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.ClassUtils;
import org.springframework.util.MimeType;
/**
@@ -44,6 +48,9 @@ class AWSTypesMessageConverter extends JsonMessageConverter {
private final JsonMapper jsonMapper;
@SuppressWarnings("rawtypes")
private final AtomicReference<S3EventSerializer> s3EventSerializer = new AtomicReference<>();
AWSTypesMessageConverter(JsonMapper jsonMapper) {
this(jsonMapper, new MimeType("application", "json"), new MimeType(CloudEventMessageUtils.APPLICATION_CLOUDEVENTS.getType(),
CloudEventMessageUtils.APPLICATION_CLOUDEVENTS.getSubtype() + "+json"));
@@ -75,7 +82,6 @@ class AWSTypesMessageConverter extends JsonMessageConverter {
if (message.getPayload().getClass().isAssignableFrom(targetClass)) {
return message.getPayload();
}
if (targetClass.getPackage() != null &&
targetClass.getPackage().getName().startsWith("com.amazonaws.services.lambda.runtime.events")) {
PojoSerializer<?> serializer = LambdaEventSerializers.serializerFor(targetClass, Thread.currentThread().getContextClassLoader());
@@ -110,12 +116,23 @@ class AWSTypesMessageConverter extends JsonMessageConverter {
}
@SuppressWarnings("unchecked")
@Override
protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers,
@Nullable Object conversionHint) {
if (payload instanceof String && headers.containsKey(AWSLambdaUtils.IS_BASE64_ENCODED) && (boolean) headers.get(AWSLambdaUtils.IS_BASE64_ENCODED)) {
return ((String) payload).getBytes(StandardCharsets.UTF_8);
}
if (payload.getClass().getName().equals("com.amazonaws.services.lambda.runtime.events.S3Event")) {
if (this.s3EventSerializer.get() == null) {
this.s3EventSerializer.set(new S3EventSerializer<>().withClassLoader(ClassUtils.getDefaultClassLoader()));
}
ByteArrayOutputStream stream = new ByteArrayOutputStream();
this.s3EventSerializer.get().toJson(payload, stream);
return stream.toByteArray();
}
return jsonMapper.toJson(payload);
}

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

@@ -28,10 +28,14 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import com.amazonaws.services.lambda.runtime.ClientContext;
import com.amazonaws.services.lambda.runtime.CognitoIdentity;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.LambdaLogger;
import com.amazonaws.services.lambda.runtime.LambdaRuntime;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
import org.springframework.cloud.function.context.config.RoutingFunction;
@@ -130,6 +134,8 @@ public final class CustomRuntimeEventLoop implements SmartLifecycle {
logger.debug("Attempting to get new event");
ResponseEntity<String> response = this.pollForData(rest, requestEntity);
Context clientContext = generateClientContext(response.getHeaders());
if (logger.isDebugEnabled()) {
logger.debug("New Event received: " + response);
}
@@ -140,9 +146,9 @@ public final class CustomRuntimeEventLoop implements SmartLifecycle {
FunctionInvocationWrapper function = locateFunction(environment, functionCatalog, response.getHeaders());
ByteArrayInputStream is = new ByteArrayInputStream(response.getBody().getBytes(StandardCharsets.UTF_8));
Message<?> requestMessage = AWSLambdaUtils.generateMessage(is, function.getInputType(), function.isSupplier(), mapper, null);
Message<?> requestMessage = AWSLambdaUtils.generateMessage(is, function.getInputType(), function.isSupplier(), mapper, clientContext);
Object functionResponse = function.apply(requestMessage);
byte[] responseBytes = AWSLambdaUtils.generateOutputFromObject(requestMessage, functionResponse, mapper, function.getOutputType());
String invocationUrl = MessageFormat
@@ -157,12 +163,91 @@ public final class CustomRuntimeEventLoop implements SmartLifecycle {
}
}
catch (Exception e) {
e.printStackTrace();
this.propagateAwsError(requestId, e, mapper, runtimeApi, rest);
}
}
}
}
private Context generateClientContext(HttpHeaders headers) {
Map<String, String> environment = System.getenv();
Context context = new Context() {
@Override
public int getRemainingTimeInMillis() {
long now = System.currentTimeMillis();
if (!headers.containsKey("Lambda-Runtime-Deadline-Ms")) {
return 0;
}
int delta = (int) (Long.parseLong(headers.getFirst("Lambda-Runtime-Deadline-Ms")) - now);
return delta > 0 ? delta : 0;
}
@Override
public int getMemoryLimitInMB() {
if (!environment.containsKey("AWS_LAMBDA_FUNCTION_MEMORY_SIZE")) {
return 128;
}
return Integer.parseInt(environment.getOrDefault("AWS_LAMBDA_FUNCTION_MEMORY_SIZE", "128"));
}
@Override
public LambdaLogger getLogger() {
return LambdaRuntime.getLogger();
}
@Override
public String getLogStreamName() {
return environment.get("LOG_STREAM_NAME");
}
@Override
public String getLogGroupName() {
return environment.get("LOG_GROUP_NAME");
}
@Override
public String getInvokedFunctionArn() {
return headers.getFirst("Lambda-Runtime-Invoked-Function-Arn");
}
@Override
public CognitoIdentity getIdentity() {
return null;
}
@Override
public String getFunctionVersion() {
return environment.get("FUNCTION_VERSION");
}
@Override
public String getFunctionName() {
return environment.get("FUNCTION_NAME");
}
@Override
public ClientContext getClientContext() {
return null;
}
@Override
public String getAwsRequestId() {
return headers.getFirst("Lambda-Runtime-Aws-Request-Id");
}
public String toString() {
return "FUNCTION NAME: " + getFunctionName() + ", FUNCTION VERSION: " + getFunctionVersion()
+ ", FUNCTION ARN: " + getInvokedFunctionArn() + ", FUNCTION MEM LIMIT: " + getMemoryLimitInMB()
+ ", FUNCTION DEADLINE: " + getRemainingTimeInMillis();
}
};
return context;
}
private void propagateAwsError(String requestId, Exception e, JsonMapper mapper, String runtimeApi, RestTemplate rest) {
String errorMessage = e.getMessage();
String errorType = e.getClass().getSimpleName();

19
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/test/java/org/springframework/cloud/function/adapter/aws/FunctionInvokerTests.java
View File

@@ -999,6 +999,18 @@ public class FunctionInvokerTests {
assertThat(result).contains("s3SchemaVersion");
}
@Test
public void testS3EventAsOutput() throws Exception {
System.setProperty("MAIN_CLASS", S3Configuration.class.getName());
System.setProperty("spring.cloud.function.definition", "outputS3Event");
FunctionInvoker invoker = new FunctionInvoker();
InputStream targetStream = new ByteArrayInputStream(this.s3Event.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
assertThat(output.toByteArray()).isNotNull();
}
@Test
public void testS3Event() throws Exception {
System.setProperty("MAIN_CLASS", S3Configuration.class.getName());
@@ -1679,6 +1691,13 @@ public class FunctionInvokerTests {
@EnableAutoConfiguration
@Configuration
public static class S3Configuration {
@Bean
public Function<S3Event, S3Event> outputS3Event() {
return v -> {
return v;
};
}
@Bean
public Function<String, String> echoString() {
return v -> v;