diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index c2c54c658..56cc72a4d 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java @@ -90,12 +90,13 @@ import static org.mockito.Mockito.when; * @author Chris Bono * @author Artem Bilan * @author Kotaro Matsumoto + * @author Omer Celik */ class BindingServiceTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - void defaultGroup() throws Exception { + void defaultGroup() { BindingServiceProperties properties = new BindingServiceProperties(); Map bindingProperties = new HashMap<>(); BindingProperties props = new BindingProperties(); @@ -175,7 +176,7 @@ class BindingServiceTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - void multipleConsumerBindingsFromIndexList() throws Exception { + void multipleConsumerBindingsFromIndexList() { BindingServiceProperties properties = new BindingServiceProperties(); Map bindingProperties = new HashMap<>(); BindingProperties props = new BindingProperties(); @@ -236,7 +237,7 @@ class BindingServiceTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - void consumerBindingWhenMultiplexingIsEnabled() throws Exception { + void consumerBindingWhenMultiplexingIsEnabled() { BindingServiceProperties properties = new BindingServiceProperties(); Map bindingProperties = new HashMap<>(); BindingProperties props = new BindingProperties(); @@ -282,7 +283,7 @@ class BindingServiceTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - void explicitGroup() throws Exception { + void explicitGroup() { BindingServiceProperties properties = new BindingServiceProperties(); Map bindingProperties = new HashMap<>(); BindingProperties props = new BindingProperties(); @@ -528,7 +529,7 @@ class BindingServiceTests { assertThat(service.getProducerBinding("output")).isSameAs(binding); - service.unbindProducers(outputChannelName); + service.unbindProducers(null, outputChannelName); verify(binder, times(2)).bindProducer(eq("foo"), same(outputChannel), any(ProducerProperties.class)); verify(delegate).unbind(); @@ -552,9 +553,8 @@ class BindingServiceTests { assertThat(inputBinding.isRunning()).isFalse(); } - @SuppressWarnings("unchecked") @Test - void bindingNameAsTopLevelProperty() throws Exception { + void bindingNameAsTopLevelProperty() { ApplicationContext context = new SpringApplicationBuilder(BarConfiguration.class) .web(WebApplicationType.NONE).run(); 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 e55c8b055..df9f12809 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 @@ -84,6 +84,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; * * @author Oleg Zhurakousky * @author Soby Chacko + * @author Omer Celik * */ class StreamBridgeTests { @@ -384,7 +385,7 @@ class StreamBridgeTests { // See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream/issues/2805 @Test - void streamBridgeSendWithBinderNameAndCustomContentType() throws Exception { + void streamBridgeSendWithBinderNameAndCustomContentType() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration .getCompleteConfiguration(ConsumerConfiguration.class, EmptyConfigurationWithCustomConverters.class)) .web(WebApplicationType.NONE).run( @@ -767,6 +768,45 @@ class StreamBridgeTests { assertThat(bindingService.getProducerBindingNames().length).isEqualTo(0); } + @Test + void dynamicDestinationWithBinderNameDestroy() { + BindingService bindingService; + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration + .getCompleteConfiguration(InterceptorConfiguration.class)) + .web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.stream.binders.kafka1.type=kafka", + "--spring.cloud.stream.binders.anotherKafka.type=kafka" + )) { + StreamBridge bridge = context.getBean(StreamBridge.class); + bridge.send("binding1", "kafka1", "Omer Celik"); + bridge.send("binding2", "anotherKafka", "Omer Celik"); + + bindingService = context.getBean(BindingService.class); + } + assertThat(bindingService.getProducerBindingNames().length).isEqualTo(0); + } + + @Test + void dynamicDestinationWithBinderNameDestroyForCacheSize() { + 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", + "--spring.cloud.stream.binders.kafka1.type=kafka", + "--spring.cloud.stream.binders.anotherKafka.type=kafka" + )) { + StreamBridge bridge = context.getBean(StreamBridge.class); + bridge.send("binding1", "kafka1", "Omer Celik"); + bridge.send("binding2", "anotherKafka", "Omer Celik"); + + BindingService bindingService = context.getBean(BindingService.class); + assertThat(bindingService.getProducerBindingNames().length).isEqualTo(1); + assertThat(bindingService.getProducerBindingNames()[0]).isEqualTo("anotherKafka:binding2"); + } + } + @Test void withIntegrationFlowBecauseMarcinSaidSo() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderWrapper.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderWrapper.java new file mode 100644 index 000000000..b56a68e48 --- /dev/null +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderWrapper.java @@ -0,0 +1,25 @@ +/* + * Copyright 2024-2024 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + +/** + * Holds information about the binder and channel. + * + * @author Omer Celik + */ +public record BinderWrapper(Binder binder, String destinationName, String cacheKey) { +} diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java index 6f3ebc69d..d4daac35a 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java @@ -135,7 +135,7 @@ public class AbstractBindableProxyFactory implements Bindable { for (Map.Entry boundTargetHolderEntry : this.outputHolders .entrySet()) { if (boundTargetHolderEntry.getValue().bindable()) { - bindingService.unbindProducers(boundTargetHolderEntry.getKey()); + bindingService.unbindProducers(null, boundTargetHolderEntry.getKey()); } } } 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 1db90cde5..80c84a6c4 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 @@ -35,6 +35,7 @@ import org.springframework.aop.framework.Advised; import org.springframework.beans.BeanUtils; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binder.BinderWrapper; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; @@ -50,6 +51,9 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; import org.springframework.validation.DataBinder; +import static org.springframework.cloud.stream.utils.CacheKeyCreatorUtils.createChannelCacheKey; +import static org.springframework.cloud.stream.utils.CacheKeyCreatorUtils.getBinderNameIfNeeded; + /** * Handles binding of input/output targets by delegating to an underlying {@link Binder}. * @@ -270,47 +274,44 @@ public class BindingService { } @SuppressWarnings({ "unchecked", "rawtypes" }) - public Binding bindProducer(T output, String outputName, boolean cache, @Nullable Binder binder) { - String bindingTarget = this.bindingServiceProperties.getBindingDestination(outputName); - Class outputClass = output.getClass(); - if (output instanceof Advised advisedOutput) { - outputClass = Stream.of(advisedOutput.getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst() - .orElse(outputClass); - } - if (binder == null) { - binder = (Binder) getBinder(outputName, outputClass); - } - + public Binding bindProducer(T output, boolean cache, BinderWrapper binderWrapper) { ProducerProperties producerProperties = this.bindingServiceProperties - .getProducerProperties(outputName); - if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) { - Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputName); + .getProducerProperties(binderWrapper.destinationName()); + if (binderWrapper.binder() instanceof ExtendedPropertiesBinder extendedPropertiesBinder) { + Object extension = extendedPropertiesBinder.getExtendedProducerProperties(binderWrapper.destinationName()); ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>( extension); BeanUtils.copyProperties(producerProperties, extendedProducerProperties); producerProperties = extendedProducerProperties; } - producerProperties.populateBindingName(outputName); + producerProperties.populateBindingName(binderWrapper.destinationName()); validate(producerProperties); - Binding binding = doBindProducer(output, bindingTarget, binder, + String bindingTarget = this.bindingServiceProperties.getBindingDestination(binderWrapper.destinationName()); + Binding binding = doBindProducer(output, bindingTarget, binderWrapper.binder(), producerProperties); // If the downstream binder modified the partition count in the extended producer properties // based on the higher number of partitions provisioned on the target middleware, update that // in the original producer properties. ProducerProperties originalProducerProperties = this.bindingServiceProperties - .getProducerProperties(outputName); + .getProducerProperties(binderWrapper.destinationName()); if (originalProducerProperties.getPartitionCount() < producerProperties.getPartitionCount()) { originalProducerProperties.setPartitionCount(producerProperties.getPartitionCount()); } if (cache) { - this.producerBindings.put(outputName, binding); + this.producerBindings.put(binderWrapper.cacheKey(), binding); } return binding; } public Binding bindProducer(T output, String outputName, boolean cache) { - return this.bindProducer(output, outputName, cache, null); + Class outputClass = output.getClass(); + if (output instanceof Advised advisedOutput) { + outputClass = Stream.of(advisedOutput.getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst() + .orElse(outputClass); + } + BinderWrapper binderWrapper = createBinderWrapper(null, outputName, outputClass); + return this.bindProducer(output, cache, binderWrapper); } public Binding bindProducer(T output, String outputName) { @@ -318,8 +319,7 @@ public class BindingService { } @SuppressWarnings("rawtypes") - public Object getExtendedProducerProperties(Object output, String outputName) { - Binder binder = getBinder(outputName, output.getClass()); + public Object getExtendedProducerProperties(Binder binder, String outputName) { if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) { return extendedPropertiesBinder.getExtendedProducerProperties(outputName); } @@ -398,8 +398,13 @@ public class BindingService { } } - public void unbindProducers(String outputName) { - Binding binding = this.producerBindings.remove(outputName); + public void unbindProducers(@Nullable String binderName, String outputName) { + String cacheKey = createChannelCacheKey(binderName, outputName, bindingServiceProperties); + unbindProducers(cacheKey); + } + + public void unbindProducers(String cacheKey) { + Binding binding = this.producerBindings.remove(cacheKey); if (binding != null) { binding.stop(); @@ -407,7 +412,7 @@ public class BindingService { binding.unbind(); } else if (this.log.isWarnEnabled()) { - this.log.warn("Trying to unbind '" + outputName + "', but no binding found."); + this.log.warn("Trying to unbind '" + cacheKey + "', but no binding found."); } } @@ -443,6 +448,14 @@ public class BindingService { } } + public BinderWrapper createBinderWrapper(@Nullable String binderName, String destinationName, Class outputClass) { + binderName = getBinderNameIfNeeded(binderName, destinationName, bindingServiceProperties); + Binder binder = binderFactory.getBinder(binderName, outputClass); + String channelCacheKey = createChannelCacheKey(binderName, destinationName); + return new BinderWrapper(binder, destinationName, channelCacheKey); + } + + public static class LateBinding implements Binding { private volatile Binding delegate; 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 7e22cdb75..640f43e47 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 @@ -43,6 +43,7 @@ import org.springframework.cloud.function.context.message.MessageUtils; import org.springframework.cloud.function.core.FunctionInvocationHelper; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binder.BinderWrapper; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.binding.DefaultPartitioningInterceptor; @@ -68,6 +69,8 @@ import org.springframework.util.MimeTypeUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; +import static org.springframework.cloud.stream.utils.CacheKeyCreatorUtils.createChannelCacheKey; + /** * A class which allows user to send data to an output binding. @@ -85,6 +88,7 @@ import org.springframework.util.StringUtils; * @author Soby Chacko * @author Byungjun You * @author Michał Rowicki + * @author Omer Celik * @since 3.0.3 * */ @@ -128,7 +132,6 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi * @param bindingServiceProperties instance of {@link BindingServiceProperties} * @param applicationContext instance of {@link ConfigurableApplicationContext} */ - @SuppressWarnings("serial") StreamBridge(FunctionCatalog functionCatalog, BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback) { this.executorService = Executors.newCachedThreadPool(); @@ -268,13 +271,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi MessageChannel resolveDestination(String destinationName, ProducerProperties producerProperties, String binderName) { lock.lock(); try { - MessageChannel messageChannel = null; - if (StringUtils.hasText(binderName)) { - messageChannel = this.channelCache.get(binderName + ":" + destinationName); - } - else { - messageChannel = this.channelCache.get(destinationName); - } + MessageChannel messageChannel = this.channelCache.get(createChannelCacheKey(binderName, destinationName, bindingServiceProperties)); if (messageChannel == null) { if (this.applicationContext.containsBean(destinationName)) { messageChannel = this.applicationContext.getBean(destinationName, MessageChannel.class); @@ -291,28 +288,20 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi messageChannel = this.isAsync() ? new ExecutorChannel(this.executorService) : new DirectWithAttributesChannel(); ((AbstractSubscribableChannel) messageChannel).setApplicationContext(applicationContext); ((AbstractSubscribableChannel) messageChannel).setComponentName(destinationName); + + BinderWrapper binderWrapper = bindingService.createBinderWrapper(binderName, destinationName, messageChannel.getClass()); if (this.destinationBindingCallback != null) { Object extendedProducerProperties = this.bindingService - .getExtendedProducerProperties(messageChannel, destinationName); + .getExtendedProducerProperties(binderWrapper.binder(), destinationName); this.destinationBindingCallback.configure(destinationName, messageChannel, producerProperties, extendedProducerProperties); } - Binder binder = null; - if (StringUtils.hasText(binderName)) { - BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class); - binder = binderFactory.getBinder(binderName, messageChannel.getClass()); - } addPartitioningInterceptorIfNeedBe(producerProperties, destinationName, (AbstractMessageChannel) messageChannel); addGlobalChannelInterceptorProcessor((AbstractMessageChannel) messageChannel, destinationName); - this.bindingService.bindProducer(messageChannel, destinationName, true, binder); - if (StringUtils.hasText(binderName)) { - this.channelCache.put(binderName + ":" + destinationName, messageChannel); - } - else { - this.channelCache.put(destinationName, messageChannel); - } + this.bindingService.bindProducer(messageChannel, true, binderWrapper); + this.channelCache.put(binderWrapper.cacheKey(), messageChannel); } } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/CacheKeyCreatorUtils.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/CacheKeyCreatorUtils.java new file mode 100644 index 000000000..4a9d12879 --- /dev/null +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/CacheKeyCreatorUtils.java @@ -0,0 +1,63 @@ +/* + * Copyright 2024-2024 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.utils; + +import java.util.Objects; + +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.lang.Nullable; +import org.springframework.util.StringUtils; + + +/** + * Utility methods related to Output Channel Cache Key. + * + * @author Omer Celik + */ +public final class CacheKeyCreatorUtils { + + private CacheKeyCreatorUtils() { + } + + public static String createChannelCacheKey(@Nullable String binderName, String outputName, + BindingServiceProperties bindingServiceProperties) { + String finalBinderName = getBinderNameIfNeeded(binderName, outputName, bindingServiceProperties); + return createChannelCacheKey(finalBinderName, outputName); + } + + public static String getBinderNameIfNeeded(@Nullable String binderName, String outputName, + BindingServiceProperties bindingServiceProperties) { + if (Objects.isNull(binderName)) { + return bindingServiceProperties.getBinder(outputName); + } + return binderName; + } + + public static String createChannelCacheKey(String outputName, BindingServiceProperties bindingServiceProperties) { + String binderName = bindingServiceProperties.getBinder(outputName); + return createChannelCacheKey(binderName, outputName); + } + + public static String createChannelCacheKey(String binderName, String outputName) { + if (StringUtils.hasText(binderName)) { + return binderName + ":" + outputName; + } + else { + return outputName; + } + } +}