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 ba052db65..c792a09bf 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/FunctionConfiguration.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 732f66232..34af48efe 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/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 } 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 3b1287f7c..947258d0a 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 @@ -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 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; } @@ -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 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; } @@ -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); } 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 50f88b36d..fcff7f46e 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 @@ -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