From 2f330c7bc477c65bd2dc571a5507164eca3d3bc3 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Fri, 29 Jan 2016 17:13:13 -0500 Subject: [PATCH] Minor polishing on dynamic channel binding --- .../binding/BinderAwareChannelResolver.java | 35 +++++++++---------- .../binding/ChannelBindingServiceTests.java | 33 ++++++++++++----- 2 files changed, 41 insertions(+), 27 deletions(-) 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 c344c2f34..bde9a86a6 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 @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.binding; -import java.util.Arrays; import java.util.Properties; import org.springframework.beans.factory.BeanFactory; @@ -29,6 +28,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; /** * A {@link org.springframework.messaging.core.DestinationResolver} implementation that @@ -79,20 +79,6 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina destinationResolutionException = e; } if (this.beanFactory != null && this.binderFactory != null) { - channel = new DirectChannel(); - this.beanFactory.registerSingleton(name, channel); - channel = (MessageChannel) this.beanFactory.initializeBean(channel, name); - String transport = null; - if (name.contains(":")) { - String[] tokens = name.split(":", 2); - if (tokens.length == 2) { - transport = tokens[0]; - } - else if (tokens.length != 1) { - throw new IllegalArgumentException("Unrecognized channel naming scheme: " + name + " , should be" + - " [:]"); - } - } String[] dynamicDestinations = null; Properties producerProperties = null; if (this.channelBindingServiceProperties != null) { @@ -100,10 +86,23 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina // TODO: need the props to return some defaults if not found producerProperties = this.channelBindingServiceProperties.getProducerProperties(name); } - boolean dynamicAllowed = dynamicDestinations == null - || dynamicDestinations.length == 0 - || Arrays.asList(dynamicDestinations).contains(name); + boolean dynamicAllowed = ObjectUtils.isEmpty(dynamicDestinations) + || ObjectUtils.containsElement(dynamicDestinations, name); if (dynamicAllowed) { + channel = new DirectChannel(); + this.beanFactory.registerSingleton(name, channel); + channel = (MessageChannel) this.beanFactory.initializeBean(channel, name); + String transport = null; + if (name.contains(":")) { + String[] tokens = name.split(":", 2); + if (tokens.length == 2) { + transport = tokens[0]; + } + else if (tokens.length != 1) { + throw new IllegalArgumentException("Unrecognized channel naming scheme: " + name + " , should be" + + " [:]"); + } + } Binder binder = binderFactory.getBinder(transport); binder.bindProducer(name, channel, producerProperties); } 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 b196d931e..ecc2574be 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 @@ -35,7 +35,6 @@ import java.util.Map; import java.util.Properties; import java.util.concurrent.atomic.AtomicReference; -import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; @@ -81,10 +80,10 @@ public class ChannelBindingServiceTests { MessageChannel inputChannel = new DirectChannel(); Binding mockBinding = Binding.forConsumer("foo", null, Mockito.mock(AbstractEndpoint.class), inputChannel, null); - Mockito.when(binder.bindConsumer("foo", null, inputChannel, new Properties())) + when(binder.bindConsumer("foo", null, inputChannel, new Properties())) .thenReturn(mockBinding); Binding binding = service.bindConsumer(inputChannel, name); - Assert.assertThat(binding, sameInstance(mockBinding)); + assertThat(binding, sameInstance(mockBinding)); service.unbindConsumers(name); verify(binder).bindConsumer(name, props.getGroup(), inputChannel, properties.getConsumerProperties(name)); verify(binder).unbind(binding); @@ -115,13 +114,34 @@ public class ChannelBindingServiceTests { Binding binding = service.bindConsumer(inputChannel, name); assertThat(binding, sameInstance(mockBinding)); + service.unbindConsumers(name); + verify(binder).bindConsumer(name, props.getGroup(), inputChannel, properties.getConsumerProperties(name)); + verify(binder).unbind(binding); + binderFactory.destroy(); + } + + @Test + public void checkDynamicBinding () { + + ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); + DefaultBinderFactory binderFactory = + new DefaultBinderFactory<>(Collections.singletonMap("mock", + new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), + new Properties(), true))); + Binder binder = binderFactory.getBinder("mock"); + + MessageChannel inputChannel = new DirectChannel(); + ChannelBindingService service = new ChannelBindingService(properties, binderFactory); + Binding mockBinding = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class), + inputChannel, null); + final AtomicReference dynamic = new AtomicReference<>(); when(binder.bindProducer( matches("mock:bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(mockBinding); BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties); ConfigurableListableBeanFactory beanFactory = mock(ConfigurableListableBeanFactory.class); when(beanFactory.getBean("mock:bar", MessageChannel.class)) - .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class)); + .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class)); doAnswer(new Answer(){ @Override @@ -154,11 +174,6 @@ public class ChannelBindingServiceTests { catch (DestinationResolutionException e) { assertThat(e.getMessage(), containsString("Failed to find MessageChannel bean with name 'mock:bar'")); } - - service.unbindConsumers(name); - verify(binder).bindConsumer(name, props.getGroup(), inputChannel, properties.getConsumerProperties(name)); - verify(binder).unbind(binding); - binderFactory.destroy(); } }