From 562e1b27ba4c9e824b613095969f14e81a7b4922 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 16 May 2022 15:44:44 -0400 Subject: [PATCH] StreamBridge binder name related changes 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 --- .../cloud/stream/function/StreamBridge.java | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 3ac90458c..8671b79f5 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -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; }