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

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.

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() +

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<byte[]> inputMessage = MessageBuilder
-					.withPayload("Hello".getBytes()).build();
+					.withPayload("Hello".getBytes())
+					.build();
 			inputDestination.send(inputMessage);
 
 			Message<byte[]> 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<byte[]> 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<AbstractSubscribableChannel> outputChannels = (List<AbstractSubscribableChannel>) 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<String> consume() {
+			return System.out::println;
+		}
+		@Bean
+		public Function<String, String> echo() {
+			return x -> x;
+		}
+	}
 
 	@EnableAutoConfiguration
 	public static class RoutingFunctionConfiguration  {