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

GH-932 Fix registration of AWSTypesMessageConverter for functional spring applications

Browse Source 
Resolves #932
This commit is contained in:
Oleg Zhurakousky
2022-11-10 12:13:57 +01:00
parent b33a106c48
commit 13c79f5283
10 changed files with 55 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,10 +16,13 @@
package org.springframework.cloud.function.adapter.aws;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cloud.function.json.JacksonMapper;
import org.springframework.cloud.function.json.JsonMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.context.ApplicationContextInitializer;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.util.CollectionUtils;
/**
*
@@ -27,11 +30,18 @@ import org.springframework.messaging.converter.MessageConverter;
* @since 3.2
*
*/
@Configuration(proxyBeanMethods = false)
public class AWSCompanionAutoConfiguration {
public class AWSCompanionAutoConfiguration implements ApplicationContextInitializer<GenericApplicationContext> {
@Bean
public MessageConverter awsTypesConverter(JsonMapper jsonMapper) {
return new AWSTypesMessageConverter(jsonMapper);
@Override
public void initialize(GenericApplicationContext applicationContext) {
applicationContext.registerBean("awsTypesMessageConverter", AWSTypesMessageConverter.class,
() -> {
if (CollectionUtils.isEmpty(applicationContext.getBeansOfType(JsonMapper.class).values())) {
return new AWSTypesMessageConverter(new JacksonMapper(new ObjectMapper()));
}
else {
return new AWSTypesMessageConverter(applicationContext.getBean(JsonMapper.class));
}
});
}
}

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

@@ -55,11 +55,11 @@ class AWSTypesMessageConverter extends JsonMessageConverter {
@Override
protected boolean canConvertFrom(Message<?> message, @Nullable Class<?> targetClass) {
if (message.getHeaders().containsKey(AWSLambdaUtils.AWS_API_GATEWAY) && ((boolean) message.getHeaders().get(AWSLambdaUtils.AWS_API_GATEWAY))) {
return true;
if (message.getHeaders().containsKey(AWSLambdaUtils.AWS_API_GATEWAY)) {
return ((boolean) message.getHeaders().get(AWSLambdaUtils.AWS_API_GATEWAY));
}
if (message.getHeaders().containsKey(AWSLambdaUtils.AWS_EVENT) && ((boolean) message.getHeaders().get(AWSLambdaUtils.AWS_EVENT))) {
return true;
if (message.getHeaders().containsKey(AWSLambdaUtils.AWS_EVENT)) {
return ((boolean) message.getHeaders().get(AWSLambdaUtils.AWS_EVENT));
}
return false;
}

2
spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/resources/META-INF/spring.factories
View File

@@ -1,4 +1,4 @@
org.springframework.context.ApplicationContextInitializer=\
org.springframework.cloud.function.adapter.aws.CustomRuntimeInitializer
org.springframework.cloud.function.adapter.aws.CustomRuntimeInitializer,org.springframework.cloud.function.adapter.aws.AWSCompanionAutoConfiguration
org.springframework.boot.env.EnvironmentPostProcessor=\
org.springframework.cloud.function.adapter.aws.CustomRuntimeEnvironmentPostProcessor

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

@@ -1,229 +0,0 @@
/*
* Copyright 2012-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.function.adapter.aws;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.util.Map;
import java.util.function.Function;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration;
import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.util.Assert;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Dave Syer
* @author Oleg Zhurakousky
*/
public class SpringBootStreamHandlerTests {
private SpringBootStreamHandler handler;
@BeforeEach
public void before() {
System.clearProperty("function.name");
}
@Test
public void functionBeanWithJacksonConfig() throws Exception {
this.handler = new SpringBootStreamHandler(FunctionConfigWithJackson.class);
this.handler.initialize(null);
ByteArrayOutputStream output = new ByteArrayOutputStream();
this.handler.handleRequest(
new ByteArrayInputStream("{\"value\":\"foo\"}".getBytes()), output, null);
assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}");
}
@Test
public void functionBeanWithoutJacksonConfig() throws Exception {
this.handler = new SpringBootStreamHandler(FunctionConfigWithoutJackson.class);
this.handler.initialize(null);
ByteArrayOutputStream output = new ByteArrayOutputStream();
this.handler.handleRequest(
new ByteArrayInputStream("{\"value\":\"foo\"}".getBytes()), output, null);
assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}");
}
@Test
public void functionNonFluxBeanNoCatalog() throws Exception {
this.handler = new SpringBootStreamHandler(NoCatalogNonFluxFunctionConfig.class);
this.handler.initialize(null);
ByteArrayOutputStream output = new ByteArrayOutputStream();
this.handler.handleRequest(
new ByteArrayInputStream("{\"value\":\"foo\"}".getBytes()), output, null);
assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}");
}
@Test
public void functionFluxBeanNoCatalog() throws Exception {
this.handler = new SpringBootStreamHandler(NoCatalogFluxFunctionConfig.class);
this.handler.initialize(null);
ByteArrayOutputStream output = new ByteArrayOutputStream();
this.handler.handleRequest(
new ByteArrayInputStream("{\"value\":\"foo\"}".getBytes()), output, null);
assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}");
}
@Test
public void typelessFunctionConfig() throws Exception {
this.handler = new SpringBootStreamHandler(TypelessFunctionConfig.class);
this.handler.initialize(null);
ByteArrayOutputStream output = new ByteArrayOutputStream();
this.handler.handleRequest(
new ByteArrayInputStream("{\"value\":\"foo\"}".getBytes()), output, null);
assertThat(output.toString()).isEqualTo("{\"value\":\"foo\"}");
}
@Test
public void inputStreamFunctionConfig() throws Exception {
this.handler = new SpringBootStreamHandler(InputStreamFunctionConfig.class);
this.handler.initialize(null);
ByteArrayOutputStream output = new ByteArrayOutputStream();
this.handler.handleRequest(
new ByteArrayInputStream("{\"value\":\"foo\"}".getBytes()), output, null);
assertThat(output.toString()).isEqualTo("{\"value\":\"FOO\"}");
}
@Configuration
protected static class NoCatalogNonFluxFunctionConfig {
@Bean
public Function<Foo, Bar> function() {
return foo -> new Bar(foo.getValue().toUpperCase());
}
}
@Configuration
protected static class NoCatalogFluxFunctionConfig {
@Bean
public Function<Flux<Foo>, Flux<Bar>> function() {
return flux -> flux.map(foo -> new Bar(foo.getValue().toUpperCase()));
}
}
@Configuration
@Import({ ContextFunctionCatalogAutoConfiguration.class,
JacksonAutoConfiguration.class })
protected static class FunctionConfigWithJackson {
@Bean
public Function<Foo, Bar> function() {
return foo -> new Bar(foo.getValue().toUpperCase());
}
}
@Configuration
@Import({ ContextFunctionCatalogAutoConfiguration.class })
protected static class FunctionConfigWithoutJackson {
@Bean
public Function<Foo, Bar> function() {
return foo -> new Bar(foo.getValue().toUpperCase());
}
}
@Configuration
@Import({ ContextFunctionCatalogAutoConfiguration.class,
JacksonAutoConfiguration.class })
protected static class TypelessFunctionConfig {
@Bean
public Function<?, ?> function() {
return value -> {
Assert.isTrue(value instanceof Map, "Expected value should be Map");
return value;
};
}
}
@Configuration
@Import({ ContextFunctionCatalogAutoConfiguration.class,
JacksonAutoConfiguration.class })
protected static class InputStreamFunctionConfig {
@Autowired
private ObjectMapper mapper;
@Bean
public Function<InputStream, ?> function() {
return value -> {
try {
Foo foo = this.mapper.readValue((InputStream) value, Foo.class);
return new Bar(foo.getValue().toUpperCase());
}
catch (Exception e) {
throw new IllegalStateException("Failed test", e);
}
};
}
}
protected static class Foo {
private String value;
public String getValue() {
return this.value;
}
public void setValue(String value) {
this.value = value;
}
}
protected static class Bar {
private String value;
public Bar() {
}
public Bar(String value) {
this.value = value;
}
public String getValue() {
return this.value;
}
public void setValue(String value) {
this.value = value;
}
}
}

1
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionalSpringApplication.java
View File

@@ -96,6 +96,7 @@ public class FunctionalSpringApplication
Assert.isInstanceOf(GenericApplicationContext.class, context,
"ApplicationContext must be an instanceof GenericApplicationContext");
for (Object source : getAllSources()) {
System.out.println("======> SOURCE: " + source);
Class<?> type = null;
Object handler = null;
if (source instanceof String) {

4
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java
View File

@@ -926,6 +926,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
if (inputValue instanceof Message && !this.isInputTypeMessage()) {
inputValue = ((Message) inputValue).getPayload();
}
System.out.println("Invoking function: " + this + "with input type: " + this.getInputType());
Object result = ((Function) this.target).apply(inputValue);
if (result instanceof Publisher && functionInvocationHelper != null) {
@@ -1096,7 +1097,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
: convertedInput;
}
if (convertedInput != null && logger.isDebugEnabled()) {
logger.debug("Converted Message: " + input + " to: " + convertedInput);
logger.debug("Converted Message: " + input + " to: " +
(convertedInput instanceof OriginalMessageHolder ? ((OriginalMessageHolder) convertedInput).value.getClass() : convertedInput));
}
}
else {

1
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogInitializer.java
View File

@@ -176,6 +176,7 @@ public class ContextFunctionCatalogInitializer implements ApplicationContextInit
List<MessageConverter> messageConverters = new ArrayList<>();
JsonMapper jsonMapper = this.context.getBean(JsonMapper.class);
messageConverters.addAll(context.getBeansOfType(MessageConverter.class).values());
messageConverters.add(new JsonMessageConverter(jsonMapper));
messageConverters.add(new ByteArrayMessageConverter());
messageConverters.add(new StringMessageConverter());

8
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java
View File

@@ -92,7 +92,13 @@ public class JsonMessageConverter extends AbstractMessageConverter {
if (payload instanceof byte[]) {
payload = new String((byte[]) payload, StandardCharsets.UTF_8);
}
logger.warn("Failed to convert value: " + payload, e);
if (logger.isDebugEnabled()) {
logger.debug("Failed to convert value: " + payload + " to: " + targetClass, e);
}
else {
logger.warn("Failed to convert value: " + payload + " to: " + targetClass);
}
}
}
}

7
spring-cloud-function-samples/function-functional-sample-aws/src/main/java/example/FunctionConfiguration.java
View File

@@ -8,6 +8,9 @@ import org.springframework.cloud.function.context.FunctionType;
import org.springframework.context.ApplicationContextInitializer;
import org.springframework.context.support.GenericApplicationContext;
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent;
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyResponseEvent;
@SpringBootConfiguration
public class FunctionConfiguration implements ApplicationContextInitializer<GenericApplicationContext> {
@@ -26,5 +29,9 @@ public class FunctionConfiguration implements ApplicationContextInitializer<Gene
context.registerBean("uppercase", FunctionRegistration.class,
() -> new FunctionRegistration<>(function).type(
FunctionType.from(String.class).to(String.class)));
context.registerBean("testFunction", FunctionRegistration.class,
() -> new FunctionRegistration<>(new TestFunction()).type(
FunctionType.from(APIGatewayProxyRequestEvent.class).to(APIGatewayProxyResponseEvent.class)));
}
}

13
spring-cloud-function-samples/function-functional-sample-aws/src/main/java/example/TestFunction.java Normal file
View File

@@ -0,0 +1,13 @@
package example;
import java.util.function.Function;
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent;
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyResponseEvent;
public class TestFunction implements Function<APIGatewayProxyRequestEvent, APIGatewayProxyResponseEvent> {
@Override
public APIGatewayProxyResponseEvent apply(APIGatewayProxyRequestEvent apiGatewayProxyRequestEvent) {
return new APIGatewayProxyResponseEvent().withStatusCode(200).withBody("ok");
}
}