BinderAwareChannelResolver to check for ExtendedProperties
- When resolving the destination names via BinderAwareChannelResolver, the underlying binder needs to be checked if it is of type `ExtendedPropertiesBinder` so that any ExtendedProducerProperties will be considered based on the binder - Also move the validation and retrieval of producer/consumer properties to a util - Update BinderAwareChannelResolverTests to use ExtendedPropertiesBinder This resolves #483 Separate tests for ExtendedPropertiesBinder Remove binder type from dynamic destination names - This leads to using `ChannelBindingService` to bind the dynamic destination names
This commit is contained in:
Ilayaperumal Gopinathan
committed by
Marius Bogoevici
Marius Bogoevici
parent
045d280542
commit
f029249e1b
@@ -18,9 +18,6 @@ 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.Binder;
|
||||
import org.springframework.cloud.stream.binder.BinderFactory;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver;
|
||||
@@ -31,36 +28,26 @@ import org.springframework.util.ObjectUtils;
|
||||
/**
|
||||
* A {@link org.springframework.messaging.core.DestinationResolver} implementation that
|
||||
* resolves the channel from the bean factory and, if not present, creates a new channel
|
||||
* and adds it to the factory after binding it to the binder. The binder is optionally
|
||||
* determined with a prefix preceding a colon.
|
||||
* and adds it to the factory after binding it to the binder.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @author Gary Russell
|
||||
* @author Ilayaperumal Gopinathan
|
||||
*/
|
||||
public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestinationResolver {
|
||||
|
||||
private final BinderFactory<MessageChannel> binderFactory;
|
||||
|
||||
private final ChannelBindingServiceProperties channelBindingServiceProperties;
|
||||
|
||||
private final DynamicDestinationsBindable dynamicDestinationsBindable;
|
||||
private final ChannelBindingService channelBindingService;
|
||||
|
||||
private final BindableChannelFactory bindableChannelFactory;
|
||||
|
||||
private ConfigurableListableBeanFactory beanFactory;
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public BinderAwareChannelResolver(BinderFactory binderFactory,
|
||||
ChannelBindingServiceProperties channelBindingServiceProperties,
|
||||
DynamicDestinationsBindable dynamicDestinationsBindable,
|
||||
public BinderAwareChannelResolver(ChannelBindingService channelBindingService,
|
||||
BindableChannelFactory bindableChannelFactory) {
|
||||
Assert.notNull(binderFactory, "'binderFactory' cannot be null");
|
||||
Assert.notNull(channelBindingServiceProperties, "'channelBindingServiceProperties' cannot be null");
|
||||
Assert.notNull(dynamicDestinationsBindable, "'dynamicDestinationBindable' cannot be null");
|
||||
Assert.notNull(channelBindingService, "'channelBindingService' cannot be null");
|
||||
Assert.notNull(bindableChannelFactory, "'bindableChannelFactory' cannot be null");
|
||||
this.binderFactory = binderFactory;
|
||||
this.channelBindingServiceProperties = channelBindingServiceProperties;
|
||||
this.dynamicDestinationsBindable = dynamicDestinationsBindable;
|
||||
this.channelBindingService = channelBindingService;
|
||||
this.bindableChannelFactory = bindableChannelFactory;
|
||||
}
|
||||
|
||||
@@ -83,37 +70,20 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
|
||||
destinationResolutionException = e;
|
||||
}
|
||||
synchronized (this) {
|
||||
if (this.beanFactory != null && this.binderFactory != null) {
|
||||
if (this.beanFactory != null) {
|
||||
String[] dynamicDestinations = null;
|
||||
if (this.channelBindingServiceProperties != null) {
|
||||
dynamicDestinations = this.channelBindingServiceProperties.getDynamicDestinations();
|
||||
ChannelBindingServiceProperties channelBindingServiceProperties =
|
||||
this.channelBindingService.getChannelBindingServiceProperties();
|
||||
if (channelBindingServiceProperties != null) {
|
||||
dynamicDestinations = channelBindingServiceProperties.getDynamicDestinations();
|
||||
}
|
||||
boolean dynamicAllowed = ObjectUtils.isEmpty(dynamicDestinations)
|
||||
|| ObjectUtils.containsElement(dynamicDestinations, channelName);
|
||||
if (dynamicAllowed) {
|
||||
String binderName = null;
|
||||
String beanName = channelName;
|
||||
if (channelName.contains(":")) {
|
||||
String[] tokens = channelName.split(":", 2);
|
||||
if (tokens.length == 2) {
|
||||
binderName = tokens[0];
|
||||
channelName = tokens[1];
|
||||
}
|
||||
else if (tokens.length != 1) {
|
||||
throw new IllegalArgumentException("Unrecognized channel naming scheme: " + channelName + " , should be" +
|
||||
" [<binder>:]<channelName>");
|
||||
}
|
||||
}
|
||||
channel = this.bindableChannelFactory.createSubscribableChannel(channelName);
|
||||
this.beanFactory.registerSingleton(beanName, channel);
|
||||
channel = (MessageChannel) this.beanFactory.initializeBean(channel, beanName);
|
||||
@SuppressWarnings("unchecked")
|
||||
Binder<MessageChannel, ?, ProducerProperties> binder =
|
||||
(Binder<MessageChannel, ?, ProducerProperties>) binderFactory.getBinder(binderName);
|
||||
ProducerProperties producerProperties = this.channelBindingServiceProperties.getProducerProperties(channelName);
|
||||
String destinationName = this.channelBindingServiceProperties.getBindingDestination(channelName);
|
||||
this.dynamicDestinationsBindable.addOutputBinding(beanName,
|
||||
binder.bindProducer(destinationName, channel, producerProperties));
|
||||
this.beanFactory.registerSingleton(channelName, channel);
|
||||
channel = (MessageChannel) this.beanFactory.initializeBean(channel, channelName);
|
||||
this.channelBindingService.bindProducer(channel, channelName);
|
||||
}
|
||||
else {
|
||||
throw destinationResolutionException;
|
||||
|
||||
@@ -25,21 +25,16 @@ import java.util.Map;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.BeanUtils;
|
||||
import org.springframework.boot.bind.RelaxedDataBinder;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.cloud.stream.binder.BinderFactory;
|
||||
import org.springframework.cloud.stream.binder.Binding;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
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.ChannelBindingServiceProperties;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
import org.springframework.validation.beanvalidation.CustomValidatorBean;
|
||||
|
||||
/**
|
||||
* Handles the operations related to channel binding including binding of input/output channels by delegating
|
||||
@@ -78,13 +73,10 @@ public class ChannelBindingService {
|
||||
ConsumerProperties consumerProperties =
|
||||
this.channelBindingServiceProperties.getConsumerProperties(inputChannelName);
|
||||
if (binder instanceof ExtendedPropertiesBinder) {
|
||||
ExtendedPropertiesBinder extendedPropertiesBinder = (ExtendedPropertiesBinder) binder;
|
||||
Object extension = extendedPropertiesBinder.getExtendedConsumerProperties(inputChannelName);
|
||||
ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties(extension);
|
||||
BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties);
|
||||
consumerProperties = extendedConsumerProperties;
|
||||
consumerProperties = ProducerConsumerPropertiesUtil.getExtendedConsumerProperties((ExtendedPropertiesBinder) binder,
|
||||
inputChannelName, consumerProperties);
|
||||
}
|
||||
ProducerConsumerPropertiesValidator.validate(consumerProperties);
|
||||
ProducerConsumerPropertiesUtil.validate(consumerProperties);
|
||||
for (String target : channelBindingTargets) {
|
||||
Binding<MessageChannel> binding = binder.bindConsumer(target, channelBindingServiceProperties.getGroup(inputChannelName), inputChannel, consumerProperties);
|
||||
bindings.add(binding);
|
||||
@@ -100,13 +92,10 @@ public class ChannelBindingService {
|
||||
(Binder<MessageChannel, ?, ProducerProperties>) getBinderForChannel(outputChannelName);
|
||||
ProducerProperties producerProperties = this.channelBindingServiceProperties.getProducerProperties(outputChannelName);
|
||||
if (binder instanceof ExtendedPropertiesBinder) {
|
||||
ExtendedPropertiesBinder extendedPropertiesBinder = (ExtendedPropertiesBinder) binder;
|
||||
Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputChannelName);
|
||||
ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(extension);
|
||||
BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
|
||||
producerProperties = extendedProducerProperties;
|
||||
producerProperties = ProducerConsumerPropertiesUtil.getExtendedProducerProperties((ExtendedPropertiesBinder) binder,
|
||||
outputChannelName, producerProperties);
|
||||
}
|
||||
ProducerConsumerPropertiesValidator.validate(producerProperties);
|
||||
ProducerConsumerPropertiesUtil.validate(producerProperties);
|
||||
Binding<MessageChannel> binding = binder.bindProducer(channelBindingTarget, outputChannel, producerProperties);
|
||||
this.producerBindings.put(outputChannelName, binding);
|
||||
return binding;
|
||||
@@ -139,22 +128,7 @@ public class ChannelBindingService {
|
||||
return binderFactory.getBinder(transport);
|
||||
}
|
||||
|
||||
private static class ProducerConsumerPropertiesValidator {
|
||||
|
||||
private static CustomValidatorBean validator;
|
||||
|
||||
static {
|
||||
validator = new CustomValidatorBean();
|
||||
validator.afterPropertiesSet();
|
||||
}
|
||||
|
||||
static void validate(Object properties) {
|
||||
RelaxedDataBinder dataBinder = new RelaxedDataBinder(properties);
|
||||
dataBinder.setValidator(validator);
|
||||
dataBinder.validate();
|
||||
if (dataBinder.getBindingResult().hasErrors()) {
|
||||
throw new IllegalStateException(dataBinder.getBindingResult().toString());
|
||||
}
|
||||
}
|
||||
public ChannelBindingServiceProperties getChannelBindingServiceProperties() {
|
||||
return this.channelBindingServiceProperties;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,55 +0,0 @@
|
||||
/*
|
||||
* Copyright 2016 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
|
||||
*
|
||||
* http://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.binding;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import org.springframework.cloud.stream.binder.Binding;
|
||||
|
||||
/**
|
||||
* A {@link BindableAdapter} that stores the dynamic destination names and handles their unbinding.
|
||||
*
|
||||
* This class is not thread-safe.
|
||||
*
|
||||
* @author Ilayaperumal Gopinathan
|
||||
*/
|
||||
public final class DynamicDestinationsBindable extends BindableAdapter {
|
||||
|
||||
/**
|
||||
* Map containing dynamic channel names and their bindings.
|
||||
*/
|
||||
private Map<String, Binding> outputBindings = new HashMap<>();
|
||||
|
||||
public void addOutputBinding(String name, Binding binding) {
|
||||
this.outputBindings.put(name, binding);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Set<String> getOutputs() {
|
||||
return Collections.unmodifiableSet(outputBindings.keySet());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbindOutputs(ChannelBindingService adapter) {
|
||||
for (Map.Entry<String, Binding> entry: outputBindings.entrySet()) {
|
||||
entry.getValue().unbind();
|
||||
}
|
||||
outputBindings.clear();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
/*
|
||||
* Copyright 2016 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
|
||||
*
|
||||
* http://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.binding;
|
||||
|
||||
import org.springframework.beans.BeanUtils;
|
||||
import org.springframework.boot.bind.RelaxedDataBinder;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.validation.beanvalidation.CustomValidatorBean;
|
||||
|
||||
/**
|
||||
* Helper class for operations involving producer/consumer properties.
|
||||
*
|
||||
* @author Ilayaperumal Gopinathan
|
||||
*/
|
||||
public class ProducerConsumerPropertiesUtil {
|
||||
|
||||
private static CustomValidatorBean validator;
|
||||
|
||||
static {
|
||||
validator = new CustomValidatorBean();
|
||||
validator.afterPropertiesSet();
|
||||
}
|
||||
|
||||
static void validate(Object properties) {
|
||||
RelaxedDataBinder dataBinder = new RelaxedDataBinder(properties);
|
||||
dataBinder.setValidator(validator);
|
||||
dataBinder.validate();
|
||||
if (dataBinder.getBindingResult().hasErrors()) {
|
||||
throw new IllegalStateException(dataBinder.getBindingResult().toString());
|
||||
}
|
||||
}
|
||||
|
||||
static ProducerProperties getExtendedProducerProperties(ExtendedPropertiesBinder extendedPropertiesBinder, String outputChannelName,
|
||||
ProducerProperties producerProperties) {
|
||||
Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputChannelName);
|
||||
ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(extension);
|
||||
BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
|
||||
return extendedProducerProperties;
|
||||
}
|
||||
|
||||
static ConsumerProperties getExtendedConsumerProperties(ExtendedPropertiesBinder extendedPropertiesBinder, String inputChannelName,
|
||||
ConsumerProperties consumerProperties) {
|
||||
Object extension = extendedPropertiesBinder.getExtendedConsumerProperties(inputChannelName);
|
||||
ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties(extension);
|
||||
BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties);
|
||||
return extendedConsumerProperties;
|
||||
}
|
||||
}
|
||||
@@ -33,13 +33,12 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.BinderFactory;
|
||||
import org.springframework.cloud.stream.binding.BindableChannelFactory;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
|
||||
import org.springframework.cloud.stream.binding.ChannelBindingService;
|
||||
import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigurer;
|
||||
import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener;
|
||||
import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory;
|
||||
import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
|
||||
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
|
||||
import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
|
||||
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
|
||||
@@ -145,14 +144,11 @@ public class ChannelBindingServiceConfiguration {
|
||||
}
|
||||
|
||||
@Bean
|
||||
public BinderAwareChannelResolver binderAwareChannelResolver(BinderFactory<MessageChannel> binderFactory,
|
||||
ChannelBindingServiceProperties channelBindingServiceProperties,
|
||||
public BinderAwareChannelResolver binderAwareChannelResolver(ChannelBindingService channelBindingService,
|
||||
BindableChannelFactory bindableChannelFactory) {
|
||||
return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties, dynamicBindable(),
|
||||
bindableChannelFactory);
|
||||
return new BinderAwareChannelResolver(channelBindingService, bindableChannelFactory);
|
||||
}
|
||||
|
||||
|
||||
@Bean
|
||||
@ConditionalOnProperty("spring.cloud.stream.bindings." + ERROR_CHANNEL_NAME + ".destination")
|
||||
public SingleChannelBindable errorChannelBindable(
|
||||
@@ -160,11 +156,6 @@ public class ChannelBindingServiceConfiguration {
|
||||
return new SingleChannelBindable(ERROR_CHANNEL_NAME, errorChannel);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DynamicDestinationsBindable dynamicBindable() {
|
||||
return new DynamicDestinationsBindable();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public CompositeMessageConverterFactory compositeMessageConverterFactory() {
|
||||
List<AbstractFromMessageConverter> messageConverters = new ArrayList<>();
|
||||
|
||||
@@ -44,9 +44,9 @@ import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
||||
import org.springframework.cloud.stream.binding.BindableChannelFactory;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
|
||||
import org.springframework.cloud.stream.binding.ChannelBindingService;
|
||||
import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory;
|
||||
import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
|
||||
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
|
||||
import org.springframework.cloud.stream.config.BindingProperties;
|
||||
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
|
||||
@@ -68,17 +68,17 @@ import org.springframework.util.Assert;
|
||||
* @author Gary Russell
|
||||
* @author Ilayaperumal Gopinathan
|
||||
*/
|
||||
public class BinderAwareChannelResolverTests {
|
||||
public class DynamicDestinationResolverTests {
|
||||
|
||||
private final StaticApplicationContext context = new StaticApplicationContext();
|
||||
protected final StaticApplicationContext context = new StaticApplicationContext();
|
||||
|
||||
private volatile BinderAwareChannelResolver resolver;
|
||||
protected volatile BinderAwareChannelResolver resolver;
|
||||
|
||||
private volatile Binder<MessageChannel, ConsumerProperties, ProducerProperties> binder;
|
||||
protected volatile Binder<MessageChannel, ConsumerProperties, ProducerProperties> binder;
|
||||
|
||||
private volatile BindableChannelFactory bindableChannelFactory;
|
||||
protected volatile BindableChannelFactory bindableChannelFactory;
|
||||
|
||||
private volatile ChannelBindingServiceProperties channelBindingServiceProperties;
|
||||
protected volatile ChannelBindingServiceProperties channelBindingServiceProperties;
|
||||
|
||||
@Before
|
||||
public void setupContext() throws Exception {
|
||||
@@ -96,14 +96,14 @@ public class BinderAwareChannelResolverTests {
|
||||
bindingProperties.setContentType("text/plain");
|
||||
bindings.put("foo", bindingProperties);
|
||||
this.channelBindingServiceProperties.setBindings(bindings);
|
||||
ChannelBindingService channelBindingService = new ChannelBindingService(channelBindingServiceProperties, binderFactory);
|
||||
MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer(
|
||||
this.channelBindingServiceProperties, new DefaultMessageBuilderFactory(),
|
||||
new CompositeMessageConverterFactory());
|
||||
messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class));
|
||||
messageConverterConfigurer.afterPropertiesSet();
|
||||
this.bindableChannelFactory = new DefaultBindableChannelFactory(messageConverterConfigurer);
|
||||
this.resolver = new BinderAwareChannelResolver(binderFactory, this.channelBindingServiceProperties,
|
||||
new DynamicDestinationsBindable(), bindableChannelFactory);
|
||||
this.resolver = new BinderAwareChannelResolver(channelBindingService, this.bindableChannelFactory);
|
||||
this.resolver.setBeanFactory(context.getBeanFactory());
|
||||
context.getBeanFactory().registerSingleton("channelResolver",
|
||||
this.resolver);
|
||||
@@ -151,7 +151,6 @@ public class BinderAwareChannelResolverTests {
|
||||
@Test
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
public void propertyPassthrough() {
|
||||
DynamicDestinationsBindable dynamicDestinationsBindable = new DynamicDestinationsBindable();
|
||||
Map<String, BindingProperties> bindings = new HashMap<>();
|
||||
BindingProperties genericProperties = new BindingProperties();
|
||||
genericProperties.setContentType("text/plain");
|
||||
@@ -169,10 +168,10 @@ public class BinderAwareChannelResolverTests {
|
||||
matches("bar"), any(DirectChannel.class), any(ProducerProperties.class))).thenReturn(barBinding);
|
||||
when(mockBinderFactory.getBinder(null)).thenReturn(binder);
|
||||
when(mockBinderFactory.getBinder("someTransport")).thenReturn(binder2);
|
||||
ChannelBindingService channelBindingService = new ChannelBindingService(channelBindingServiceProperties, mockBinderFactory);
|
||||
@SuppressWarnings("unchecked")
|
||||
BinderAwareChannelResolver resolver =
|
||||
new BinderAwareChannelResolver(mockBinderFactory, this.channelBindingServiceProperties, dynamicDestinationsBindable,
|
||||
this.bindableChannelFactory);
|
||||
new BinderAwareChannelResolver(channelBindingService, this.bindableChannelFactory);
|
||||
BeanFactory beanFactory = new DefaultListableBeanFactory();
|
||||
resolver.setBeanFactory(beanFactory);
|
||||
SubscribableChannel resolved = (SubscribableChannel) resolver.resolveDestination("foo");
|
||||
@@ -182,12 +181,6 @@ public class BinderAwareChannelResolverTests {
|
||||
Assert.isTrue(dataTypes[0].equals(String.class), "Data type should be of type String");
|
||||
verify(binder).bindProducer(eq("foo"), any(MessageChannel.class), any(ProducerProperties.class));
|
||||
assertSame(resolved, beanFactory.getBean("foo"));
|
||||
resolved = (SubscribableChannel) resolver.resolveDestination("someTransport:bar");
|
||||
verify(binder2).bindProducer(eq("bar"), any(MessageChannel.class), any(ProducerProperties.class));
|
||||
assertSame(resolved, beanFactory.getBean("someTransport:bar"));
|
||||
assertTrue("Dynamic bindable should have two destination names", dynamicDestinationsBindable.getOutputs().size() == 2);
|
||||
assertTrue("Dynamic bindable should have the destination name 'foo'", dynamicDestinationsBindable.getOutputs().contains("foo"));
|
||||
assertTrue("Dynamic bindable should have the destination name 'bar'", dynamicDestinationsBindable.getOutputs().contains("someTransport:bar"));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -0,0 +1,188 @@
|
||||
/*
|
||||
* Copyright 2013-2016 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
|
||||
*
|
||||
* http://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;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.cloud.stream.binding.ChannelBindingService;
|
||||
import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
|
||||
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
|
||||
import org.springframework.cloud.stream.config.BindingProperties;
|
||||
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
|
||||
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.support.DefaultMessageBuilderFactory;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.integration.support.utils.IntegrationUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
|
||||
/**
|
||||
* @author Mark Fisher
|
||||
* @author Gary Russell
|
||||
* @author Ilayaperumal Gopinathan
|
||||
*/
|
||||
public class ExtendedPropertiesDynamicDestinationResolverTests extends DynamicDestinationResolverTests {
|
||||
|
||||
private volatile ExtendedPropertiesBinder<MessageChannel, ExtendedConsumerProperties, ExtendedProducerProperties> binder;
|
||||
|
||||
@Before
|
||||
@Override
|
||||
public void setupContext() throws Exception {
|
||||
this.binder = new TestBinder();
|
||||
BinderFactory binderFactory = new BinderFactory<MessageChannel>() {
|
||||
|
||||
@Override
|
||||
public ExtendedPropertiesBinder<MessageChannel, ExtendedConsumerProperties, ExtendedProducerProperties> getBinder(String configurationName) {
|
||||
return binder;
|
||||
}
|
||||
};
|
||||
this.channelBindingServiceProperties = new ChannelBindingServiceProperties();
|
||||
Map<String, BindingProperties> bindings = new HashMap<String, BindingProperties>();
|
||||
BindingProperties bindingProperties = new BindingProperties();
|
||||
bindingProperties.setContentType("text/plain");
|
||||
bindings.put("foo", bindingProperties);
|
||||
this.channelBindingServiceProperties.setBindings(bindings);
|
||||
MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer(
|
||||
this.channelBindingServiceProperties, new DefaultMessageBuilderFactory(),
|
||||
new CompositeMessageConverterFactory());
|
||||
messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class));
|
||||
messageConverterConfigurer.afterPropertiesSet();
|
||||
this.bindableChannelFactory = new DefaultBindableChannelFactory(messageConverterConfigurer);
|
||||
ChannelBindingService channelBindingService = new ChannelBindingService(channelBindingServiceProperties, binderFactory);
|
||||
this.resolver = new BinderAwareChannelResolver(channelBindingService, bindableChannelFactory);
|
||||
this.resolver.setBeanFactory(context.getBeanFactory());
|
||||
context.getBeanFactory().registerSingleton("channelResolver",
|
||||
this.resolver);
|
||||
context.registerSingleton("other", DirectChannel.class);
|
||||
context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
|
||||
DefaultMessageBuilderFactory.class);
|
||||
context.refresh();
|
||||
}
|
||||
|
||||
@Test
|
||||
@Override
|
||||
public void resolveChannel() {
|
||||
MessageChannel registered = resolver.resolveDestination("foo");
|
||||
DirectChannel testChannel = new DirectChannel();
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
final List<Message<?>> received = new ArrayList<>();
|
||||
testChannel.subscribe(new MessageHandler() {
|
||||
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
received.add(message);
|
||||
latch.countDown();
|
||||
}
|
||||
});
|
||||
binder.bindConsumer("foo", null, testChannel, new ExtendedConsumerProperties(new ConsumerProperties()));
|
||||
assertEquals(0, received.size());
|
||||
registered.send(MessageBuilder.withPayload("hello").build());
|
||||
try {
|
||||
assertTrue("latch timed out", latch.await(1, TimeUnit.SECONDS));
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
fail("interrupted while awaiting latch");
|
||||
}
|
||||
assertEquals(1, received.size());
|
||||
assertEquals("hello", received.get(0).getPayload());
|
||||
context.close();
|
||||
}
|
||||
|
||||
/**
|
||||
* A simple test binder that creates queues for the destinations. Ignores groups.
|
||||
*/
|
||||
private class TestBinder implements ExtendedPropertiesBinder<MessageChannel, ExtendedConsumerProperties, ExtendedProducerProperties> {
|
||||
|
||||
private final Map<String, DirectChannel> destinations = new ConcurrentHashMap<>();
|
||||
|
||||
@Override
|
||||
public Binding<MessageChannel> bindConsumer(String name, String group,
|
||||
MessageChannel inboundBindTarget, ExtendedConsumerProperties properties) {
|
||||
synchronized (destinations) {
|
||||
if (!destinations.containsKey(name)) {
|
||||
destinations.put(name, new DirectChannel());
|
||||
}
|
||||
}
|
||||
DirectHandler directHandler = new DirectHandler(inboundBindTarget);
|
||||
destinations.get(name).subscribe(directHandler);
|
||||
return new TestBinding(name, directHandler);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public Binding<MessageChannel> bindProducer(String name,
|
||||
MessageChannel outboundBindTarget, ExtendedProducerProperties properties) {
|
||||
synchronized (destinations) {
|
||||
if (!destinations.containsKey(name)) {
|
||||
destinations.put(name, new DirectChannel());
|
||||
}
|
||||
}
|
||||
DirectHandler directHandler = new DirectHandler(destinations.get(name));
|
||||
// for test purposes we can assume it is a SubscribableChannel
|
||||
((SubscribableChannel) outboundBindTarget).subscribe(directHandler);
|
||||
return new TestBinding(name, directHandler);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ExtendedConsumerProperties getExtendedConsumerProperties(String channelName) {
|
||||
return new ExtendedConsumerProperties(new ConsumerProperties());
|
||||
}
|
||||
|
||||
@Override
|
||||
public ExtendedProducerProperties getExtendedProducerProperties(String channelName) {
|
||||
return new ExtendedProducerProperties(new ProducerProperties());
|
||||
}
|
||||
|
||||
private class TestBinding implements Binding<MessageChannel> {
|
||||
|
||||
private final String name;
|
||||
|
||||
private final DirectHandler directHandler;
|
||||
|
||||
private TestBinding(String name, DirectHandler directHandler) {
|
||||
this.name = name;
|
||||
this.directHandler = directHandler;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbind() {
|
||||
destinations.get(name).unsubscribe(directHandler);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -203,9 +203,7 @@ public class ChannelBindingServiceTests {
|
||||
|
||||
@Test
|
||||
public void checkDynamicBinding() {
|
||||
|
||||
ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties();
|
||||
DynamicDestinationsBindable dynamicDestinationsBindable = new DynamicDestinationsBindable();
|
||||
DefaultBinderFactory<MessageChannel> binderFactory = new DefaultBinderFactory<>(
|
||||
Collections
|
||||
.singletonMap("mock",
|
||||
@@ -215,22 +213,23 @@ public class ChannelBindingServiceTests {
|
||||
MockBinderConfiguration.class }),
|
||||
new Properties(), true)));
|
||||
Binder binder = binderFactory.getBinder("mock");
|
||||
|
||||
MessageChannel inputChannel = new DirectChannel();
|
||||
@SuppressWarnings("unchecked")
|
||||
Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
|
||||
@SuppressWarnings("unchecked")
|
||||
final AtomicReference<MessageChannel> dynamic = new AtomicReference<>();
|
||||
when(binder.bindProducer(matches("bar"), any(DirectChannel.class),
|
||||
when(binder.bindProducer(matches("foo"), any(DirectChannel.class),
|
||||
any(ProducerProperties.class))).thenReturn(mockBinding);
|
||||
ChannelBindingService channelBindingService = new ChannelBindingService(properties, binderFactory);
|
||||
BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(
|
||||
binderFactory, properties, dynamicDestinationsBindable,
|
||||
channelBindingService,
|
||||
new DefaultBindableChannelFactory(new MessageConverterConfigurer(
|
||||
properties, new DefaultMessageBuilderFactory(),
|
||||
new CompositeMessageConverterFactory())));
|
||||
ConfigurableListableBeanFactory beanFactory = mock(
|
||||
ConfigurableListableBeanFactory.class);
|
||||
when(beanFactory.getBean("mock:bar", MessageChannel.class))
|
||||
when(beanFactory.getBean("foo", MessageChannel.class))
|
||||
.thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class));
|
||||
when(beanFactory.getBean("bar", MessageChannel.class))
|
||||
.thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class));
|
||||
doAnswer(new Answer<Void>() {
|
||||
|
||||
@@ -240,7 +239,7 @@ public class ChannelBindingServiceTests {
|
||||
return null;
|
||||
}
|
||||
|
||||
}).when(beanFactory).registerSingleton(eq("bar"), any(MessageChannel.class));
|
||||
}).when(beanFactory).registerSingleton(eq("foo"), any(MessageChannel.class));
|
||||
doAnswer(new Answer<Object>() {
|
||||
|
||||
@Override
|
||||
@@ -248,23 +247,23 @@ public class ChannelBindingServiceTests {
|
||||
return dynamic.get();
|
||||
}
|
||||
|
||||
}).when(beanFactory).initializeBean(any(MessageChannel.class), eq("bar"));
|
||||
}).when(beanFactory).initializeBean(any(MessageChannel.class), eq("foo"));
|
||||
resolver.setBeanFactory(beanFactory);
|
||||
MessageChannel resolved = resolver.resolveDestination("mock:bar");
|
||||
MessageChannel resolved = resolver.resolveDestination("foo");
|
||||
assertThat(resolved, sameInstance(dynamic.get()));
|
||||
verify(binder).bindProducer(eq("bar"), eq(dynamic.get()),
|
||||
verify(binder).bindProducer(eq("foo"), eq(dynamic.get()),
|
||||
any(ProducerProperties.class));
|
||||
properties.setDynamicDestinations(new String[] { "mock:bar" });
|
||||
resolved = resolver.resolveDestination("mock:bar");
|
||||
properties.setDynamicDestinations(new String[] { "foo" });
|
||||
resolved = resolver.resolveDestination("foo");
|
||||
assertThat(resolved, sameInstance(dynamic.get()));
|
||||
properties.setDynamicDestinations(new String[] { "foo:bar" });
|
||||
properties.setDynamicDestinations(new String[] { "test" });
|
||||
try {
|
||||
resolved = resolver.resolveDestination("mock:bar");
|
||||
resolved = resolver.resolveDestination("bar");
|
||||
fail();
|
||||
}
|
||||
catch (DestinationResolutionException e) {
|
||||
assertThat(e.getMessage(), containsString(
|
||||
"Failed to find MessageChannel bean with name 'mock:bar'"));
|
||||
"Failed to find MessageChannel bean with name 'bar'"));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user