From 20f738b1e9f98dc948eda57bcd555d7de1b6c943 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 Resolves #2401 --- .../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; }