GH-1210 Add S3EventSerializer logic to ensure S3Event deserialization

Resolves #1210

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

@@ -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);
 	}
 

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

@@ -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;