GH-2452 Clean up bindings in Stream bridge once they are removed from cache

Resolves #2452

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 <T> Binding<T> doBindProducer(T output, String bindingTarget,
 			Binder<T, ?, ProducerProperties> binder,
 			ProducerProperties producerProperties) {

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -235,7 +235,7 @@ public class FunctionConfiguration {
 							PollableBean pollable = null;
 							try {
 								pollable = extractPollableAnnotation(functionProperties, context, proxyFactory);
-							} 
+							}
 							catch (Exception e) {
 								// Will fix itself once https://github.com/spring-projects/spring-framework/issues/28748 is fixed
 							}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java

@@ -21,7 +21,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;
@@ -122,8 +121,11 @@ public final class StreamBridge implements SmartInitializingSingleton {
 			@Override
 			protected boolean removeEldestEntry(Map.Entry<String, MessageChannel> eldest) {
 				boolean remove = size() > bindingServiceProperties.getDynamicDestinationCacheSize();
-				if (remove && logger.isDebugEnabled()) {
+				if (remove) {
-					logger.debug("Removing message channel from cache " + eldest.getKey());
+					if (logger.isDebugEnabled()) {
+						logger.debug("Removing message channel from cache " + eldest.getKey());
+					}
+					bindingService.unbindProducers(eldest.getKey());
 				}
 				return remove;
 			}
@@ -261,11 +263,11 @@ public final class StreamBridge implements SmartInitializingSingleton {
 		Type functionType = ResolvableType.forClassWithGenerics(Function.class, Object.class, Object.class).getType();
 		this.functionRegistry.register(fr.type(functionType));
 		Map<String, DirectWithAttributesChannel> channels = applicationContext.getBeansOfType(DirectWithAttributesChannel.class);
-		for (Entry<String, DirectWithAttributesChannel> channelEntry : channels.entrySet()) {
+//		for (Entry<String, DirectWithAttributesChannel> channelEntry : channels.entrySet()) {
-			if (channelEntry.getValue().getAttribute("type").equals("output")) {
+//			if (channelEntry.getValue().getAttribute("type").equals("output")) {
-				this.channelCache.put(channelEntry.getKey(), channelEntry.getValue());
+//				this.channelCache.put(channelEntry.getKey(), channelEntry.getValue());
-			}
+//			}
-		}
+//		}
 		this.initialized = true;
 	}
 
@@ -304,7 +306,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);
 				}

core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java

@@ -44,6 +44,7 @@ import org.springframework.cloud.function.context.message.MessageUtils;
 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.BindingService;
 import org.springframework.cloud.stream.binding.NewDestinationBindingCallback;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.context.ConfigurableApplicationContext;
@@ -273,7 +274,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");
@@ -287,6 +289,27 @@ public class StreamBridgeTests {
 		}
 	}
 
+	@Test
+	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
 	void testWithInterceptorsRegisteredOnlyOnOutputChannel() throws InterruptedException {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration