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

GH-1837 Fix dynamic destination resolution for Supplier

Browse Source 
Resolves #1837
This commit is contained in:
Oleg Zhurakousky
2019-10-31 20:05:29 +01:00
parent b8de26b4f7
commit e2a41e7871
1 changed files with 9 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -141,7 +141,7 @@ public class FunctionConfiguration {
@Bean
InitializingBean supplierInitializer(FunctionCatalog functionCatalog, StreamFunctionProperties functionProperties,
GenericApplicationContext context, BindingServiceProperties serviceProperties,
@Nullable BindableFunctionProxyFactory[] proxyFactories) {
@Nullable BindableFunctionProxyFactory[] proxyFactories, BinderAwareChannelResolver dynamicDestinationResolver) {
if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class)) || proxyFactories == null) {
return null;
@@ -173,8 +173,15 @@ public class FunctionConfiguration {
if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) {
String integrationFlowName = proxyFactory.getFunctionDefinition() + "_integrationflow";
PollableBean pollable = extractPollableAnnotation(functionProperties, context, proxyFactory);
IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(functionWrapper, beginPublishingTrigger, pollable)
.channel(outputName).get();
.route(Message.class, message -> {
if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination");
return dynamicDestinationResolver.resolveDestination(destinationName);
}
return outputName;
}).get();
IntegrationFlow postProcessedFlow = (IntegrationFlow) context.getAutowireCapableBeanFactory()
.applyBeanPostProcessorsBeforeInitialization(integrationFlow, integrationFlowName);
context.registerBean(integrationFlowName, IntegrationFlow.class, () -> postProcessedFlow);