diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 8a8b532dd..647c36975 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1627,6 +1627,38 @@ environment: profiles: active: myBinderProfile ---- + +[[binder-customizer]] +=== Customizing binders in multi binder applications + +When an application has multiple binders in it and wants to customize the binders, then that can be achieved by providing a `BinderCustomizer` implementation. +In the case of applications with a single binder, this special customizer is not necessary since the binder context can access the customization beans directly. +However, this is not the case in a multi-binder scenario, since various binders live in different application contexts. +By providing an implementation of `BinderCustomizer` interface, the binders, although reside in different application contexts, will receive the customization. +Spring Cloud Stream ensures that the customizations take place before the applications start using the binders. +The user must check for the binder type and then apply the necessary customizations. + +Here is an example of providing a `BinderCustomizer` bean. + +``` +@Bean +public BinderCustomizer binderCustomizer() { + return (binder, binderName) -> { + if (binder instanceof KafkaMessageChannelBinder) { + ((KafkaMessageChannelBinder) binder).setRebalanceListener(...); + } + else if (binder instanceof KStreamBinder) { + ... + } + else if (binder instanceof RabbitMessageChannelBinder) { + ... + } + }; +} +``` + +Note that, when there are more than one instance of the same type of the binder, the binder name can be used to filter customization. + [[binding_visualization_control]] === Binding visualization and control Since version 2.0, Spring Cloud Stream supports visualization and control of the Bindings through Actuator endpoints. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderCustomizer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderCustomizer.java new file mode 100644 index 000000000..5e15de867 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderCustomizer.java @@ -0,0 +1,40 @@ +/* + * Copyright 2015-2020 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; + +/** + * @author Soby Chacko + * @since 3.1.0 + */ +public interface BinderCustomizer { + + /** + * When customization beans are present in an application that uses a single binder, + * those beans are detected by the binder. However, this is not the case in a multi-binder + * scenario, since various binders live in different application contexts. This customizer + * enables the application to properly apply customizations in all the binders. + * By providing an implementation of this interface, the binders, although reside + * in different application contexts, will receive the customization. + * Spring Cloud Stream ensures that the customizations take place before the binders are + * accessed. The user must check for the binder type and then apply the necessary customizations. + * In the case of a single binder, the use of this customizer is redundant. + * + * @param binder to be customized + * @param binderName binder name to distinguish between multiple instances of the same binder type + */ + void customize(Binder binder, String binderName); +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index bd730234a..d5291d54b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -77,6 +77,8 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl private final BinderTypeRegistry binderTypeRegistry; + private final BinderCustomizer binderCustomizer; + private volatile ConfigurableApplicationContext context; private Collection listeners; @@ -84,9 +86,10 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl private volatile String defaultBinder; public DefaultBinderFactory(Map binderConfigurations, - BinderTypeRegistry binderTypeRegistry) { + BinderTypeRegistry binderTypeRegistry, BinderCustomizer binderCustomizer) { this.binderConfigurations = new HashMap<>(binderConfigurations); this.binderTypeRegistry = binderTypeRegistry; + this.binderCustomizer = binderCustomizer; } @Override @@ -139,6 +142,9 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl */ binder = this.doGetBinder(binderName, bindingTargetType); } + if (this.binderCustomizer != null) { + this.binderCustomizer.customize(binder, binderName); + } return binder; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 7ec5e5510..1b4170425 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -23,6 +23,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.BeanDefinition; @@ -32,6 +33,7 @@ import org.springframework.boot.autoconfigure.condition.SearchStrategy; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.boot.context.properties.source.ConfigurationPropertyName; import org.springframework.cloud.stream.binder.BinderConfiguration; +import org.springframework.cloud.stream.binder.BinderCustomizer; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.BinderType; import org.springframework.cloud.stream.binder.BinderTypeRegistry; @@ -177,11 +179,12 @@ public class BindingServiceConfiguration { @Bean @ConditionalOnMissingBean(BinderFactory.class) public BinderFactory binderFactory(BinderTypeRegistry binderTypeRegistry, - BindingServiceProperties bindingServiceProperties) { + BindingServiceProperties bindingServiceProperties, + ObjectProvider binderCustomizerProvider) { DefaultBinderFactory binderFactory = new DefaultBinderFactory( getBinderConfigurations(binderTypeRegistry, bindingServiceProperties), - binderTypeRegistry); + binderTypeRegistry, binderCustomizerProvider.getIfUnique()); binderFactory.setDefaultBinder(bindingServiceProperties.getDefaultBinder()); binderFactory.setListeners(this.binderFactoryListeners); return binderFactory; diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java index 3e9f85e64..3f74cfc02 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java @@ -289,8 +289,10 @@ public class BinderFactoryAutoConfigurationTests { Binder binder1 = binderFactory.getBinder("binder1", MessageChannel.class); assertThat(binder1).isInstanceOf(StubBinder1.class); + assertThat(((StubBinder1) binder1).getFromCustomization()).isEqualTo("customizer-applied-binder1"); Binder binder2 = binderFactory.getBinder("binder2", MessageChannel.class); assertThat(binder2).isInstanceOf(StubBinder2.class); + assertThat(((StubBinder2) binder2).getFromCustomization()).isEqualTo("customizer-applied-binder2"); Binder defaultBinder = binderFactory.getBinder(null, MessageChannel.class); assertThat(defaultBinder).isSameAs(binder2); @@ -312,6 +314,17 @@ public class BinderFactoryAutoConfigurationTests { } + @Bean + public BinderCustomizer binderCustomizer() { + return (binder, binderName) -> { + if (binder instanceof StubBinder1) { + ((StubBinder1) binder).setFromCustomization("customizer-applied-binder1"); + } + else if (binder instanceof StubBinder2) { + ((StubBinder2) binder).setFromCustomization("customizer-applied-binder2"); + } + }; + } } @Configuration @@ -321,7 +334,6 @@ public class BinderFactoryAutoConfigurationTests { public String fooBean() { return "foo"; } - } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDefaultTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDefaultTests.java index 0cc87e7b0..636e2d06d 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDefaultTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDefaultTests.java @@ -62,7 +62,7 @@ public class ExtendedPropertiesDefaultTests { return new DefaultBinderFactory( Collections.singletonMap("mock", new BinderConfiguration("mock", new HashMap<>(), true, true)), - binderTypeRegistry); + binderTypeRegistry, null); } private DefaultBinderTypeRegistry createMockExtendedBinderTypeRegistry() { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java index ed8dc9cae..f18f24c07 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java @@ -34,6 +34,8 @@ public class StubBinder1 private ConfigurableApplicationContext outerContext; + private String fromCustomization; + public ConfigurableApplicationContext getOuterContext() { return this.outerContext; } @@ -62,4 +64,11 @@ public class StubBinder1 return null; } + public String getFromCustomization() { + return fromCustomization; + } + + public void setFromCustomization(String fromCustomization) { + this.fromCustomization = fromCustomization; + } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java index a2433505a..48a6da9a7 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java @@ -31,6 +31,8 @@ public class StubBinder2 @SuppressWarnings("unused") private final StubBinder2Dependency stubBinder2Dependency; + private String fromCustomization; + public StubBinder2(StubBinder2Dependency stubBinder2Dependency) { this.stubBinder2Dependency = stubBinder2Dependency; } @@ -47,4 +49,11 @@ public class StubBinder2 return null; } + public String getFromCustomization() { + return fromCustomization; + } + + public void setFromCustomization(String fromCustomization) { + this.fromCustomization = fromCustomization; + } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index 21be8b35b..c9383e1b2 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java @@ -38,6 +38,7 @@ import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.beans.factory.NoSuchBeanDefinitionException; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -530,6 +531,7 @@ public class BindingServiceTests { } } + @SuppressWarnings("unchecked") @Test public void testUnrecognizedBinderAllowedIfNotUsed() { HashMap properties = new HashMap<>(); @@ -541,13 +543,14 @@ public class BindingServiceTests { BindingServiceProperties bindingServiceProperties = createBindingServiceProperties( properties); BinderFactory binderFactory = new BindingServiceConfiguration() - .binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties); + .binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties, Mockito.mock(ObjectProvider.class)); BindingService bindingService = new BindingService(bindingServiceProperties, binderFactory); bindingService.bindConsumer(new DirectChannel(), "input"); bindingService.bindProducer(new DirectChannel(), "output"); } + @SuppressWarnings("unchecked") @Test public void testUnrecognizedBinderDisallowedIfUsed() { HashMap properties = new HashMap<>(); @@ -560,7 +563,7 @@ public class BindingServiceTests { BindingServiceProperties bindingServiceProperties = createBindingServiceProperties( properties); BinderFactory binderFactory = new BindingServiceConfiguration() - .binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties); + .binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties, Mockito.mock(ObjectProvider.class)); BindingService bindingService = new BindingService(bindingServiceProperties, binderFactory); bindingService.bindConsumer(new DirectChannel(), "input"); @@ -693,7 +696,7 @@ public class BindingServiceTests { return new DefaultBinderFactory( Collections.singletonMap("mock", new BinderConfiguration("mock", new HashMap<>(), true, true)), - binderTypeRegistry); + binderTypeRegistry, null); } private DefaultBinderTypeRegistry createMockBinderTypeRegistry() {