diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index f82582ad2..226745dc6 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -639,6 +639,25 @@ public class StreamBridgeTests { } } + @Test + void testDynamicDestinationDestroy() { + BindingService bindingService; + 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"); + + bindingService = context.getBean(BindingService.class); + assertThat(bindingService.getProducerBindingNames().length).isEqualTo(1); + assertThat(bindingService.getProducerBindingNames()[0]).isEqualTo("a"); + } + assertThat(bindingService.getProducerBindingNames().length).isEqualTo(0); + } + @Test void testWithIntegrationFlowBecauseMarcinSaidSo() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration 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 c27f68849..df85dc557 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 @@ -26,6 +26,7 @@ import java.util.function.Function; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionRegistration; @@ -77,7 +78,7 @@ import org.springframework.util.StringUtils; * */ @SuppressWarnings("rawtypes") -public final class StreamBridge implements StreamOperations, SmartInitializingSingleton { +public final class StreamBridge implements StreamOperations, SmartInitializingSingleton, DisposableBean { private static final String STREAM_BRIDGE_FUNC_NAME = "streamBridge"; @@ -297,4 +298,11 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi this.applicationContext.getBean(GlobalChannelInterceptorProcessor.class); globalChannelInterceptorProcessor.postProcessAfterInitialization(messageChannel, destinationName); } + + @Override + public void destroy() throws Exception { + channelCache.keySet().forEach(bindingService::unbindProducers); + channelCache.clear(); + } + }