From 83fa5f3858621ae0382dbef7358366a588804222 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 10 Nov 2017 09:54:08 -0500 Subject: [PATCH] GH-1009: Producer properties for dynamic bindings Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/1009 Provide a callback mechanism to set properties on dynamically created bindings. Polishing - callback name. Resolves #1009 Resolves #1132 --- .../spring-cloud-stream-overview.adoc | 31 ++++++++- .../binding/BinderAwareChannelResolver.java | 49 +++++++++++++++ .../cloud/stream/binding/BindingService.java | 19 ++++-- .../config/BindingServiceConfiguration.java | 9 ++- .../config/BindingServiceProperties.java | 17 +++-- .../stream/binding/BindingServiceTests.java | 63 ++++++++++++++----- 6 files changed, 160 insertions(+), 28 deletions(-) diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 4e04b1e7d..1f7e75193 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -1282,7 +1282,7 @@ This is useful, for example, when the target destination needs to be determined Applications can do so by using the `BinderAwareChannelResolver` bean, registered automatically by the `@EnableBinding` annotation. The property 'spring.cloud.stream.dynamicDestinations' can be used for restricting the dynamic destination names to a set known beforehand (whitelisting). -If the property is not set, any destination can be bound dynamicaly. +If the property is not set, any destination can be bound dynamically. The `BinderAwareChannelResolver` can be used directly as in the following example, in which a REST controller uses a path variable to decide the target channel. @@ -1360,6 +1360,35 @@ public class SourceWithDynamicDestination { } ---- +The https://github.com/spring-cloud-stream-app-starters/router[Router Sink Application] uses this technique to create the destinations on-demand. + +If the channel names are known in advance, you can configure the producer properties as with any other destination. +Alternatively, if you register a `NewBindingCallback<>` bean, it will be invoked just before the binding is created. +The callback takes the generic type of the extended producer properties used by the binder; it has one method: + +[source, java] +---- +void configure(String channelName, MessageChannel channel, ProducerProperties producerProperties, + T extendedProducerProperties); +---- + +The following is an example using the RabbitMQ binder: + +[source, xml] +---- +@Bean +public NewBindingCallback dynamicConfigurer() { + return (name, channel, props, extended) -> { + props.setRequiredGroups("bindThisQueue"); + extended.setQueueNameGroupOnly(true); + extended.setAutoBindDlq(true); + extended.setDeadLetterQueueName("myDLQ"); + }; +} +---- + +NOTE: If you need to support dynamic destinations with multiple binder types, use `Object` for the generic type and cast the `extended` argument as needed. + [[contenttypemanagement]] == Content Type and Transformation diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java index f0821b536..42e7d81c4 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binding; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver; @@ -43,16 +44,27 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina private final DynamicDestinationsBindable dynamicDestinationsBindable; + @SuppressWarnings("rawtypes") + private final NewDestinationBindingCallback newBindingCallback; + private ConfigurableListableBeanFactory beanFactory; public BinderAwareChannelResolver(BindingService bindingService, AbstractBindingTargetFactory bindingTargetFactory, DynamicDestinationsBindable dynamicDestinationsBindable) { + this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null); + } + + @SuppressWarnings("rawtypes") + public BinderAwareChannelResolver(BindingService bindingService, + AbstractBindingTargetFactory bindingTargetFactory, + DynamicDestinationsBindable dynamicDestinationsBindable, NewDestinationBindingCallback callback) { this.dynamicDestinationsBindable = dynamicDestinationsBindable; Assert.notNull(bindingService, "'bindingService' cannot be null"); Assert.notNull(bindingTargetFactory, "'bindingTargetFactory' cannot be null"); this.bindingService = bindingService; this.bindingTargetFactory = bindingTargetFactory; + this.newBindingCallback = callback; } @Override @@ -63,6 +75,7 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina } } + @SuppressWarnings("unchecked") @Override public MessageChannel resolveDestination(String channelName) { try { @@ -93,6 +106,16 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina channel = this.bindingTargetFactory.createOutput(channelName); this.beanFactory.registerSingleton(channelName, channel); channel = (MessageChannel) this.beanFactory.initializeBean(channel, channelName); + if (this.newBindingCallback != null) { + ProducerProperties producerProperties = this.bindingService.getBindingServiceProperties() + .getProducerProperties(channelName); + Object extendedProducerProperties = + this.bindingService.getExtendedProducerProperties(channel, channelName); + this.newBindingCallback.configure(channelName, channel, producerProperties, + extendedProducerProperties); + this.bindingService.getBindingServiceProperties().updateProducerProperties(channelName, + producerProperties); + } Binding binding = this.bindingService.bindProducer(channel, channelName); this.dynamicDestinationsBindable.addOutputBinding(channelName, binding); } @@ -103,4 +126,30 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina return channel; } } + + /** + * Configure a new destination before it is bound. + * @param the extended properties type. If you need to support dynamic binding + * with multiple binders, use {@link Object} and cast as needed. + * + * @since 2.0 + * + */ + @FunctionalInterface + public interface NewDestinationBindingCallback { + + /** + * Configure the properties or channel before binding. + * @param channelName the name of the new channel. + * @param channel the channel that is about to be bound. + * @param producerProperties the producer properties. + * @param extendedProducerProperties the extended producer properties (type + * depends on binder type and may be null if the binder doesn't support + * extended properties). + */ + void configure(String channelName, MessageChannel channel, ProducerProperties producerProperties, + T extendedProducerProperties); + + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index bc2718a64..706a7932e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java @@ -63,7 +63,7 @@ public class BindingService { private final Map>> consumerBindings = new HashMap<>(); - private BinderFactory binderFactory; + private final BinderFactory binderFactory; public BindingService( BindingServiceProperties bindingServiceProperties, @@ -74,7 +74,7 @@ public class BindingService { this.validator.afterPropertiesSet(); } - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) public Collection> bindConsumer(T input, String inputName) { String bindingTarget = this.bindingServiceProperties .getBindingDestination(inputName); @@ -105,7 +105,7 @@ public class BindingService { return bindings; } - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) public Binding bindProducer(T output, String outputName) { String bindingTarget = this.bindingServiceProperties .getBindingDestination(outputName); @@ -128,6 +128,17 @@ public class BindingService { return binding; } + @SuppressWarnings("rawtypes") + public Object getExtendedProducerProperties(Object output, String outputName) { + Binder binder = getBinder(outputName, output.getClass()); + if (binder instanceof ExtendedPropertiesBinder) { + return ((ExtendedPropertiesBinder) binder).getExtendedProducerProperties(outputName); + } + else { + return null; + } + } + public void unbindConsumers(String inputName) { List> bindings = this.consumerBindings.remove(inputName); if (bindings != null && !CollectionUtils.isEmpty(bindings)) { @@ -150,7 +161,7 @@ public class BindingService { } } - private Binder getBinder(String channelName, Class bindableType) { + protected Binder getBinder(String channelName, Class bindableType) { String binderConfigurationName = this.bindingServiceProperties.getBinder(channelName); return binderFactory.getBinder(binderConfigurationName, bindableType); } 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 3c8d82874..ceae60c79 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 @@ -21,7 +21,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; - import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactoryUtils; @@ -59,6 +58,7 @@ import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.json.JsonPropertyAccessor; +import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.core.DestinationResolutionException; @@ -150,11 +150,14 @@ public class BindingServiceConfiguration { return new ContextStartAfterRefreshListener(); } + @SuppressWarnings("rawtypes") @Bean public BinderAwareChannelResolver binderAwareChannelResolver(BindingService bindingService, AbstractBindingTargetFactory bindingTargetFactory, - DynamicDestinationsBindable dynamicDestinationsBindable) { - return new BinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable); + DynamicDestinationsBindable dynamicDestinationsBindable, + @Nullable BinderAwareChannelResolver.NewDestinationBindingCallback callback) { + return new BinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable, + callback); } @Bean diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index ae06a2730..2a34dad25 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java @@ -168,10 +168,11 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia public ConsumerProperties getConsumerProperties(String inputBindingName) { Assert.notNull(inputBindingName, "The input binding name cannot be null"); - ConsumerProperties consumerProperties = getBindingProperties(inputBindingName) - .getConsumer(); + BindingProperties bindingProperties = getBindingProperties(inputBindingName); + ConsumerProperties consumerProperties = bindingProperties.getConsumer(); if (consumerProperties == null) { consumerProperties = new ConsumerProperties(); + bindingProperties.setConsumer(consumerProperties); } // propagate instance count and instance index if not already set if (consumerProperties.getInstanceCount() < 0) { @@ -185,10 +186,11 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia public ProducerProperties getProducerProperties(String outputBindingName) { Assert.notNull(outputBindingName, "The output binding name cannot be null"); - ProducerProperties producerProperties = getBindingProperties(outputBindingName) - .getProducer(); + BindingProperties bindingProperties = getBindingProperties(outputBindingName); + ProducerProperties producerProperties = bindingProperties.getProducer(); if (producerProperties == null) { producerProperties = new ProducerProperties(); + bindingProperties.setProducer(producerProperties); } return producerProperties; } @@ -211,4 +213,11 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia public String getBindingDestination(String bindingName) { return getBindingProperties(bindingName).getDestination(); } + + public void updateProducerProperties(String bindingName, ProducerProperties producerProperties) { + if (this.bindings.containsKey(bindingName)) { + this.bindings.get(bindingName).setProducer(producerProperties); + } + } + } 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 01cc6a1ad..cd1f91857 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 @@ -22,9 +22,11 @@ import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.Properties; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -40,6 +42,8 @@ import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.DefaultBinderFactory; import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.config.BinderFactoryConfiguration; import org.springframework.cloud.stream.config.BindingProperties; @@ -53,12 +57,14 @@ import org.springframework.messaging.core.DestinationResolutionException; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.fail; -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.eq; -import static org.mockito.Matchers.isNull; -import static org.mockito.Matchers.matches; -import static org.mockito.Matchers.same; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.ArgumentMatchers.matches; +import static org.mockito.ArgumentMatchers.same; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -196,25 +202,46 @@ public class BindingServiceTests { binderFactory.destroy(); } + @SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void checkDynamicBinding() { BindingServiceProperties properties = new BindingServiceProperties(); + BindingProperties bindingProperties = new BindingProperties(); + bindingProperties.setProducer(new ProducerProperties()); + properties.setBindings(Collections.singletonMap("foo", bindingProperties)); DefaultBinderFactory binderFactory = createMockBinderFactory(); - Binder binder = binderFactory.getBinder("mock", MessageChannel.class); - @SuppressWarnings("unchecked") + final ExtendedPropertiesBinder binder = mock(ExtendedPropertiesBinder.class); + Properties extendedProps = new Properties(); + when(binder.getExtendedProducerProperties(anyString())).thenReturn(extendedProps); Binding mockBinding = Mockito.mock(Binding.class); - @SuppressWarnings("unchecked") final AtomicReference dynamic = new AtomicReference<>(); when(binder.bindProducer(matches("foo"), any(DirectChannel.class), any(ProducerProperties.class))).thenReturn(mockBinding); - BindingService bindingService = new BindingService(properties, binderFactory); - SubscribableChannelBindingTargetFactory bindableSubscribableChannelFactory = new SubscribableChannelBindingTargetFactory( - new MessageConverterConfigurer(properties, new CompositeMessageConverterFactory())); + BindingService bindingService = new BindingService(properties, binderFactory) { + + @Override + protected Binder getBinder(String channelName, Class bindableType) { + return binder; + } + + }; + SubscribableChannelBindingTargetFactory bindableSubscribableChannelFactory = + new SubscribableChannelBindingTargetFactory( + new MessageConverterConfigurer(properties, new CompositeMessageConverterFactory())); + final AtomicBoolean callbackInvoked = new AtomicBoolean(); BinderAwareChannelResolver resolver = new BinderAwareChannelResolver( bindingService, bindableSubscribableChannelFactory, - new DynamicDestinationsBindable()); - ConfigurableListableBeanFactory beanFactory = mock( - ConfigurableListableBeanFactory.class); + new DynamicDestinationsBindable(), + (name, channel, props, extended) -> { + callbackInvoked.set(true); + assertThat(name).isEqualTo("foo"); + assertThat(channel).isNotNull(); + assertThat(props).isNotNull(); + assertThat(extended).isSameAs(extendedProps); + props.setUseNativeEncoding(true); + extendedProps.setProperty("bar", "baz"); + }); + ConfigurableListableBeanFactory beanFactory = mock(ConfigurableListableBeanFactory.class); when(beanFactory.getBean("foo", MessageChannel.class)) .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class)); when(beanFactory.getBean("bar", MessageChannel.class)) @@ -239,8 +266,12 @@ public class BindingServiceTests { resolver.setBeanFactory(beanFactory); MessageChannel resolved = resolver.resolveDestination("foo"); assertThat(resolved).isSameAs(dynamic.get()); - verify(binder).bindProducer(eq("foo"), eq(dynamic.get()), - any(ProducerProperties.class)); + ArgumentCaptor captor = ArgumentCaptor.forClass(ProducerProperties.class); + verify(binder).bindProducer(eq("foo"), eq(dynamic.get()), captor.capture()); + assertThat(captor.getValue().isUseNativeEncoding()).isTrue(); + assertThat(captor.getValue()).isInstanceOf(ExtendedProducerProperties.class); + assertThat(((ExtendedProducerProperties) captor.getValue()).getExtension()).isSameAs(extendedProps); + doReturn(dynamic.get()).when(beanFactory).getBean("foo", MessageChannel.class); properties.setDynamicDestinations(new String[] { "foo" }); resolved = resolver.resolveDestination("foo"); assertThat(resolved).isSameAs(dynamic.get());