SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1978 Fix dynamic destination resolutino for reactive function

Browse Source 
Resolves #1978
This commit is contained in:
Oleg Zhurakousky
2020-06-04 09:50:59 +02:00
parent 2337646a66
commit a8f8a1659e
2 changed files with 52 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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();
});

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

@@ -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 {