SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1009: Producer properties for dynamic bindings

Browse Source 
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
This commit is contained in:
Gary Russell
2017-11-10 09:54:08 -05:00
committed by Oleg Zhurakousky
parent 73f0fc38a6
commit 83fa5f3858
6 changed files with 160 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc
View File

@@ -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<RabbitProducerProperties> 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

49
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java
View File

@@ -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<? extends MessageChannel> bindingTargetFactory,
DynamicDestinationsBindable dynamicDestinationsBindable) {
this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null);
}
@SuppressWarnings("rawtypes")
public BinderAwareChannelResolver(BindingService bindingService,
AbstractBindingTargetFactory<? extends MessageChannel> 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<MessageChannel> 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 <T> 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<T> {
/**
* 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);
}
}

19
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
View File

@@ -63,7 +63,7 @@ public class BindingService {
private final Map<String, List<Binding<?>>> 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 <T> Collection<Binding<T>> 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 <T> Binding<T> 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<Binding<?>> bindings = this.consumerBindings.remove(inputName);
if (bindings != null && !CollectionUtils.isEmpty(bindings)) {
@@ -150,7 +161,7 @@ public class BindingService {
}
}
private <T> Binder<T, ?, ?> getBinder(String channelName, Class<T> bindableType) {
protected <T> Binder<T, ?, ?> getBinder(String channelName, Class<T> bindableType) {
String binderConfigurationName = this.bindingServiceProperties.getBinder(channelName);
return binderFactory.getBinder(binderConfigurationName, bindableType);
}

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
View File

@@ -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<? extends MessageChannel> bindingTargetFactory,
DynamicDestinationsBindable dynamicDestinationsBindable) {
return new BinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable);
DynamicDestinationsBindable dynamicDestinationsBindable,
@Nullable BinderAwareChannelResolver.NewDestinationBindingCallback callback) {
return new BinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable,
callback);
}
@Bean

17
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java
View File

@@ -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);
}
}
}

63
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java
View File

@@ -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<MessageChannel> mockBinding = Mockito.mock(Binding.class);
@SuppressWarnings("unchecked")
final AtomicReference<MessageChannel> 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 <T> Binder<T, ?, ?> getBinder(String channelName, Class<T> 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<ProducerProperties> 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());