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 {