From 34bde8f15baea83b48cfd558fe298dd1278ee6c0 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 13 Apr 2023 13:52:24 +0200 Subject: [PATCH] GH-2707 Add support for function post processing --- .../function/FunctionPostProcessingTests.java | 245 ++++++++++++++++++ .../function/FunctionConfiguration.java | 2 + 2 files changed, 247 insertions(+) create mode 100644 core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/FunctionPostProcessingTests.java diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/FunctionPostProcessingTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/FunctionPostProcessingTests.java new file mode 100644 index 000000000..9ba017000 --- /dev/null +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/FunctionPostProcessingTests.java @@ -0,0 +1,245 @@ +/* + * Copyright 2023-2023 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.stream.function; + +import java.util.function.Function; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.function.context.PostProcessingFunction; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * + * @author Oleg Zhurakousky + * + */ +public class FunctionPostProcessingTests { + + @Test + void testNothingIsBroken() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(PostProcessingTestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=echo")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder.withPayload("hello".getBytes()).build(); + inputDestination.send(inputMessage); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + assertThat(outputDestination.receive().getPayload()).isEqualTo("hello".getBytes()); + } + } + + @Test + void testSuccessfulPostProcessingOfSingleFunction() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(PostProcessingTestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=uppercase")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder.withPayload("hello".getBytes()).build(); + inputDestination.send(inputMessage); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + assertThat(outputDestination.receive().getPayload()).isEqualTo("HELLO".getBytes()); + assertThat(context.getBean(SingleFunctionPostProcessingFunction.class).success).isTrue(); + } + } + + @Test + void testNoPostProcessingOnError() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(PostProcessingTestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=uppercase")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder.withPayload("error".getBytes()).build(); + inputDestination.send(inputMessage); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + assertThat(outputDestination.receive()).isNull(); + assertThat(context.getBean(SingleFunctionPostProcessingFunction.class).success).isFalse(); + } + } + + @Test + void testNoFailureOnPostProcessingError() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(PostProcessingTestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=uppercase")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder.withPayload("post_processing_error".getBytes()).build(); + inputDestination.send(inputMessage); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + assertThat(outputDestination.receive().getPayload()).isEqualTo("POST_PROCESSING_ERROR".getBytes()); + assertThat(context.getBean(SingleFunctionPostProcessingFunction.class).success).isFalse(); + } + } + + + @Test + void testWithCompositionLastFunctionIsPostProcessing() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(PostProcessingTestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=echo|uppercase")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder.withPayload("hello".getBytes()).build(); + inputDestination.send(inputMessage); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + assertThat(outputDestination.receive().getPayload()).isEqualTo("HELLO".getBytes()); + assertThat(context.getBean(SingleFunctionPostProcessingFunction.class).success).isTrue(); + } + } + + @Test + void testWithCompositionFirstFunctionIsPostProcessing() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(PostProcessingTestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=uppercase|echo")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder.withPayload("hello".getBytes()).build(); + inputDestination.send(inputMessage); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + assertThat(outputDestination.receive().getPayload()).isEqualTo("HELLO".getBytes()); + assertThat(context.getBean(SingleFunctionPostProcessingFunction.class).success).isFalse(); + } + } + + @Test + void testOlnyLastPostProcessorInvoked() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(PostProcessingTestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=echo|uppercase|reverse")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder.withPayload("hello".getBytes()).build(); + inputDestination.send(inputMessage); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + assertThat(outputDestination.receive().getPayload()).isEqualTo("OLLEH".getBytes()); + assertThat(context.getBean(SingleFunctionPostProcessingFunction.class).success).isFalse(); + assertThat(context.getBean(SingleFunctionPostProcessingFunction2.class).success).isTrue(); + } + } + + @Test + void testOlnyLastPostProcessorInvoked2() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(PostProcessingTestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=uppercase|echo|reverse")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder.withPayload("hello".getBytes()).build(); + inputDestination.send(inputMessage); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + assertThat(outputDestination.receive().getPayload()).isEqualTo("OLLEH".getBytes()); + assertThat(context.getBean(SingleFunctionPostProcessingFunction.class).success).isFalse(); + assertThat(context.getBean(SingleFunctionPostProcessingFunction2.class).success).isTrue(); + } + } + + + @EnableAutoConfiguration + public static class PostProcessingTestConfiguration { + + @Bean + public Function echo() { + return x -> x; + } + + @Bean + public Function uppercase() { + return new SingleFunctionPostProcessingFunction(); + } + + @Bean + public Function reverse() { + return new SingleFunctionPostProcessingFunction2(); + } + } + + private static class SingleFunctionPostProcessingFunction implements PostProcessingFunction { + + private boolean success; + + @Override + public String apply(String input) { + if (input.equals("error")) { + throw new RuntimeException("intentional"); + } + return input.toUpperCase(); + } + + @Override + public void postProcess(Message result) { + if (result.getPayload().equals("POST_PROCESSING_ERROR")) { + throw new RuntimeException("intentional"); + } + success = true; + } + } + + private static class SingleFunctionPostProcessingFunction2 implements PostProcessingFunction { + + private boolean success; + + @Override + public String apply(String input) { + return new StringBuilder(input).reverse().toString(); + } + + @Override + public void postProcess(Message result) { + success = true; + } + } +} diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index e0ed79821..cf3c83430 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -614,6 +614,7 @@ public class FunctionConfiguration { MessagingTemplate template = new MessagingTemplate(); template.setBeanFactory(applicationContext.getBeanFactory()); + AbstractMessageHandler handler = new AbstractMessageHandler() { @SuppressWarnings("unchecked") @Override @@ -659,6 +660,7 @@ public class FunctionConfiguration { } streamBridge.send(function.getFunctionDefinition() + "-out-0", result); } + function.postProcess(); } };