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

scsm-88: Support Dynamic Binding Channel Limits

Browse Source 
Required by scsm-88.

If `dynamicDestinations` is not zero length, only dynamically bind if
the channel is is in the list.

Polishing - Test Support for Dynamic Bindings
This commit is contained in:
Gary Russell
2015-12-21 18:13:25 -05:00
committed by Marius Bogoevici
parent 31961ae595
commit 16e7893e49
6 changed files with 134 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java
View File

@@ -20,6 +20,8 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingDeque;
import org.springframework.cloud.stream.binder.Binder;
@@ -47,6 +49,7 @@ public class TestSupportBinder implements Binder<MessageChannel> {
private final MessageCollectorImpl messageCollector = new MessageCollectorImpl();
private final ConcurrentMap<String, MessageChannel> messageChannels = new ConcurrentHashMap<>();
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel inboundBindTarget, Properties properties) {
@@ -65,19 +68,25 @@ public class TestSupportBinder implements Binder<MessageChannel> {
queue.add(message);
}
});
this.messageChannels.put(name, outboundBindTarget);
return null;
}
@Override
public void unbind(Binding<MessageChannel> binding) {
if (Binding.Type.producer.equals(binding.getType()))
messageCollector.unregister(binding.getTarget());
if (Binding.Type.producer.equals(binding.getType())) {
messageCollector.unregister(binding.getTarget());
}
}
public MessageCollector messageCollector() {
return messageCollector;
}
public MessageChannel getChannelForName(String name) {
return this.messageChannels.get(name);
}
/**
* Maintains mappings between channels and queues.
*

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

@@ -16,17 +16,19 @@
package org.springframework.cloud.stream.binding;
import java.util.Arrays;
import java.util.Properties;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
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.config.ChannelBindingServiceProperties;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.util.Assert;
/**
* A {@link org.springframework.messaging.core.DestinationResolver} implementation that
@@ -40,36 +42,41 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
private final BinderFactory<MessageChannel> binderFactory;
private final Properties producerProperties;
private final ChannelBindingServiceProperties channelBindingServiceProperties;
private DefaultListableBeanFactory beanFactory;
private ConfigurableListableBeanFactory beanFactory;
public BinderAwareChannelResolver(BinderFactory<MessageChannel> binderFactory, Properties producerProperties) {
public BinderAwareChannelResolver(BinderFactory<MessageChannel> binderFactory,
ChannelBindingServiceProperties channelBindingServiceProperties) {
Assert.notNull(binderFactory, "'binderFactory' cannot be null");
this.binderFactory = binderFactory;
this.producerProperties = producerProperties;
this.channelBindingServiceProperties = channelBindingServiceProperties;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
super.setBeanFactory(beanFactory);
if (beanFactory instanceof ConfigurableBeanFactory) {
this.beanFactory = (DefaultListableBeanFactory) beanFactory;
if (beanFactory instanceof ConfigurableListableBeanFactory) {
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
}
}
@Override
public MessageChannel resolveDestination(String name) {
MessageChannel channel = null;
DestinationResolutionException destinationResolutionException;
try {
return super.resolveDestination(name);
}
catch (DestinationResolutionException e) {
destinationResolutionException = e;
}
synchronized (this) {
try {
return super.resolveDestination(name);
}
catch (DestinationResolutionException e) {
destinationResolutionException = e;
}
if (this.beanFactory != null && this.binderFactory != null) {
channel = new DirectChannel();
@@ -86,8 +93,23 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
" [<transport>:]<name>");
}
}
Binder<MessageChannel> binder = binderFactory.getBinder(transport);
binder.bindProducer(name, channel, this.producerProperties);
String[] dynamicDestinations = null;
Properties producerProperties = null;
if (this.channelBindingServiceProperties != null) {
dynamicDestinations = this.channelBindingServiceProperties.getDynamicDestinations();
// 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);
if (dynamicAllowed) {
Binder<MessageChannel> binder = binderFactory.getBinder(transport);
binder.bindProducer(name, channel, producerProperties);
}
else {
throw destinationResolutionException;
}
}
return channel;
}

11
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* Copyright 2015-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.
@@ -20,7 +20,6 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactoryUtils;
@@ -31,7 +30,6 @@ import org.springframework.boot.context.properties.ConfigurationPropertiesBindin
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.MessageChannelConfigurer;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
import org.springframework.cloud.stream.binding.ChannelBindingService;
@@ -39,6 +37,7 @@ import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigure
import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener;
import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory;
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
import org.springframework.cloud.stream.binding.MessageHistoryTrackerConfigurer;
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
@@ -62,6 +61,7 @@ import org.springframework.messaging.core.DestinationResolver;
* @author David Turanski
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Gary Russell
*/
@Configuration
@EnableConfigurationProperties(ChannelBindingServiceProperties.class)
@@ -124,8 +124,9 @@ public class ChannelBindingServiceConfiguration {
@Bean
public BinderAwareChannelResolver binderAwareChannelResolver(
BinderFactory<MessageChannel> binderFactory) {
return new BinderAwareChannelResolver(binderFactory, new Properties());
BinderFactory<MessageChannel> binderFactory,
ChannelBindingServiceProperties channelBindingServiceProperties) {
return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties);
}
@Bean

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java
View File

