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 ad6784f01..388ea6499 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 @@ -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 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" + - " [:]"); - } - } channel = this.bindableChannelFactory.createSubscribableChannel(channelName); - this.beanFactory.registerSingleton(beanName, channel); - channel = (MessageChannel) this.beanFactory.initializeBean(channel, beanName); - @SuppressWarnings("unchecked") - Binder binder = - (Binder) 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; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java index c2d301ac7..7e0c8fb1d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java @@ -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 binding = binder.bindConsumer(target, channelBindingServiceProperties.getGroup(inputChannelName), inputChannel, consumerProperties); bindings.add(binding); @@ -100,13 +92,10 @@ public class ChannelBindingService { (Binder) 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 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; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java deleted file mode 100644 index 63a8a0ec7..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java +++ /dev/null @@ -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 outputBindings = new HashMap<>(); - - public void addOutputBinding(String name, Binding binding) { - this.outputBindings.put(name, binding); - } - - @Override - public Set getOutputs() { - return Collections.unmodifiableSet(outputBindings.keySet()); - } - - @Override - public void unbindOutputs(ChannelBindingService adapter) { - for (Map.Entry entry: outputBindings.entrySet()) { - entry.getValue().unbind(); - } - outputBindings.clear(); - } -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ProducerConsumerPropertiesUtil.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ProducerConsumerPropertiesUtil.java new file mode 100644 index 000000000..d99e5c11a --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ProducerConsumerPropertiesUtil.java @@ -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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java index efbf505cf..993c793a8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java @@ -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 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 messageConverters = new ArrayList<>(); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DynamicDestinationResolverTests.java similarity index 85% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java rename to spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DynamicDestinationResolverTests.java index c10ab52ec..f0de280cd 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DynamicDestinationResolverTests.java @@ -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 binder; + protected volatile Binder 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 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")); } /** diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDynamicDestinationResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDynamicDestinationResolverTests.java new file mode 100644 index 000000000..a491e0e21 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDynamicDestinationResolverTests.java @@ -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 binder; + + @Before + @Override + public void setupContext() throws Exception { + this.binder = new TestBinder(); + BinderFactory binderFactory = new BinderFactory() { + + @Override + public ExtendedPropertiesBinder getBinder(String configurationName) { + return binder; + } + }; + this.channelBindingServiceProperties = new ChannelBindingServiceProperties(); + Map bindings = new HashMap(); + 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> 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 { + + private final Map destinations = new ConcurrentHashMap<>(); + + @Override + public Binding 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 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 { + + 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); + } + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java index db9a923fe..efa10da34 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java @@ -203,9 +203,7 @@ public class ChannelBindingServiceTests { @Test public void checkDynamicBinding() { - ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); - DynamicDestinationsBindable dynamicDestinationsBindable = new DynamicDestinationsBindable(); DefaultBinderFactory 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 mockBinding = Mockito.mock(Binding.class); @SuppressWarnings("unchecked") final AtomicReference 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() { @@ -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() { @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'")); } }