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

GH-562 Add type conversion documentation

Browse Source 
Add test in AWS to showcase type conversion
Fix AWS FunctionInvoker to delegate to effectively delegate type conversion to the native mechanism of spring-cloud-function

Resolves #562
This commit is contained in:
Oleg Zhurakousky
2020-07-29 20:17:03 +02:00
parent 48b8f4ce89
commit 6b9ce4cb0c
5 changed files with 236 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,10 +31,6 @@ import java.util.Map;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
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.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonDeserializer;
@@ -165,7 +161,7 @@ public class FunctionInvoker implements RequestStreamHandler {
@SuppressWarnings({ "unchecked", "rawtypes" })
private Message<byte[]> generateMessage(InputStream input, Context context) throws IOException {
byte[] payload = StreamUtils.copyToByteArray(input);
final byte[] payload = StreamUtils.copyToByteArray(input);
if (logger.isInfoEnabled()) {
logger.info("Incoming JSON for ApiGateway Event: " + new String(payload));
@@ -183,34 +179,8 @@ public class FunctionInvoker implements RequestStreamHandler {
if (requestMap.containsKey("Records")) {
List<Map<String, ?>> records = (List<Map<String, ?>>) requestMap.get("Records");
Assert.notEmpty(records, "Incoming event has no records: " + requestMap);
if (this.isKinesisEvent(records.get(0))) {
logger.info("Incoming request is Kinesis Event");
Assert.isTrue(inputType instanceof Class && KinesisEvent.class.isAssignableFrom((Class<?>) inputType) || mapInputType,
"Only KinesisEvent or Map type is supported as input type for functions that accept Kinesis Event");
Object event = mapInputType ? requestMap : this.mapper.convertValue(requestMap, KinesisEvent.class);
messageBuilder = MessageBuilder.withPayload(event);
}
else if (this.isS3Event(records.get(0))) {
logger.info("Incoming request is S3 Event");
Assert.isTrue(inputType instanceof Class && S3Event.class.isAssignableFrom((Class<?>) inputType) || mapInputType,
"Only S3Event or Map type is supported as input type for functions that accept S3 Event");
Object event = mapInputType ? requestMap : this.mapper.convertValue(requestMap, S3Event.class);
messageBuilder = MessageBuilder.withPayload(event);
}
else if (this.isSNSEvent(records.get(0))) {
logger.info("Incoming request is SNS Event");
Assert.isTrue(inputType instanceof Class && SNSEvent.class.isAssignableFrom((Class<?>) inputType) || mapInputType,
"Only SNSEvent or Map type is supported as input type for functions that accept SNSEvent");
Object event = mapInputType ? requestMap : this.mapper.convertValue(requestMap, SNSEvent.class);
messageBuilder = MessageBuilder.withPayload(event);
}
else {
logger.info("Incoming request is SQS Event");
Assert.isTrue(inputType instanceof Class && SQSEvent.class.isAssignableFrom((Class<?>) inputType) || mapInputType,
"Only SQSEvent or Map type is supported as input type for functions that accept SQS Event");
Object event = mapInputType ? requestMap : this.mapper.convertValue(requestMap, SQSEvent.class);
messageBuilder = MessageBuilder.withPayload(event);
}
this.logEvent(records);
messageBuilder = MessageBuilder.withPayload(payload);
}
else if (requestMap.containsKey("httpMethod")) { // API Gateway
logger.info("Incoming request is API Gateway");
@@ -219,14 +189,12 @@ public class FunctionInvoker implements RequestStreamHandler {
messageBuilder = MessageBuilder.withPayload(gatewayEvent);
}
else if (mapInputType) {
messageBuilder = MessageBuilder.withPayload(requestMap)
.setHeader("httpMethod", requestMap.get("httpMethod"));
messageBuilder = MessageBuilder.withPayload(requestMap).setHeader("httpMethod", requestMap.get("httpMethod"));
}
else {
Object body = requestMap.remove("body");
body = body instanceof String ? ("\"" + body + "\"").getBytes(StandardCharsets.UTF_8) : mapper.writeValueAsBytes(body);
messageBuilder = MessageBuilder.withPayload(body)
.copyHeaders(requestMap);
messageBuilder = MessageBuilder.withPayload(body).copyHeaders(requestMap);
}
}
}
@@ -236,6 +204,21 @@ public class FunctionInvoker implements RequestStreamHandler {
return messageBuilder.setHeader("aws-context", context).build();
}
private void logEvent(List<Map<String, ?>> records) {
if (this.isKinesisEvent(records.get(0))) {
logger.info("Incoming request is Kinesis Event");
}
else if (this.isS3Event(records.get(0))) {
logger.info("Incoming request is S3 Event");
}
else if (this.isSNSEvent(records.get(0))) {
logger.info("Incoming request is SNS Event");
}
else {
logger.info("Incoming request is SQS Event");
}
}
private boolean isSNSEvent(Map<String, ?> record) {
return record.containsKey("Sns");
}

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

@@ -29,13 +29,14 @@ 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.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.util.MimeType;
import static org.assertj.core.api.Assertions.assertThat;
@@ -359,15 +360,15 @@ public class FunctionInvokerTests {
@Test
public void testKinesisStringEvent() throws Exception {
Assertions.assertThrows(IllegalArgumentException.class, () -> {
System.setProperty("MAIN_CLASS", KinesisConfiguration.class.getName());
System.setProperty("spring.cloud.function.definition", "echoString");
FunctionInvoker invoker = new FunctionInvoker();
System.setProperty("MAIN_CLASS", KinesisConfiguration.class.getName());
System.setProperty("spring.cloud.function.definition", "echoString");
FunctionInvoker invoker = new FunctionInvoker();
InputStream targetStream = new ByteArrayInputStream(this.sampleKinesisEvent.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
});
InputStream targetStream = new ByteArrayInputStream(this.sampleKinesisEvent.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
String result = new String(output.toByteArray(), StandardCharsets.UTF_8);
assertThat(result).contains("kinesisSchemaVersion");
}
@Test
@@ -414,15 +415,15 @@ public class FunctionInvokerTests {
@Test
public void testSQSStringEvent() throws Exception {
Assertions.assertThrows(IllegalArgumentException.class, () -> {
System.setProperty("MAIN_CLASS", SQSConfiguration.class.getName());
System.setProperty("spring.cloud.function.definition", "echoString");
FunctionInvoker invoker = new FunctionInvoker();
System.setProperty("MAIN_CLASS", SQSConfiguration.class.getName());
System.setProperty("spring.cloud.function.definition", "echoString");
FunctionInvoker invoker = new FunctionInvoker();
InputStream targetStream = new ByteArrayInputStream(this.sampleSQSEvent.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
});
InputStream targetStream = new ByteArrayInputStream(this.sampleSQSEvent.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
String result = new String(output.toByteArray(), StandardCharsets.UTF_8);
assertThat(result.length()).isEqualTo(14); // some additional JSON formatting
}
@Test
@@ -469,15 +470,15 @@ public class FunctionInvokerTests {
@Test
public void testSNSStringEvent() throws Exception {
Assertions.assertThrows(IllegalArgumentException.class, () -> {
System.setProperty("MAIN_CLASS", SNSConfiguration.class.getName());
System.setProperty("spring.cloud.function.definition", "echoString");
FunctionInvoker invoker = new FunctionInvoker();
System.setProperty("MAIN_CLASS", SNSConfiguration.class.getName());
System.setProperty("spring.cloud.function.definition", "echoString");
FunctionInvoker invoker = new FunctionInvoker();
InputStream targetStream = new ByteArrayInputStream(this.sampleSNSEvent.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
});
InputStream targetStream = new ByteArrayInputStream(this.sampleSNSEvent.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
String result = new String(output.toByteArray(), StandardCharsets.UTF_8);
assertThat(result).contains("arn:aws:sns");
}
@Test
@@ -522,18 +523,17 @@ public class FunctionInvokerTests {
assertThat(result).contains("arn:aws:sns");
}
@Test
public void testS3StringEvent() throws Exception {
Assertions.assertThrows(IllegalArgumentException.class, () -> {
System.setProperty("MAIN_CLASS", S3Configuration.class.getName());
System.setProperty("spring.cloud.function.definition", "echoString");
FunctionInvoker invoker = new FunctionInvoker();
System.setProperty("MAIN_CLASS", S3Configuration.class.getName());
System.setProperty("spring.cloud.function.definition", "echoString");
FunctionInvoker invoker = new FunctionInvoker();
InputStream targetStream = new ByteArrayInputStream(this.s3Event.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
});
InputStream targetStream = new ByteArrayInputStream(this.s3Event.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
String result = new String(output.toByteArray(), StandardCharsets.UTF_8);
assertThat(result).contains("s3SchemaVersion");
}
@Test
@@ -693,8 +693,11 @@ public class FunctionInvokerTests {
@Configuration
public static class SQSConfiguration {
@Bean
public Function<String, String> echoString() {
return v -> v;
public Function<Person, String> echoString() {
return v -> {
System.out.println("Echo: " + v);
return v.toString();
};
}
@Bean
@@ -720,6 +723,32 @@ public class FunctionInvokerTests {
return v.toString();
};
}
@Bean
public MyCustomMessageConverter messageConverter() {
return new MyCustomMessageConverter();
}
}
public static class MyCustomMessageConverter extends AbstractMessageConverter {
public MyCustomMessageConverter() {
super(new MimeType("*", "*"));
}
@Override
protected boolean supports(Class<?> clazz) {
return (Person.class.equals(clazz));
}
@Override
protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
Object payload = message.getPayload();
String v = payload instanceof String ? (String) payload : new String((byte[]) payload);
Person person = new Person();
person.setName(v.substring(0, 10));
return person;
}
}
@EnableAutoConfiguration
@@ -727,7 +756,10 @@ public class FunctionInvokerTests {
public static class SNSConfiguration {
@Bean
public Function<String, String> echoString() {
return v -> v;
return v -> {
System.out.println("Received: " + v);
return v.toString();
};
}
@Bean
@@ -834,5 +866,10 @@ public class FunctionInvokerTests {
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return this.name;
}
}
}