@@ -50,6 +50,8 @@ public class ChannelBindingServiceProperties {
private String defaultBinder;
private String[] dynamicDestinations = new String[0];
public Map<String, BindingProperties> getBindings() {
return bindings;
}
@@ -90,6 +92,14 @@ public class ChannelBindingServiceProperties {
this.instanceCount = instanceCount;
}
public String[] getDynamicDestinations() {
return dynamicDestinations;
}
public void setDynamicDestinations(String[] dynamicDestinations) {
this.dynamicDestinations = dynamicDestinations;
}
public String getBindingDestination(String channelName) {
BindingProperties bindingProperties = bindings.get(channelName);
// we may shortcut directly to the path

22
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
View File

@@ -25,9 +25,12 @@ import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -40,6 +43,8 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.cloud.stream.binder.local.LocalMessageChannelBinder;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.scheduling.PollerMetadata;
@@ -128,21 +133,26 @@ public class BinderAwareChannelResolverTests {
@Test
@SuppressWarnings("rawtypes")
public void propertyPassthrough() {
Properties properties = new Properties();
ChannelBindingServiceProperties bindingServiceProperties = new ChannelBindingServiceProperties();
Map<String, BindingProperties> bindings = new HashMap<String, BindingProperties>();
BindingProperties bindingProperties = new BindingProperties();
bindingProperties.setContentType("text/plain");
bindings.put("foo", bindingProperties);
bindingServiceProperties.setBindings(bindings);
@SuppressWarnings("unchecked")
Binder<MessageChannel> binderFactory = mock(Binder.class);
Binder<MessageChannel> binder = mock(Binder.class);
BinderFactory mockBinderFactory = Mockito.mock(BinderFactory.class);
Mockito.when(mockBinderFactory.getBinder(anyString())).thenReturn(binderFactory);
when(mockBinderFactory.getBinder(anyString())).thenReturn(binder);
@SuppressWarnings("unchecked")
BinderAwareChannelResolver resolver =
new BinderAwareChannelResolver(mockBinderFactory, properties);
new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties);
BeanFactory beanFactory = new DefaultListableBeanFactory();
resolver.setBeanFactory(beanFactory);
MessageChannel resolved = resolver.resolveDestination("foo");
verify(binderFactory).bindProducer(eq("foo"), any(MessageChannel.class), eq(properties));
verify(binder).bindProducer(eq("foo"), any(MessageChannel.class), any(Properties.class));
assertSame(resolved, beanFactory.getBean("foo"));
resolved = resolver.resolveDestination("someTransport:foo");
verify(binderFactory).bindProducer(eq("someTransport:foo"), any(MessageChannel.class), eq(properties));
verify(binder).bindProducer(eq("someTransport:foo"), any(MessageChannel.class), any(Properties.class));
assertSame(resolved, beanFactory.getBean("someTransport:foo"));
}

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

@@ -16,18 +16,33 @@
package org.springframework.cloud.stream.binding;
import static org.hamcrest.CoreMatchers.sameInstance;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.matches;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Collections;
import java.util.HashMap;
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;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderConfiguration;
import org.springframework.cloud.stream.binder.BinderType;
@@ -39,6 +54,7 @@ import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolutionException;
/**
* @author Gary Russell
@@ -94,10 +110,51 @@ public class ChannelBindingServiceTests {
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding = Binding.forConsumer("foo", "fooGroup", Mockito.mock(AbstractEndpoint.class),
inputChannel, null);
Mockito.when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties()))
when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties()))
.thenReturn(mockBinding);
Binding<MessageChannel> binding = service.bindConsumer(inputChannel, name);
Assert.assertThat(binding, sameInstance(mockBinding));
assertThat(binding, sameInstance(mockBinding));
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));
doAnswer(new Answer<Void>(){
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
dynamic.set(invocation.getArgumentAt(1, MessageChannel.class));
return null;
}
}).when(beanFactory).registerSingleton(eq("mock:bar"), any(MessageChannel.class));
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return dynamic.get();
}
}).when(beanFactory).initializeBean(any(MessageChannel.class), eq("mock:bar"));
resolver.setBeanFactory(beanFactory);
MessageChannel resolved = resolver.resolveDestination("mock:bar");
assertThat(resolved, sameInstance(dynamic.get()));
verify(binder).bindProducer(eq("mock:bar"), eq(dynamic.get()), any(Properties.class));
properties.setDynamicDestinations(new String[] { "mock:bar" });
resolved = resolver.resolveDestination("mock:bar");
assertThat(resolved, sameInstance(dynamic.get()));
properties.setDynamicDestinations(new String[] { "foo:bar" });
try {
resolved = resolver.resolveDestination("mock:bar");
fail();
}
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);