GH-1924 Add support for dynamic destinations to StramBridge

Resolves #1924

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -48,7 +48,6 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.FunctionProperties;
@@ -122,7 +121,7 @@ public class FunctionConfiguration {
 	private final static String SOURCE_PROPERY = "spring.cloud.stream.source";
 
 	@Bean
-	@ConditionalOnProperty(SOURCE_PROPERY)
+//	@ConditionalOnProperty(SOURCE_PROPERY)
 	public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry,
 			BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext) {
 		return new StreamBridge(functionCatalog, functionRegistry, bindingServiceProperties, applicationContext);
@@ -188,7 +187,7 @@ public class FunctionConfiguration {
 							String integrationFlowName = proxyFactory.getFunctionDefinition() + "_integrationflow";
 							PollableBean pollable = extractPollableAnnotation(functionProperties, context, proxyFactory);
 
-							Type functionType = ((FunctionInvocationWrapper) functionWrapper).getFunctionType();
+							Type functionType = functionWrapper.getFunctionType();
 							IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(functionWrapper, context, producerProperties),
 									beginPublishingTrigger, pollable, context, taskScheduler, functionType)
 									.route(Message.class, message -> {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java

@@ -21,19 +21,24 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.function.Function;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.FunctionRegistration;
 import org.springframework.cloud.function.context.FunctionRegistry;
 import org.springframework.cloud.function.context.FunctionType;
 import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper;
 import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.messaging.Message;
-import org.springframework.util.Assert;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
@@ -55,7 +60,9 @@ import org.springframework.util.MimeTypeUtils;
  */
 public final class StreamBridge implements SmartInitializingSingleton {
 
-	private final Map<String, DirectWithAttributesChannel> outputChannelsOnly = new HashMap<>();
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	private final Map<String, MessageChannel> outputChannelsOnly = new HashMap<>();
 
 	private final FunctionCatalog functionCatalog;
 
@@ -67,6 +74,9 @@ public final class StreamBridge implements SmartInitializingSingleton {
 
 	private boolean initialized;
 
+	@Autowired
+	private BinderAwareChannelResolver dynamicDestinationResolver;
+
 	/**
 	 *
 	 * @param functionCatalog instance of {@link FunctionCatalog}
@@ -97,6 +107,11 @@ public final class StreamBridge implements SmartInitializingSingleton {
 	 * Sends 'data' to an output binding specified by 'bindingName' argument while
 	 * using the content type specified by the 'outputContentType' argument to deal
 	 * with output type conversion (if necessary).
+	 * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property.
+	 * However, this operation also supports sending to dynamic destinations. This means if the name
+	 * provided via 'bindingName' does not have a corresponding binding such name will be
+	 * treated as dynamic destination.
+	 *
 	 * @param bindingName the name of the output binding
 	 * @param data the data to send
 	 * @param outputContentType content type to be used to deal with output type conversion
@@ -104,7 +119,15 @@ public final class StreamBridge implements SmartInitializingSingleton {
 	 */
 	@SuppressWarnings("unchecked")
 	public boolean send(String bindingName, Object data, MimeType outputContentType) {
-		Assert.isTrue(this.outputChannelsOnly.containsKey(bindingName), "Binding name '" + bindingName + "' does not exist.");
+		if (!this.outputChannelsOnly.containsKey(bindingName)) {
+			logger.info("Binding name '" + bindingName + "' does not exist. This means that value '"
+					+ bindingName + "' will be treated as dynamic destination. If this is not your intention please "
+							+ "provide 'spring.cloud.stream.source' property");
+			this.outputChannelsOnly.put(bindingName, dynamicDestinationResolver.resolveDestination(bindingName));
+			FunctionRegistration<Function<Object, Object>> fr = new FunctionRegistration<>(v -> v, bindingName);
+			this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message()));
+		}
+
 		FunctionInvocationWrapper functionWrapper = this.functionCatalog.lookup(bindingName, outputContentType.toString());
 
 		BindingProperties bindingProperties = this.bindingServiceProperties.getBindings().get(bindingName);

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java

@@ -92,7 +92,7 @@ public class StreamBridgeTests {
 			OutputDestination outputDestination = context.getBean(OutputDestination.class);
 
 
-			Message message = outputDestination.receive(100, "foo-out-0");
+			Message<?> message = outputDestination.receive(100, "foo-out-0");
 			assertThat(message.getPayload()).isEqualTo("hello foo".getBytes());
 			assertThat(message.getHeaders().get("foo")).isEqualTo("foo");
 
@@ -146,6 +146,30 @@ public class StreamBridgeTests {
 		}
 	}
 
+	@Test
+	public void testDynamicDestination() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(TestConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
+
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			StreamBridge bridge = context.getBean(StreamBridge.class);
+			bridge.send("foo-out-0", "b");
+			bridge.send("bar", "hello");
+			bridge.send("blah", MessageBuilder.withPayload("message").setHeader("foo", "foo").build());
+
+			Message<byte[]> message = outputDestination.receive(100, "foo-out-0");
+			assertThat(new String(message.getPayload())).isEqualTo("b");
+
+			message = outputDestination.receive(100, "bar");
+			assertThat(new String(message.getPayload())).isEqualTo("hello");
+
+			message = outputDestination.receive(100, "blah");
+			assertThat(new String(message.getPayload())).isEqualTo("message");
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class EmptyConfiguration {