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

Unbind dynamic producers

Browse Source 
- Add `SmartLifecycle` to BinderAwareChannelResolver
  - Add resolved destinations' bindings to the list of `Bindings` that need to be onbound
  - Add lowest phase for its lifecycle so that bindings are unbound at the latest

This resolves #370

Use a BindableAdapter to register the output channels

 - Add `DyanamicBindable` which is a BindableAdapter that stores the dynamic destination names and handle their unbinding
 - Inject `ChannelBindingService` into BinderAwareChannelResolver so that the dynamic binding gets registered with `producerBindings` and can subsequently be used when unbinding
 - Fix the `BinderAwareChannelResolver` to resolve the `transport:name` correctly by setting the bean name of the channel to use `name` only (without including transport:)
 - Update tests

Use Binding to unbind instead of ChannelBindingService
This commit is contained in:
Ilayaperumal Gopinathan
2016-03-01 19:35:29 +05:30
committed by Marius Bogoevici
parent 41a24a13e6
commit 66a78bf685
5 changed files with 112 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,8 +35,10 @@ import org.springframework.util.ObjectUtils;
* 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.
*
* @author Mark Fisher
* @author Gary Russell
* @author Ilayaperumal Gopinathan
*/
public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestinationResolver {
@@ -44,13 +46,16 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
private final ChannelBindingServiceProperties channelBindingServiceProperties;
private final DynamicBindable dynamicBindable;
private ConfigurableListableBeanFactory beanFactory;
public BinderAwareChannelResolver(BinderFactory<MessageChannel> binderFactory,
ChannelBindingServiceProperties channelBindingServiceProperties) {
public BinderAwareChannelResolver(BinderFactory binderFactory,
ChannelBindingServiceProperties channelBindingServiceProperties, DynamicBindable dynamicBindable) {
Assert.notNull(binderFactory, "'binderFactory' cannot be null");
this.binderFactory = binderFactory;
this.channelBindingServiceProperties = channelBindingServiceProperties;
this.dynamicBindable = dynamicBindable;
}
@Override
@@ -62,49 +67,45 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
}
@Override
public MessageChannel resolveDestination(String name) {
public MessageChannel resolveDestination(String destinationName) {
MessageChannel channel = null;
DestinationResolutionException destinationResolutionException;
try {
return super.resolveDestination(name);
return super.resolveDestination(destinationName);
}
catch (DestinationResolutionException e) {
destinationResolutionException = e;
}
synchronized (this) {
try {
return super.resolveDestination(name);
}
catch (DestinationResolutionException e) {
destinationResolutionException = e;
}
if (this.beanFactory != null && this.binderFactory != null) {
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);
producerProperties = this.channelBindingServiceProperties.getProducerProperties(destinationName);
}
boolean dynamicAllowed = ObjectUtils.isEmpty(dynamicDestinations)
|| ObjectUtils.containsElement(dynamicDestinations, name);
|| ObjectUtils.containsElement(dynamicDestinations, destinationName);
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);
String beanName = destinationName;
if (destinationName.contains(":")) {
String[] tokens = destinationName.split(":", 2);
if (tokens.length == 2) {
transport = tokens[0];
destinationName = tokens[1];
}
else if (tokens.length != 1) {
throw new IllegalArgumentException("Unrecognized channel naming scheme: " + name + " , should be" +
" [<transport>:]<name>");
throw new IllegalArgumentException("Unrecognized channel naming scheme: " + destinationName + " , should be" +
" [<transport>:]<destinationName>");
}
}
channel = new DirectChannel();
this.beanFactory.registerSingleton(beanName, channel);
channel = (MessageChannel) this.beanFactory.initializeBean(channel, beanName);
Binder<MessageChannel> binder = binderFactory.getBinder(transport);
binder.bindProducer(name, channel, producerProperties);
this.dynamicBindable.addDynamicOutputs(beanName, binder.bindProducer(destinationName, channel, producerProperties));
}
else {
throw destinationResolutionException;
@@ -113,5 +114,4 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
return channel;
}
}
}

52
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicBindable.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* 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 handle their unbinding.
*
* @author Ilayaperumal Gopinathan
*/
public final class DynamicBindable extends BindableAdapter {
/**
* Map containing dynamic destination names and their bindings.
*/
private Map<String, Binding> outputBindings = new HashMap<>();
void addDynamicOutputs(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();
}
}
}

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

