From b8a70edc21603cac138c0b0e6bde93c4a37c40e1 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 18 Jan 2022 11:51:22 +0100 Subject: [PATCH] GH-2268 Ensure StreamBridge works with nullChannel Resolves #2268 --- .../stream/function/FunctionConfiguration.java | 2 +- .../cloud/stream/function/StreamBridge.java | 16 ++++++++-------- .../cloud/stream/function/StreamBridgeTests.java | 12 ++++++++++++ 3 files changed, 21 insertions(+), 9 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 12ccab77d..04178e058 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -601,7 +601,7 @@ public class FunctionConfiguration { private void doSendMessage(Object result, Message requestMessage) { if (result instanceof Message && ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination") != null) { String destinationName = (String) ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination"); - SubscribableChannel outputChannel = streamBridge.resolveDestination(destinationName, producerProperties, null); + MessageChannel outputChannel = streamBridge.resolveDestination(destinationName, producerProperties, null); if (logger.isInfoEnabled()) { logger.info("Output message is sent to '" + destinationName + "' destination"); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 9ea88d556..2d259e07b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -47,7 +47,7 @@ import org.springframework.integration.config.GlobalChannelInterceptorProcessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.MessageChannel; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; import org.springframework.util.StringUtils; @@ -76,7 +76,7 @@ public final class StreamBridge implements SmartInitializingSingleton { private final Log logger = LogFactory.getLog(getClass()); - private final Map channelCache; + private final Map channelCache; private final FunctionCatalog functionCatalog; @@ -111,9 +111,9 @@ public final class StreamBridge implements SmartInitializingSingleton { this.applicationContext = applicationContext; this.bindingServiceProperties = bindingServiceProperties; this.destinationBindingCallback = destinationBindingCallback; - this.channelCache = new LinkedHashMap() { + this.channelCache = new LinkedHashMap() { @Override - protected boolean removeEldestEntry(Map.Entry eldest) { + protected boolean removeEldestEntry(Map.Entry eldest) { boolean remove = size() > bindingServiceProperties.getDynamicDestinationCacheSize(); if (remove && logger.isDebugEnabled()) { logger.debug("Removing message channel from cache " + eldest.getKey()); @@ -207,7 +207,7 @@ public final class StreamBridge implements SmartInitializingSingleton { data = MessageBuilder.withPayload(data).build(); } ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); - SubscribableChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName); + MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName); Function functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties); @@ -252,10 +252,10 @@ public final class StreamBridge implements SmartInitializingSingleton { } @SuppressWarnings({ "unchecked", "rawtypes"}) - synchronized SubscribableChannel resolveDestination(String destinationName, ProducerProperties producerProperties, String binderName) { - SubscribableChannel messageChannel = this.channelCache.get(destinationName); + synchronized MessageChannel resolveDestination(String destinationName, ProducerProperties producerProperties, String binderName) { + MessageChannel messageChannel = this.channelCache.get(destinationName); if (messageChannel == null && this.applicationContext.containsBean(destinationName)) { - messageChannel = this.applicationContext.getBean(destinationName, SubscribableChannel.class); + messageChannel = this.applicationContext.getBean(destinationName, MessageChannel.class); } if (messageChannel == null) { messageChannel = new DirectWithAttributesChannel(); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index 546538440..47d70d09f 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -186,6 +186,18 @@ public class StreamBridgeTests { } } + @Test // validate that there is no exception thrown when sending to null channel + public void test_2268() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration + .getCompleteConfiguration(InterceptorConfiguration.class)) + .web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false")) { + StreamBridge bridge = context.getBean(StreamBridge.class); + + bridge.send("nullChannel", "blah"); + } + } + @Test public void testInterceptorIsNotAddedMultipleTimesToTheMessageChannel() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration