GH-929 Ensure AWS Functioininvoker handles Mono<Void> return the same way as imperative Consumer

Resolves #929
This commit is contained in:
Oleg Zhurakousky
2022-10-19 10:51:55 +02:00
parent 47f3964435
commit b44984f59c
2 changed files with 34 additions and 10 deletions

View File

@@ -90,28 +90,31 @@ public class FunctionInvoker implements RequestStreamHandler {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private byte[] buildResult(Message<?> requestMessage, Object output) throws IOException { private byte[] buildResult(Message<?> requestMessage, Object output) throws IOException {
Message<byte[]> responseMessage; Message<byte[]> responseMessage = null;
if (output instanceof Publisher<?>) { if (output instanceof Publisher<?>) {
List<Object> result = new ArrayList<>(); List<Object> result = new ArrayList<>();
for (Object value : Flux.from((Publisher<?>) output).toIterable()) { for (Object value : Flux.from((Publisher<?>) output).toIterable()) {
if (logger.isInfoEnabled()) { if (logger.isDebugEnabled()) {
logger.info("Response value: " + value); logger.debug("Response value: " + value);
} }
result.add(value); result.add(value);
} }
if (result.size() > 1) { if (result.size() > 1) {
output = result; output = result;
} }
else { else if (result.size() == 1) {
output = result.get(0); output = result.get(0);
} }
else {
if (logger.isInfoEnabled()) { output = null;
logger.info("OUTPUT: " + output + " - " + output.getClass().getName()); }
if (output != null) {
if (logger.isDebugEnabled()) {
logger.debug("OUTPUT: " + output + " - " + output.getClass().getName());
}
byte[] payload = this.jsonMapper.toJson(output);
responseMessage = MessageBuilder.withPayload(payload).build();
} }
byte[] payload = this.jsonMapper.toJson(output);
responseMessage = MessageBuilder.withPayload(payload).build();
} }
else { else {
responseMessage = (Message<byte[]>) output; responseMessage = (Message<byte[]>) output;

View File

@@ -44,6 +44,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.function.json.JsonMapper; import org.springframework.cloud.function.json.JsonMapper;
@@ -1024,6 +1025,21 @@ public class FunctionInvokerTests {
assertThat(result.get("body")).isEqualTo("\"OK\""); assertThat(result.get("body")).isEqualTo("\"OK\"");
} }
@SuppressWarnings("rawtypes")
@Test
public void testApiGatewayWithMonoVoidAsReturn() throws Exception {
System.setProperty("MAIN_CLASS", ApiGatewayConfiguration.class.getName());
System.setProperty("spring.cloud.function.definition", "reactiveWithVoidReturn");
FunctionInvoker invoker = new FunctionInvoker();
InputStream targetStream = new ByteArrayInputStream(this.apiGatewayEvent.getBytes());
ByteArrayOutputStream output = new ByteArrayOutputStream();
invoker.handleRequest(targetStream, output, null);
Map result = mapper.readValue(output.toByteArray(), Map.class);
assertThat(result.get("body")).isEqualTo("\"OK\"");
}
@Test @Test
public void testWithDefaultRoutingFailure() throws Exception { public void testWithDefaultRoutingFailure() throws Exception {
System.setProperty("MAIN_CLASS", SampleConfiguration.class.getName()); System.setProperty("MAIN_CLASS", SampleConfiguration.class.getName());
@@ -1315,6 +1331,11 @@ public class FunctionInvokerTests {
return v -> v.toUpperCase(); return v -> v.toUpperCase();
} }
@Bean
public Function<Mono<String>, Mono<Void>> reactiveWithVoidReturn() {
return v -> Mono.empty();
}
@Bean @Bean
public Function<Person, String> uppercasePojo() { public Function<Person, String> uppercasePojo() {
return v -> { return v -> {