@@ -39,6 +39,7 @@ 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.DynamicBindable;
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
@@ -134,10 +135,9 @@ public class ChannelBindingServiceConfiguration {
}
@Bean
public BinderAwareChannelResolver binderAwareChannelResolver(
BinderFactory<MessageChannel> binderFactory,
public BinderAwareChannelResolver binderAwareChannelResolver(BinderFactory<MessageChannel> binderFactory,
ChannelBindingServiceProperties channelBindingServiceProperties) {
return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties);
return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties, dynamicBindable());
}
@Bean
@@ -153,6 +153,11 @@ public class ChannelBindingServiceConfiguration {
return new SingleChannelBindable(ERROR_CHANNEL_NAME, errorChannel);
}
@Bean
public DynamicBindable dynamicBindable() {
return new DynamicBindable();
}
// IMPORTANT: Nested class to avoid instantiating all of the above early
@Configuration
protected static class PostProcessorConfiguration {

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

@@ -21,8 +21,8 @@ import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.matches;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -43,6 +43,7 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binding.DynamicBindable;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.context.support.StaticApplicationContext;
@@ -59,6 +60,7 @@ import org.springframework.messaging.SubscribableChannel;
/**
* @author Mark Fisher
* @author Gary Russell
* @author Ilayaperumal Gopinathan
*/
public class BinderAwareChannelResolverTests {
@@ -71,12 +73,13 @@ public class BinderAwareChannelResolverTests {
@Before
public void setupContext() throws Exception {
this.binder = new TestBinder();
this.resolver = new BinderAwareChannelResolver(new BinderFactory<MessageChannel>() {
BinderFactory binderFactory = new BinderFactory<MessageChannel>() {
@Override
public Binder<MessageChannel> getBinder(String configurationName) {
return binder;
}
}, null);
};
this.resolver = new BinderAwareChannelResolver(binderFactory, null, new DynamicBindable());
this.resolver.setBeanFactory(context.getBeanFactory());
context.getBeanFactory().registerSingleton("channelResolver",
this.resolver);
@@ -125,6 +128,7 @@ public class BinderAwareChannelResolverTests {
@SuppressWarnings("rawtypes")
public void propertyPassthrough() {
ChannelBindingServiceProperties bindingServiceProperties = new ChannelBindingServiceProperties();
DynamicBindable dynamicBindable = new DynamicBindable();
Map<String, BindingProperties> bindings = new HashMap<String, BindingProperties>();
BindingProperties bindingProperties = new BindingProperties();
bindingProperties.setContentType("text/plain");
@@ -132,19 +136,30 @@ public class BinderAwareChannelResolverTests {
bindingServiceProperties.setBindings(bindings);
@SuppressWarnings("unchecked")
Binder<MessageChannel> binder = mock(Binder.class);
Binder<MessageChannel> binder2 = mock(Binder.class);
BinderFactory mockBinderFactory = Mockito.mock(BinderFactory.class);
when(mockBinderFactory.getBinder(anyString())).thenReturn(binder);
Binding<MessageChannel> fooBinding = Mockito.mock(Binding.class);
Binding<MessageChannel> barBinding = Mockito.mock(Binding.class);
when(binder.bindProducer(
matches("foo"), any(DirectChannel.class), any(Properties.class))).thenReturn(fooBinding);
when(binder2.bindProducer(
matches("bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(barBinding);
when(mockBinderFactory.getBinder(null)).thenReturn(binder);
when(mockBinderFactory.getBinder("someTransport")).thenReturn(binder2);
@SuppressWarnings("unchecked")
BinderAwareChannelResolver resolver =
new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties);
new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties, dynamicBindable);
BeanFactory beanFactory = new DefaultListableBeanFactory();
resolver.setBeanFactory(beanFactory);
MessageChannel resolved = resolver.resolveDestination("foo");
verify(binder).bindProducer(eq("foo"), any(MessageChannel.class), any(Properties.class));
assertSame(resolved, beanFactory.getBean("foo"));
resolved = resolver.resolveDestination("someTransport:foo");
verify(binder).bindProducer(eq("someTransport:foo"), any(MessageChannel.class), any(Properties.class));
assertSame(resolved, beanFactory.getBean("someTransport:foo"));
resolved = resolver.resolveDestination("someTransport:bar");
verify(binder2).bindProducer(eq("bar"), any(MessageChannel.class), any(Properties.class));
assertSame(resolved, beanFactory.getBean("someTransport:bar"));
assertTrue("Dynamic bindable should have two destination names", dynamicBindable.getOutputs().size() == 2);
assertTrue("Dynamic bindable should have the destination name 'foo'", dynamicBindable.getOutputs().contains("foo"));
assertTrue("Dynamic bindable should have the destination name 'bar'", dynamicBindable.getOutputs().contains("someTransport:bar"));
}
/**
@@ -156,7 +171,7 @@ public class BinderAwareChannelResolverTests {
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel inboundBindTarget,
Properties properties) {
Properties properties) {
synchronized (destinations) {
if (!destinations.containsKey(name)) {
destinations.put(name, new DirectChannel());

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

@@ -61,6 +61,7 @@ import org.springframework.messaging.core.DestinationResolutionException;
* @author Gary Russell
* @author Mark Fisher
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class ChannelBindingServiceTests {
@@ -181,6 +182,7 @@ public class ChannelBindingServiceTests {
public void checkDynamicBinding () {
ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties();
DynamicBindable dynamicBindable = new DynamicBindable();
DefaultBinderFactory<MessageChannel> binderFactory =
new DefaultBinderFactory<>(Collections.singletonMap("mock",
new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}),
@@ -193,8 +195,8 @@ public class ChannelBindingServiceTests {
@SuppressWarnings("unchecked")
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);
matches("bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(mockBinding);
BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties, dynamicBindable);
ConfigurableListableBeanFactory beanFactory = mock(ConfigurableListableBeanFactory.class);
when(beanFactory.getBean("mock:bar", MessageChannel.class))
.thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class));
@@ -206,7 +208,7 @@ public class ChannelBindingServiceTests {
return null;
}
}).when(beanFactory).registerSingleton(eq("mock:bar"), any(MessageChannel.class));
}).when(beanFactory).registerSingleton(eq("bar"), any(MessageChannel.class));
doAnswer(new Answer<Object>() {
@Override
@@ -214,11 +216,11 @@ public class ChannelBindingServiceTests {
return dynamic.get();
}
}).when(beanFactory).initializeBean(any(MessageChannel.class), eq("mock:bar"));
}).when(beanFactory).initializeBean(any(MessageChannel.class), eq("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));
verify(binder).bindProducer(eq("bar"), eq(dynamic.get()), any(Properties.class));
properties.setDynamicDestinations(new String[] { "mock:bar" });
resolved = resolver.resolveDestination("mock:bar");
assertThat(resolved, sameInstance(dynamic.get()));