From 8ab9e37c9a2ef7ff7aafec05f063d84c6f233ceb 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 9a695c6a6..f6000de56 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 @@ -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; }