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

StreamBridge binder name related changes

Browse Source 
An internal helper method used by StreamBridge for resolving the
binder target type to attach the target binder protocol, does not
take into account the binder name when resolving the binder configuration.
Fixing this issue.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2379
Resolves #2401
This commit is contained in:
Soby Chacko
2022-05-16 15:44:44 -04:00
committed by Oleg Zhurakousky
parent b8a5c1d3af
commit 20f738b1e9
1 changed files with 5 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -221,7 +221,8 @@ public final class StreamBridge implements SmartInitializingSingleton {
functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);
}
String targetType = this.resolveBinderTargetType(bindingName, MessageChannel.class, this.applicationContext.getBean(BinderFactory.class));
String targetType = this.resolveBinderTargetType(bindingName, binderName, MessageChannel.class,
this.applicationContext.getBean(BinderFactory.class));
Message<?> messageToSend = data instanceof Message
? MessageBuilder.fromMessage((Message) data).setHeaderIfAbsent(MessageUtils.TARGET_PROTOCOL, targetType).build()
@@ -305,10 +306,10 @@ public final class StreamBridge implements SmartInitializingSingleton {
return messageChannel;
}
private String resolveBinderTargetType(String channelName, Class<?> bindableType, BinderFactory binderFactory) {
String binderConfigurationName = this.bindingServiceProperties
private String resolveBinderTargetType(String channelName, String binderName, Class<?> bindableType, BinderFactory binderFactory) {
String binderConfigurationName = binderName != null ? binderName : this.bindingServiceProperties
.getBinder(channelName);
Binder binder = binderFactory.getBinder(binderConfigurationName, bindableType);
Binder<?, ?, ?> binder = binderFactory.getBinder(binderConfigurationName, bindableType);
String targetProtocol = binder.getClass().getSimpleName().startsWith("Rabbit") ? "amqp" : "kafka";
return targetProtocol;
}