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

Minor polishing on dynamic channel binding

Browse Source
This commit is contained in:
Marius Bogoevici
2016-01-29 17:13:13 -05:00
parent 16e7893e49
commit 2f330c7bc4
2 changed files with 41 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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" +
" [<transport>:]<name>");
}
}
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" +
" [<transport>:]<name>");
}
}
Binder<MessageChannel> binder = binderFactory.getBinder(transport);
binder.bindProducer(name, channel, producerProperties);
}

33
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java
View File

@@ -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<MessageChannel> 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<MessageChannel> 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<MessageChannel> 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<MessageChannel> binderFactory =
new DefaultBinderFactory<>(Collections.singletonMap("mock",
new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}),
new Properties(), true)));
Binder<MessageChannel> binder = binderFactory.getBinder("mock");
MessageChannel inputChannel = new DirectChannel();
ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
Binding<MessageChannel> mockBinding = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class),
inputChannel, null);
final AtomicReference<MessageChannel> 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<Void>(){
@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();
}
}