From 9d14b050775bdad956ff9cd3ee9d091f379d66e6 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 25 Jul 2022 16:37:54 +0200 Subject: [PATCH] GH-2452 Clean up bindings in Stream bridge once they are removed from cache Resolves #2452 polish --- .../cloud/stream/binding/BindingService.java | 8 ++++++ .../cloud/stream/function/StreamBridge.java | 20 ++++++++------- .../stream/function/StreamBridgeTests.java | 25 ++++++++++++++++++- 3 files changed, 43 insertions(+), 10 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index 938874841..aa9441496 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java @@ -316,6 +316,14 @@ public class BindingService { return null; } + public String[] getProducerBindingNames() { + return this.producerBindings.keySet().toArray(new String[] {}); + } + + public String[] getConsumerBindingNames() { + return this.consumerBindings.keySet().toArray(new String[] {}); + } + public Binding doBindProducer(T output, String bindingTarget, Binder binder, ProducerProperties producerProperties) { 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 13ab534fe..d3025a919 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 @@ -20,7 +20,6 @@ import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; -import java.util.Map.Entry; import java.util.function.Function; import org.apache.commons.logging.Log; @@ -120,8 +119,11 @@ public final class StreamBridge implements SmartInitializingSingleton { @Override protected boolean removeEldestEntry(Map.Entry eldest) { boolean remove = size() > bindingServiceProperties.getDynamicDestinationCacheSize(); - if (remove && logger.isDebugEnabled()) { - logger.debug("Removing message channel from cache " + eldest.getKey()); + if (remove) { + if (logger.isDebugEnabled()) { + logger.debug("Removing message channel from cache " + eldest.getKey()); + } + bindingService.unbindProducers(eldest.getKey()); } return remove; } @@ -257,11 +259,11 @@ public final class StreamBridge implements SmartInitializingSingleton { fr.getProperties().put("singleton", "false"); this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message())); Map channels = applicationContext.getBeansOfType(DirectWithAttributesChannel.class); - for (Entry channelEntry : channels.entrySet()) { - if (channelEntry.getValue().getAttribute("type").equals("output")) { - this.channelCache.put(channelEntry.getKey(), channelEntry.getValue()); - } - } +// for (Entry channelEntry : channels.entrySet()) { +// if (channelEntry.getValue().getAttribute("type").equals("output")) { +// this.channelCache.put(channelEntry.getKey(), channelEntry.getValue()); +// } +// } this.initialized = true; } @@ -300,7 +302,7 @@ public final class StreamBridge implements SmartInitializingSingleton { } this.addInterceptors((AbstractMessageChannel) messageChannel, destinationName); - this.bindingService.bindProducer(messageChannel, destinationName, false, binder); + this.bindingService.bindProducer(messageChannel, destinationName, true, binder); if (StringUtils.hasText(binderName)) { this.channelCache.put(binderName + ":" + destinationName, messageChannel); } diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index b96bf4458..3e7045650 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -42,6 +42,7 @@ import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver.NewDestinationBindingCallback; +import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; @@ -229,7 +230,8 @@ public class StreamBridgeTests { "--spring.cloud.stream.dynamic-destination-cache-size=1", "--spring.cloud.stream.output-bindings=outputA;outputB", "--spring.cloud.stream.bindings.outputA-out-0.destination=outputA", - "--spring.cloud.stream.bindings.outputB-out-0.destination=outputB")) { + "--spring.cloud.stream.bindings.outputB-out-0.destination=outputB" + )) { StreamBridge bridge = context.getBean(StreamBridge.class); bridge.send("outputA-out-0", "hello foo"); @@ -243,6 +245,27 @@ public class StreamBridgeTests { } } + @Test + public void testBindingsAreRemovedWithCache() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration + .getCompleteConfiguration(InterceptorConfiguration.class)) + .web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.stream.dynamic-destination-cache-size=1" + )) { + StreamBridge bridge = context.getBean(StreamBridge.class); + + bridge.send("a", "hello foo"); + bridge.send("b", "hello foo"); + bridge.send("c", "hello foo"); + bridge.send("d", "hello foo"); + + BindingService bindingService = context.getBean(BindingService.class); + assertThat(bindingService.getProducerBindingNames().length).isEqualTo(1); + assertThat(bindingService.getProducerBindingNames()[0]).isEqualTo("d"); + } + } + @Test public void testWithInterceptorsRegisteredOnlyOnOutputChannel() throws InterruptedException { try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration