GH-1978 Fix dynamic destination resolutino for reactive function

Resolves #1978

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

@@ -375,10 +375,12 @@ public class FunctionConfiguration {
 				}
 
 				Object resultPublishers = functionToInvoke.apply(inputPublishers.length == 1 ? inputPublishers[0] : Tuples.fromArray(inputPublishers));
+
 				if (!(resultPublishers instanceof Iterable)) {
 					resultPublishers = Collections.singletonList(resultPublishers);
 				}
 				Iterator<String> outputBindingIter = outputBindingNames.iterator();
+
 				((Iterable) resultPublishers).forEach(publisher -> {
 					Flux flux = Flux.from((Publisher) publisher)
 							.onErrorContinue((ex, pay) -> {
@@ -386,7 +388,20 @@ public class FunctionConfiguration {
 							});
 					if (!CollectionUtils.isEmpty(outputBindingNames)) {
 						MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class);
-						flux = flux.doOnNext(message -> outputChannel.send((Message) message));
+						flux = flux.doOnNext(message -> {
+							if (message instanceof Message && ((Message<?>) message).getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
+								String destinationName = (String) ((Message<?>) message).getHeaders().get("spring.cloud.stream.sendto.destination");
+								ProducerProperties producerProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()).getProducer();
+								MessageChannel dynamicChannel = streamBridge.resolveDestination(destinationName, producerProperties);
+								if (logger.isInfoEnabled()) {
+									logger.info("Output message is sent to '" + destinationName + "' destination");
+								}
+								dynamicChannel.send((Message) message);
+							}
+							else {
+								outputChannel.send((Message) message);
+							}
+						});
 					}
 					flux.subscribe();
 				});

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

@@ -733,6 +733,32 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@Test
+	public void testReactiveSendToDestinationConfiguration() {
+		System.clearProperty("spring.cloud.function.definition");
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(SendToDestinationConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+			Message<byte[]> inputMessage = MessageBuilder.withPayload("aa".getBytes()).build();
+			inputDestination.send(inputMessage, "echo-in-0");
+			Message<byte[]> receivedMessage = outputDestination.receive(1000, "aa");
+
+			assertThat(receivedMessage.getPayload()).isEqualTo("aa".getBytes());
+			assertThat(receivedMessage.getHeaders().get("spring.cloud.stream.sendto.destination")).isNotNull();
+
+			inputMessage = MessageBuilder.withPayload("bb".getBytes()).build();
+			inputDestination.send(inputMessage, "echo-in-0");
+			receivedMessage = outputDestination.receive(1000, "bb");
+
+			assertThat(receivedMessage.getPayload()).isEqualTo("bb".getBytes());
+			assertThat(receivedMessage.getHeaders().get("spring.cloud.stream.sendto.destination")).isNotNull();
+		}
+	}
+
+
 	@EnableAutoConfiguration
 	public static class NoEnableBindingConfiguration {
 
@@ -859,6 +885,16 @@ public class ImplicitFunctionBindingTests {
 
 	}
 
+	@EnableAutoConfiguration
+	public static class SendToDestinationConfiguration {
+		@Bean
+		public Function<Flux<String>, Flux<Message<String>>> echo() {
+			return flux -> flux.map(v -> {
+				return MessageBuilder.withPayload(v).setHeader("spring.cloud.stream.sendto.destination", v).build();
+			});
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class SCF_GH_409Configuration {