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 19b867cbf5
commit 8ab9e37c9a
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

@@ -218,7 +218,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()
@@ -299,10 +300,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;
}