From b2b95dc7d292578c56ae42b066f437b0f1436e65 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 7 May 2021 15:21:59 +0200 Subject: [PATCH] GH-2168 Ensure output binding is not assumed for RoutingFunction Given that RoutingFunction may actualy route to Consumer, there is no need for output binding. This fix addresses it where no output binding is created initially and instead the output binding will be created if and when the firt output is produced. Resolves #2168 --- .../main/asciidoc/spring-cloud-stream.adoc | 16 ++++++ .../function/FunctionConfiguration.java | 17 +++---- .../stream/function/RoutingFunctionTests.java | 49 ++++++++++++++++++- 3 files changed, 72 insertions(+), 10 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 20b3d2cf4..827f5dd37 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1239,6 +1239,22 @@ public class RoutingStreamApplication { IMPORTANT: Passing instructions via application properties is especially important for reactive functions given that a reactive function is only invoked once to pass the Publisher, so access to the individual items is limited. +===== Routing Function and output binding + +`RoutingFunction` is a `Function` and as such treated no differently than any other function. Well. . . almost. + +When `RoutingFunction` routes to another `Function`, its output is sent to the output binding of the `RoutingFunction` which +is `functionRouter-in-0` as expected. But what if `RoutingFunction` routes to a `Consumer`? In other words the result of invocation +of the `RoutingFunction` may not produce anything to be sent to the output binding, thus making it necessary to even have one. +So, we do treat `RoutingFunction` a little bit differently when we create bindings. And even though it is transparent to you as a user +(there is really nothing for you to do), being aware of some of the mechanics would help you understand its inner workings. + +So, the rule is; +We never create output binding for the `RoutingFunction`, only input. So when you routing to `Consumer`, the `RoutingFunction` effectively +becomes as a `Consumer` by not having any output bindings. However, if `RoutingFunction` happen to route to another `Function` which produces +the output, the output binding for the `RoutingFunction` will be create dynamically at which point `RoutingFunction` will act as a regular `Function` +with regards to bindings (having both input and output bindings). + ==== Routing FROM Consumer Aside from static destinations, Spring Cloud Stream lets applications send messages to dynamically bound destinations. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index c63219c57..f5b8a6b99 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -592,6 +592,12 @@ public class FunctionConfiguration { } template.send(outputChannelName, (Message) result); } + else if (function.getFunctionDefinition().equals(RoutingFunction.FUNCTION_NAME)) { + if (!(result instanceof Message)) { + result = MessageBuilder.withPayload(result).copyHeadersIfAbsent(requestMessage.getHeaders()).build(); + } + streamBridge.send(RoutingFunction.FUNCTION_NAME + "-out-0", result); + } } }; @@ -797,7 +803,7 @@ public class FunctionConfiguration { this.inputCount = 0; this.outputCount = this.getOutputCount(functionType, true); } - else if (function.isConsumer()) { + else if (function.isConsumer() || functionDefinition.equals(RoutingFunction.FUNCTION_NAME)) { this.inputCount = FunctionTypeUtils.getInputCount(functionType); this.outputCount = 0; } @@ -810,14 +816,7 @@ public class FunctionConfiguration { functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount); functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount); functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties); - try { - String name = functionDefinition + "_binding"; - registry.registerBeanDefinition(name, functionBindableProxyDefinition); - } - catch (Exception e) { - e.printStackTrace(); - } - + registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition); } else { logger.warn("The function definition '" + streamFunctionProperties.getDefinition() + diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/RoutingFunctionTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/RoutingFunctionTests.java index 118733dd6..ef2ff54d1 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/RoutingFunctionTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/RoutingFunctionTests.java @@ -16,6 +16,9 @@ package org.springframework.cloud.stream.function; +import java.lang.reflect.Field; +import java.util.List; +import java.util.function.Consumer; import java.util.function.Function; import org.junit.Before; @@ -33,10 +36,12 @@ import org.springframework.cloud.stream.binder.test.TestChannelBinder; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.integration.channel.AbstractSubscribableChannel; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.util.MimeTypeUtils; +import org.springframework.util.ReflectionUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -267,7 +272,8 @@ public class RoutingFunctionTests { .getBean(OutputDestination.class); Message inputMessage = MessageBuilder - .withPayload("Hello".getBytes()).build(); + .withPayload("Hello".getBytes()) + .build(); inputDestination.send(inputMessage); Message outputMessage = outputDestination.receive(); @@ -297,7 +303,48 @@ public class RoutingFunctionTests { } } + @SuppressWarnings("unchecked") + @Test + public void testRoutingToConsumers() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + RoutingConsumerConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.routing-expression=headers['func_name']")) { + InputDestination inputDestination = context.getBean(InputDestination.class); + Message inputMessage = MessageBuilder + .withPayload("foo".getBytes()) + .setHeader("func_name", "consume") + .build(); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + Field chField = ReflectionUtils.findField(outputDestination.getClass(), "channels"); + chField.setAccessible(true); + List outputChannels = (List) chField.get(outputDestination); + assertThat(outputChannels.isEmpty()); + inputDestination.send(inputMessage); + assertThat(outputChannels.isEmpty()); + inputMessage = MessageBuilder + .withPayload("foo".getBytes()) + .setHeader("func_name", "echo") + .build(); + inputDestination.send(inputMessage); + assertThat(outputChannels.size()).isEqualTo(1); + } + } + + @EnableAutoConfiguration + public static class RoutingConsumerConfiguration { + @Bean + public Consumer consume() { + return System.out::println; + } + @Bean + public Function echo() { + return x -> x; + } + } @EnableAutoConfiguration public static class RoutingFunctionConfiguration {