GH-1127: Add stub Binder backed by SI
Resolves: spring-cloud/spring-cloud-stream#1127 - Added initial implementation of SI-backed stub binder to facilitate more consistent testing - Improved BinderAwareChannelResolverTests and ExtendedPropertiesBinderAwareChannelResolverTests to make use of it - Added sample Application that demonstrates the binder usage Addressed PR comments
This commit is contained in:
Oleg Zhurakousky
committed by
Artem Bilan
Artem Bilan
parent
024e3cffcc
commit
21a2a84d86
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2013-2016 the original author or authors.
|
||||
* Copyright 2013-2017 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.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
@@ -28,25 +27,19 @@ import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration;
|
||||
import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory;
|
||||
import org.springframework.cloud.stream.binding.Bindable;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
|
||||
import org.springframework.cloud.stream.binding.BindingService;
|
||||
import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
|
||||
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
|
||||
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
|
||||
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
|
||||
import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory;
|
||||
import org.springframework.cloud.stream.config.BindingProperties;
|
||||
import org.springframework.cloud.stream.config.BindingServiceProperties;
|
||||
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
|
||||
import org.springframework.context.support.StaticApplicationContext;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
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;
|
||||
@@ -54,10 +47,11 @@ import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
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.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.ArgumentMatchers.matches;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
@@ -66,10 +60,11 @@ import static org.mockito.Mockito.when;
|
||||
* @author Mark Fisher
|
||||
* @author Gary Russell
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author Oleg Zhurakousky
|
||||
*/
|
||||
public class BinderAwareChannelResolverTests {
|
||||
|
||||
protected final StaticApplicationContext context = new StaticApplicationContext();
|
||||
protected ConfigurableApplicationContext context;
|
||||
|
||||
protected volatile BinderAwareChannelResolver resolver;
|
||||
|
||||
@@ -81,68 +76,43 @@ public class BinderAwareChannelResolverTests {
|
||||
|
||||
protected volatile DynamicDestinationsBindable dynamicDestinationsBindable;
|
||||
|
||||
private volatile List<TestBinder.TestBinding> producerBindings;
|
||||
|
||||
@Before
|
||||
public void setupContext() throws Exception {
|
||||
producerBindings = new ArrayList<>();
|
||||
this.binder = new TestBinder();
|
||||
BinderFactory binderFactory = new BinderFactory() {
|
||||
|
||||
@Override
|
||||
public <T> Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties> getBinder(
|
||||
String configurationName, Class<? extends T> bindableType) {
|
||||
return (Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties>) binder;
|
||||
}
|
||||
};
|
||||
this.bindingServiceProperties = new BindingServiceProperties();
|
||||
Map<String, BindingProperties> bindings = new HashMap<>();
|
||||
BindingProperties bindingProperties = new BindingProperties();
|
||||
bindingProperties.setContentType("text/plain");
|
||||
bindings.put("foo", bindingProperties);
|
||||
this.bindingServiceProperties.setBindings(bindings);
|
||||
BindingService bindingService = new BindingService(bindingServiceProperties,
|
||||
binderFactory);
|
||||
MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer(
|
||||
this.bindingServiceProperties,
|
||||
new CompositeMessageConverterFactory());
|
||||
messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class));
|
||||
messageConverterConfigurer.afterPropertiesSet();
|
||||
this.bindingTargetFactory = new SubscribableChannelBindingTargetFactory(messageConverterConfigurer);
|
||||
dynamicDestinationsBindable = new DynamicDestinationsBindable();
|
||||
this.resolver = new BinderAwareChannelResolver(bindingService, this.bindingTargetFactory,
|
||||
dynamicDestinationsBindable);
|
||||
this.resolver.setBeanFactory(context.getBeanFactory());
|
||||
context.getBeanFactory().registerSingleton("channelResolver", this.resolver);
|
||||
context.getBeanFactory().registerSingleton("dynamicDestinationBindable", this.dynamicDestinationsBindable);
|
||||
context.registerSingleton("other", DirectChannel.class);
|
||||
context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
|
||||
DefaultMessageBuilderFactory.class);
|
||||
context.getBeanFactory().registerSingleton("bindingService", bindingService);
|
||||
context.registerSingleton("inputBindingLifecycle", InputBindingLifecycle.class);
|
||||
context.registerSingleton("outputBindingLifecycle", OutputBindingLifecycle.class);
|
||||
context.refresh();
|
||||
this.context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration()).web(WebApplicationType.NONE).run();
|
||||
this.resolver = context.getBean(BinderAwareChannelResolver.class);
|
||||
this.binder = context.getBean(Binder.class);
|
||||
this.bindingServiceProperties = context.getBean(BindingServiceProperties.class);
|
||||
this.bindingTargetFactory = context.getBean(AbstractBindingTargetFactory.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolveChannel() {
|
||||
assertThat(producerBindings).hasSize(0);
|
||||
Map<String, Bindable> bindables = context.getBeansOfType(Bindable.class);
|
||||
assertThat(bindables).hasSize(1);
|
||||
for (Bindable bindable : bindables.values()) {
|
||||
assertEquals(0, bindable.getInputs().size()); // producer
|
||||
assertEquals(0, bindable.getOutputs().size());// consumer
|
||||
}
|
||||
MessageChannel registered = resolver.resolveDestination("foo");
|
||||
assertThat(producerBindings).hasSize(1);
|
||||
TestBinder.TestBinding binding = producerBindings.get(0);
|
||||
assertThat(binding.isBound()).describedAs("Must be bound");
|
||||
|
||||
bindables = context.getBeansOfType(Bindable.class);
|
||||
assertThat(bindables).hasSize(1);
|
||||
for (Bindable bindable : bindables.values()) {
|
||||
assertEquals(0, bindable.getInputs().size()); // producer
|
||||
assertEquals(1, bindable.getOutputs().size());// consumer
|
||||
}
|
||||
DirectChannel testChannel = new DirectChannel();
|
||||
testChannel.setComponentName("INPUT");
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
final List<Message<?>> 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 ConsumerProperties());
|
||||
this.binder.bindConsumer("foo", null, testChannel, new ConsumerProperties());
|
||||
assertThat(received).hasSize(0);
|
||||
registered.send(MessageBuilder.withPayload("hello").build());
|
||||
try {
|
||||
@@ -154,15 +124,18 @@ public class BinderAwareChannelResolverTests {
|
||||
}
|
||||
assertThat(received).hasSize(1);
|
||||
assertThat(new String((byte[])received.get(0).getPayload())).isEqualTo("hello");
|
||||
context.close();
|
||||
assertThat(producerBindings).hasSize(1);
|
||||
assertThat(binding.isBound()).isFalse().describedAs("Must not be bound");
|
||||
this.context.close();
|
||||
for (Bindable bindable : bindables.values()) {
|
||||
assertEquals(0, bindable.getInputs().size());
|
||||
assertEquals(0, bindable.getOutputs().size());//Must not be bound"
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolveNonRegisteredChannel() {
|
||||
MessageChannel other = resolver.resolveDestination("other");
|
||||
assertThat(context.getBean("other")).isSameAs(other);
|
||||
this.context.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -173,7 +146,6 @@ public class BinderAwareChannelResolverTests {
|
||||
genericProperties.setContentType("text/plain");
|
||||
bindings.put("foo", genericProperties);
|
||||
this.bindingServiceProperties.setBindings(bindings);
|
||||
@SuppressWarnings("unchecked")
|
||||
Binder binder = mock(Binder.class);
|
||||
Binder binder2 = mock(Binder.class);
|
||||
BinderFactory mockBinderFactory = Mockito.mock(BinderFactory.class);
|
||||
@@ -187,74 +159,12 @@ public class BinderAwareChannelResolverTests {
|
||||
when(mockBinderFactory.getBinder("someTransport", DirectChannel.class)).thenReturn(binder2);
|
||||
BindingService bindingService = new BindingService(bindingServiceProperties,
|
||||
mockBinderFactory);
|
||||
@SuppressWarnings("unchecked")
|
||||
BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(bindingService, this.bindingTargetFactory,
|
||||
new DynamicDestinationsBindable());
|
||||
BeanFactory beanFactory = new DefaultListableBeanFactory();
|
||||
resolver.setBeanFactory(beanFactory);
|
||||
resolver.setBeanFactory(context.getBeanFactory());
|
||||
SubscribableChannel resolved = (SubscribableChannel) resolver.resolveDestination("foo");
|
||||
verify(binder).bindProducer(eq("foo"), any(MessageChannel.class), any(ProducerProperties.class));
|
||||
assertThat(resolved).isSameAs(beanFactory.getBean("foo"));
|
||||
}
|
||||
|
||||
/**
|
||||
* A simple test binder that creates queues for the destinations. Ignores groups.
|
||||
*/
|
||||
private class TestBinder implements Binder<MessageChannel, ConsumerProperties, ProducerProperties> {
|
||||
|
||||
private final Map<String, DirectChannel> destinations = new ConcurrentHashMap<>();
|
||||
|
||||
@Override
|
||||
public Binding<MessageChannel> bindConsumer(String name, String group,
|
||||
MessageChannel inboundBindTarget, ConsumerProperties 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<MessageChannel> bindProducer(String name,
|
||||
MessageChannel outboundBindTarget, ProducerProperties 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);
|
||||
TestBinding binding = new TestBinding(name, directHandler);
|
||||
producerBindings.add(binding);
|
||||
return binding;
|
||||
}
|
||||
|
||||
private final class TestBinding implements Binding<MessageChannel> {
|
||||
|
||||
private final String name;
|
||||
|
||||
private final DirectHandler directHandler;
|
||||
|
||||
private boolean bound = true;
|
||||
|
||||
private TestBinding(String name, DirectHandler directHandler) {
|
||||
this.name = name;
|
||||
this.directHandler = directHandler;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbind() {
|
||||
bound = false;
|
||||
destinations.get(name).unsubscribe(directHandler);
|
||||
}
|
||||
|
||||
public boolean isBound() {
|
||||
return bound;
|
||||
}
|
||||
}
|
||||
assertThat(resolved).isSameAs(context.getBean("foo"));
|
||||
this.context.close();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2013-2016 the original author or authors.
|
||||
* Copyright 2013-2017 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.
|
||||
@@ -17,102 +17,49 @@
|
||||
package org.springframework.cloud.stream.binder;
|
||||
|
||||
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.BinderAwareChannelResolver;
|
||||
import org.springframework.cloud.stream.binding.BindingService;
|
||||
import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
|
||||
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
|
||||
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
|
||||
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
|
||||
import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory;
|
||||
import org.springframework.cloud.stream.config.BindingProperties;
|
||||
import org.springframework.cloud.stream.config.BindingServiceProperties;
|
||||
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
|
||||
import org.springframework.cloud.stream.binding.Bindable;
|
||||
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;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
/**
|
||||
* @author Mark Fisher
|
||||
* @author Gary Russell
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author Oleg Zhurakousky
|
||||
*/
|
||||
public class ExtendedPropertiesBinderAwareChannelResolverTests extends BinderAwareChannelResolverTests {
|
||||
|
||||
private volatile ExtendedPropertiesBinder<MessageChannel, ExtendedConsumerProperties, ExtendedProducerProperties> binder;
|
||||
|
||||
private volatile List<TestBinder.TestBinding> producerBindings;
|
||||
|
||||
@Before
|
||||
@Override
|
||||
public void setupContext() throws Exception {
|
||||
producerBindings = new ArrayList<>();
|
||||
this.binder = new TestBinder();
|
||||
BinderFactory binderFactory = new BinderFactory() {
|
||||
|
||||
@Override
|
||||
public <T> Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties> getBinder(
|
||||
String configurationName, Class<? extends T> bindableType) {
|
||||
return (Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties>) binder;
|
||||
}
|
||||
};
|
||||
this.bindingServiceProperties = new BindingServiceProperties();
|
||||
Map<String, BindingProperties> bindings = new HashMap<String, BindingProperties>();
|
||||
BindingProperties bindingProperties = new BindingProperties();
|
||||
bindingProperties.setContentType("text/plain");
|
||||
bindings.put("foo", bindingProperties);
|
||||
this.bindingServiceProperties.setBindings(bindings);
|
||||
MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer(
|
||||
this.bindingServiceProperties,
|
||||
new CompositeMessageConverterFactory());
|
||||
messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class));
|
||||
messageConverterConfigurer.afterPropertiesSet();
|
||||
this.bindingTargetFactory = new SubscribableChannelBindingTargetFactory(messageConverterConfigurer);
|
||||
dynamicDestinationsBindable = new DynamicDestinationsBindable();
|
||||
BindingService bindingService = new BindingService(bindingServiceProperties,
|
||||
binderFactory);
|
||||
this.resolver = new BinderAwareChannelResolver(bindingService, this.bindingTargetFactory,
|
||||
dynamicDestinationsBindable);
|
||||
this.resolver.setBeanFactory(context.getBeanFactory());
|
||||
context.getBeanFactory().registerSingleton("channelResolver", this.resolver);
|
||||
context.getBeanFactory().registerSingleton("dynamicDestinationBindable", this.dynamicDestinationsBindable);
|
||||
context.registerSingleton("other", DirectChannel.class);
|
||||
context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
|
||||
DefaultMessageBuilderFactory.class);
|
||||
context.getBeanFactory().registerSingleton("bindingService", bindingService);
|
||||
context.registerSingleton("inputBindingLifecycle", InputBindingLifecycle.class);
|
||||
context.registerSingleton("outputBindingLifecycle", OutputBindingLifecycle.class);
|
||||
context.refresh();
|
||||
}
|
||||
|
||||
@Test
|
||||
@Override
|
||||
public void resolveChannel() {
|
||||
assertThat(producerBindings).hasSize(0);
|
||||
Map<String, Bindable> bindables = context.getBeansOfType(Bindable.class);
|
||||
assertThat(bindables).hasSize(1);
|
||||
for (Bindable bindable : bindables.values()) {
|
||||
assertEquals(0, bindable.getInputs().size()); // producer
|
||||
assertEquals(0, bindable.getOutputs().size());// consumer
|
||||
}
|
||||
MessageChannel registered = resolver.resolveDestination("foo");
|
||||
assertThat(producerBindings).hasSize(1);
|
||||
TestBinder.TestBinding binding = producerBindings.get(0);
|
||||
assertThat(binding.isBound()).describedAs("Must be bound");
|
||||
bindables = context.getBeansOfType(Bindable.class);
|
||||
assertThat(bindables).hasSize(1);
|
||||
for (Bindable bindable : bindables.values()) {
|
||||
assertEquals(0, bindable.getInputs().size()); // producer
|
||||
assertEquals(1, bindable.getOutputs().size());// consumer
|
||||
}
|
||||
DirectChannel testChannel = new DirectChannel();
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
final List<Message<?>> received = new ArrayList<>();
|
||||
@@ -124,7 +71,7 @@ public class ExtendedPropertiesBinderAwareChannelResolverTests extends BinderAwa
|
||||
latch.countDown();
|
||||
}
|
||||
});
|
||||
binder.bindConsumer("foo", null, testChannel, new ExtendedConsumerProperties(new ConsumerProperties()));
|
||||
binder.bindConsumer("foo", null, testChannel, new ExtendedConsumerProperties<ConsumerProperties>(new ConsumerProperties()));
|
||||
assertThat(received).hasSize(0);
|
||||
registered.send(MessageBuilder.withPayload("hello").build());
|
||||
try {
|
||||
@@ -137,79 +84,9 @@ public class ExtendedPropertiesBinderAwareChannelResolverTests extends BinderAwa
|
||||
assertThat(received).hasSize(1);
|
||||
assertThat(new String((byte[])received.get(0).getPayload())).isEqualTo("hello");
|
||||
context.close();
|
||||
assertThat(producerBindings).hasSize(1);
|
||||
assertThat(binding.isBound()).isFalse().describedAs("Must not be bound");
|
||||
}
|
||||
|
||||
/**
|
||||
* A simple test binder that creates queues for the destinations. Ignores groups.
|
||||
*/
|
||||
class TestBinder implements
|
||||
ExtendedPropertiesBinder<MessageChannel, ExtendedConsumerProperties, ExtendedProducerProperties> {
|
||||
|
||||
private final Map<String, DirectChannel> destinations = new ConcurrentHashMap<>();
|
||||
|
||||
@Override
|
||||
public Binding<MessageChannel> 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<MessageChannel> 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);
|
||||
TestBinding binding = new TestBinding(name, directHandler);
|
||||
producerBindings.add(binding);
|
||||
return binding;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ExtendedConsumerProperties getExtendedConsumerProperties(String channelName) {
|
||||
return new ExtendedConsumerProperties(new ConsumerProperties());
|
||||
}
|
||||
|
||||
@Override
|
||||
public ExtendedProducerProperties getExtendedProducerProperties(String channelName) {
|
||||
return new ExtendedProducerProperties(new ProducerProperties());
|
||||
}
|
||||
|
||||
private final class TestBinding implements Binding<MessageChannel> {
|
||||
|
||||
private final String name;
|
||||
|
||||
private final DirectHandler directHandler;
|
||||
|
||||
private boolean bound = true;
|
||||
|
||||
private TestBinding(String name, DirectHandler directHandler) {
|
||||
this.name = name;
|
||||
this.directHandler = directHandler;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbind() {
|
||||
bound = false;
|
||||
destinations.get(name).unsubscribe(directHandler);
|
||||
}
|
||||
|
||||
public boolean isBound() {
|
||||
return bound;
|
||||
}
|
||||
for (Bindable bindable : bindables.values()) {
|
||||
assertEquals(0, bindable.getInputs().size());
|
||||
assertEquals(0, bindable.getOutputs().size());//Must not be bound"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
/*
|
||||
* Copyright 2017 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.integration;
|
||||
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
|
||||
/**
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
abstract class AbstractDestination {
|
||||
|
||||
private SubscribableChannel channel;
|
||||
|
||||
SubscribableChannel getChannel() {
|
||||
return channel;
|
||||
}
|
||||
|
||||
void setChannel(SubscribableChannel channel) {
|
||||
this.channel = channel;
|
||||
this.afterChannelIsSet();
|
||||
}
|
||||
|
||||
void afterChannelIsSet() {
|
||||
// noop
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
/*
|
||||
* Copyright 2017 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.integration;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.annotation.StreamListener;
|
||||
import org.springframework.cloud.stream.messaging.Processor;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.handler.annotation.SendTo;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
/**
|
||||
* Sample spring cloud stream application that demonstrates the usage of {@link SpringIntegrationChannelBinder}.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableBinding(Processor.class)
|
||||
@Import(SpringIntegrationBinderConfiguration.class)
|
||||
public class SampleStreamApp {
|
||||
|
||||
public static void main(String[] args) {
|
||||
ApplicationContext context = new SpringApplicationBuilder(SampleStreamApp.class).web(WebApplicationType.NONE)
|
||||
.run("--server.port=0");
|
||||
SourceDestination source = context.getBean(SourceDestination.class);
|
||||
TargetDestination target = context.getBean(TargetDestination.class);
|
||||
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
|
||||
|
||||
Message<?> message = target.receive();
|
||||
assertEquals("Hello", new String((byte[])message.getPayload(), StandardCharsets.UTF_8));
|
||||
}
|
||||
|
||||
@StreamListener(Processor.INPUT)
|
||||
@SendTo(Processor.OUTPUT)
|
||||
public String receive(String value) {
|
||||
System.out.println("Handling payload: " + value);
|
||||
return value;
|
||||
}
|
||||
|
||||
@ServiceActivator(inputChannel="input.anonymous.errors")
|
||||
public void error(String value) {
|
||||
System.out.println("Handling ERROR payload: " + value);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
/*
|
||||
* Copyright 2017 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.integration;
|
||||
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
* Implementation of binder endpoint that represents the source destination
|
||||
* (e.g., destination from which messages will be received by Processor.INPUT).
|
||||
* <br>
|
||||
* You can interact with it by calling {@link #send(Message)} operation.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
public class SourceDestination extends AbstractDestination {
|
||||
|
||||
/**
|
||||
* Allows the {@link Message} to be sent to a Binder to be delegated
|
||||
* to binder's input destination (e.g., Processor.INPUT).
|
||||
*
|
||||
*/
|
||||
public void send(Message<?> message) {
|
||||
this.getChannel().send(message);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
/*
|
||||
* Copyright 2017 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.integration;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.cloud.stream.binder.BinderType;
|
||||
import org.springframework.cloud.stream.binder.BinderTypeRegistry;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.core.annotation.AnnotationUtils;
|
||||
import org.springframework.integration.config.EnableIntegration;
|
||||
|
||||
/**
|
||||
* {@link Binder} configuration backed by Spring Integration.
|
||||
*
|
||||
* Please see {@link SpringIntegrationChannelBinder} for more details.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
* @see SpringIntegrationChannelBinder
|
||||
*/
|
||||
@Configuration
|
||||
@ConditionalOnMissingBean(Binder.class)
|
||||
@EnableIntegration
|
||||
public class SpringIntegrationBinderConfiguration<T> {
|
||||
|
||||
public static final String NAME = "integration";
|
||||
|
||||
/**
|
||||
* Utility operation to return an array of configuration classes
|
||||
* defined in {@link EnableBinding} annotation.
|
||||
* Typically used for tests that do not rely on creating an SCSt boot
|
||||
* application annotated with {@link EnableBinding}, yet require
|
||||
* full {@link Binder} configuration.
|
||||
*/
|
||||
public static Class<?>[] getCompleteConfiguration() {
|
||||
List<Class<?>> configClasses = new ArrayList<>();
|
||||
configClasses.add(SpringIntegrationBinderConfiguration.class);
|
||||
Import annotation = AnnotationUtils.getAnnotation(EnableBinding.class, Import.class);
|
||||
Map<String, Object> annotationAttributes = AnnotationUtils.getAnnotationAttributes(annotation);
|
||||
configClasses.addAll(Arrays.asList((Class<?>[])annotationAttributes.get("value")));
|
||||
return configClasses.toArray(new Class<?>[] {});
|
||||
}
|
||||
|
||||
@Bean
|
||||
public BinderTypeRegistry binderTypeRegistry() {
|
||||
BinderType binderType = new BinderType(NAME, new Class[] {SpringIntegrationBinderConfiguration.class});
|
||||
BinderTypeRegistry btr = new DefaultBinderTypeRegistry(Collections.singletonMap(NAME, binderType));
|
||||
return btr;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public SourceDestination sourceDestination() {
|
||||
return new SourceDestination();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public TargetDestination targetDestination() {
|
||||
return new TargetDestination();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Bean
|
||||
public Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties> springIntegrationChannelBinder(SpringIntegrationProvisioner provisioner) {
|
||||
return (Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties>) new SpringIntegrationChannelBinder(provisioner);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public SpringIntegrationProvisioner springIntegrationProvisioner() {
|
||||
return new SpringIntegrationProvisioner();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,257 @@
|
||||
/*
|
||||
* Copyright 2015-2017 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.integration;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner.SpringIntegrationConsumerDestination;
|
||||
import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner.SpringIntegrationProducerDestination;
|
||||
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
|
||||
import org.springframework.cloud.stream.provisioning.ProducerDestination;
|
||||
import org.springframework.core.AttributeAccessor;
|
||||
import org.springframework.integration.core.MessageProducer;
|
||||
import org.springframework.integration.endpoint.MessageProducerSupport;
|
||||
import org.springframework.integration.handler.BridgeHandler;
|
||||
import org.springframework.integration.support.DefaultErrorMessageStrategy;
|
||||
import org.springframework.integration.support.ErrorMessageStrategy;
|
||||
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;
|
||||
import org.springframework.retry.RecoveryCallback;
|
||||
import org.springframework.retry.RetryCallback;
|
||||
import org.springframework.retry.RetryContext;
|
||||
import org.springframework.retry.RetryListener;
|
||||
import org.springframework.retry.support.RetryTemplate;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Implementation of {@link Binder} backed by Spring Integration framework.
|
||||
* It is useful for localized demos and testing.
|
||||
* <p>
|
||||
* This binder extends from the same base class ({@link AbstractMessageChannelBinder}) as
|
||||
* other binders (i.e., Rabbit, Kafka etc). Interaction with this binder is done via source
|
||||
* and target destination which emulate real binder's destinations (i.e., Kafka topic)
|
||||
* <br>
|
||||
* The destination classes are
|
||||
* <ul>
|
||||
* <li>{@link SourceDestination}</li>
|
||||
* <li>{@link TargetDestination}</li>
|
||||
* </ul>
|
||||
* Simply autowire them in your your application and send/receive messages.
|
||||
* </p>
|
||||
* You must also add {@link SpringIntegrationBinderConfiguration} to your configuration.
|
||||
* Below is the example using Spring Boot test.
|
||||
* <pre class="code">
|
||||
*
|
||||
* @RunWith(SpringJUnit4ClassRunner.class)
|
||||
* @SpringBootTest(classes = {SpringIntegrationBinderConfiguration.class, TestWithSIBinder.MyProcessor.class})
|
||||
* public class TestWithSIBinder {
|
||||
* @Autowired
|
||||
* private SourceDestination sourceDestination;
|
||||
*
|
||||
* @Autowired
|
||||
* private TargetDestination targetDestination;
|
||||
*
|
||||
* @Test
|
||||
* public void testWiring() {
|
||||
* sourceDestination.send(new GenericMessage<String>("Hello"));
|
||||
* assertEquals("Hello world", new String((byte[])targetDestination.receive().getPayload(), StandardCharsets.UTF_8));
|
||||
* }
|
||||
*
|
||||
* @SpringBootApplication
|
||||
* @EnableBinding(Processor.class)
|
||||
* public static class MyProcessor {
|
||||
* @StreamListener(Processor.INPUT)
|
||||
* @SendTo(Processor.OUTPUT)
|
||||
* public String transform(String in) {
|
||||
* return in + " world";
|
||||
* }
|
||||
* }
|
||||
* }
|
||||
* </pre>
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*/
|
||||
class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder<ConsumerProperties,
|
||||
ProducerProperties, SpringIntegrationProvisioner> {
|
||||
|
||||
@Autowired
|
||||
private BeanFactory beanFactory;
|
||||
|
||||
SpringIntegrationChannelBinder(SpringIntegrationProvisioner provisioningProvider) {
|
||||
super(new String[] {}, provisioningProvider);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
|
||||
ProducerProperties producerProperties, MessageChannel errorChannel) throws Exception {
|
||||
BridgeHandler handler = new BridgeHandler();
|
||||
handler.setBeanFactory(this.beanFactory);
|
||||
handler.setOutputChannel(((SpringIntegrationProducerDestination)destination).getChannel());
|
||||
return handler;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, ConsumerProperties properties)
|
||||
throws Exception {
|
||||
ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
|
||||
SubscribableChannel siBinderInputChannel = ((SpringIntegrationConsumerDestination)destination).getChannel();
|
||||
|
||||
IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer();
|
||||
IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(messageListenerContainer);
|
||||
|
||||
String groupName = StringUtils.hasText(group) ? group : "anonymous";
|
||||
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, groupName, properties);
|
||||
if (properties.getMaxAttempts() > 1) {
|
||||
adapter.setRetryTemplate(buildRetryTemplate(properties));
|
||||
}
|
||||
else {
|
||||
adapter.setErrorMessageStrategy(errorMessageStrategy);
|
||||
adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
|
||||
}
|
||||
|
||||
siBinderInputChannel.subscribe(messageListenerContainer);
|
||||
|
||||
return adapter;
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementation of simple message listener container modeled after AMQP SimpleMessageListenerContainer
|
||||
*/
|
||||
private static class IntegrationMessageListeningContainer implements MessageHandler {
|
||||
|
||||
private Consumer<Message<?>> listener;
|
||||
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
this.listener.accept(message);
|
||||
}
|
||||
|
||||
public void setMessageListener(Consumer<Message<?>> listener) {
|
||||
this.listener = listener;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementation of inbound channel adapter modeled after AmqpInboundChannelAdapter
|
||||
*/
|
||||
private static class IntegrationBinderInboundChannelAdapter extends MessageProducerSupport {
|
||||
|
||||
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
|
||||
|
||||
private final IntegrationMessageListeningContainer listenerContainer;
|
||||
|
||||
private RetryTemplate retryTemplate;
|
||||
|
||||
private RecoveryCallback<? extends Object> recoveryCallback;
|
||||
|
||||
IntegrationBinderInboundChannelAdapter(IntegrationMessageListeningContainer listenerContainer) {
|
||||
this.listenerContainer = listenerContainer;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
// Temporarily unused until DLQ strategy for this binder becomes a requirement
|
||||
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
|
||||
this.recoveryCallback = recoveryCallback;
|
||||
}
|
||||
|
||||
public void setRetryTemplate(RetryTemplate retryTemplate) {
|
||||
this.retryTemplate = retryTemplate;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onInit() {
|
||||
if (this.retryTemplate != null) {
|
||||
Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
|
||||
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
|
||||
+ "send an error message when retries are exhausted");
|
||||
}
|
||||
Listener messageListener = new Listener();
|
||||
if (this.retryTemplate != null) {
|
||||
this.retryTemplate.registerListener(messageListener);
|
||||
}
|
||||
this.listenerContainer.setMessageListener(messageListener);
|
||||
}
|
||||
|
||||
protected class Listener implements RetryListener, Consumer<Message<?>> {
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public void accept(Message<?> message) {
|
||||
try {
|
||||
if (IntegrationBinderInboundChannelAdapter.this.retryTemplate == null) {
|
||||
try {
|
||||
processMessage(message);
|
||||
}
|
||||
finally {
|
||||
attributesHolder.remove();
|
||||
}
|
||||
}
|
||||
else {
|
||||
IntegrationBinderInboundChannelAdapter.this.retryTemplate.execute(context -> {
|
||||
processMessage(message);
|
||||
return null;
|
||||
},
|
||||
(RecoveryCallback<Object>) IntegrationBinderInboundChannelAdapter.this.recoveryCallback);
|
||||
}
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
if (getErrorChannel() != null) {
|
||||
getMessagingTemplate().send(getErrorChannel(), buildErrorMessage(null,
|
||||
new IllegalStateException("Message conversion failed: " + message, e)));
|
||||
}
|
||||
else {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void processMessage(Message<?> message) {
|
||||
sendMessage(message);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
|
||||
if (IntegrationBinderInboundChannelAdapter.this.recoveryCallback != null) {
|
||||
attributesHolder.set(context);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
|
||||
Throwable throwable) {
|
||||
attributesHolder.remove();
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
|
||||
Throwable throwable) {
|
||||
// Empty
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
/*
|
||||
* Copyright 2017 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.integration;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
|
||||
import org.springframework.cloud.stream.provisioning.ProducerDestination;
|
||||
import org.springframework.cloud.stream.provisioning.ProvisioningException;
|
||||
import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
|
||||
import org.springframework.integration.channel.AbstractMessageChannel;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.PublishSubscribeChannel;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
|
||||
/**
|
||||
* {@link ProvisioningProvider} to support {@link SpringIntegrationChannelBinder}. It
|
||||
* exists primarily to support {@link AbstractMessageChannel} semantics for creating
|
||||
* {@link ConsumerDestination} and {@link ProducerDestination}, to interact with this
|
||||
* {@link Binder}.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
class SpringIntegrationProvisioner implements ProvisioningProvider<ConsumerProperties, ProducerProperties> {
|
||||
|
||||
private final Map<String, SubscribableChannel> provisionedDestinations = new HashMap<>();
|
||||
|
||||
@Autowired
|
||||
private SourceDestination source;
|
||||
|
||||
@Autowired
|
||||
private TargetDestination target;
|
||||
|
||||
/**
|
||||
* Will provision producer destination as an SI {@link PublishSubscribeChannel}.
|
||||
* <br>
|
||||
* This provides convenience of registering additional subscriber (handler in the test method)
|
||||
* along side of being able to call {@link TargetDestination#receive()} to get a
|
||||
* {@link Message} for additional assertions.
|
||||
*/
|
||||
@Override
|
||||
public ProducerDestination provisionProducerDestination(String name, ProducerProperties properties) throws ProvisioningException {
|
||||
SubscribableChannel destination = this.provisionDestination(name, true);
|
||||
this.target.setChannel(destination);
|
||||
return new SpringIntegrationProducerDestination(name, destination);
|
||||
}
|
||||
|
||||
/**
|
||||
* Will provision consumer destination as SI {@link DirectChannel}
|
||||
*/
|
||||
@Override
|
||||
public ConsumerDestination provisionConsumerDestination(String name, String group, ConsumerProperties properties) throws ProvisioningException {
|
||||
SubscribableChannel destination = this.provisionDestination(name, false);
|
||||
this.source.setChannel(destination);
|
||||
return new SpringIntegrationConsumerDestination(name, destination);
|
||||
}
|
||||
|
||||
private SubscribableChannel provisionDestination(String name, boolean pubSub) {
|
||||
String destinationName = name + ".destination";
|
||||
SubscribableChannel destination = this.provisionedDestinations.get(destinationName);
|
||||
if (destination == null) {
|
||||
destination = pubSub ? new PublishSubscribeChannel() : new DirectChannel();
|
||||
((AbstractMessageChannel)destination).setBeanName(destinationName);
|
||||
((AbstractMessageChannel)destination).setComponentName(destinationName);
|
||||
this.provisionedDestinations.put(destinationName, destination);
|
||||
}
|
||||
return destination;
|
||||
}
|
||||
|
||||
class SpringIntegrationConsumerDestination implements ConsumerDestination {
|
||||
private final String name;
|
||||
private final SubscribableChannel channel;
|
||||
|
||||
SpringIntegrationConsumerDestination(String name, SubscribableChannel channel) {
|
||||
this.name = name;
|
||||
this.channel = channel;
|
||||
}
|
||||
|
||||
public SubscribableChannel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getName() {
|
||||
return this.name;
|
||||
}
|
||||
}
|
||||
|
||||
class SpringIntegrationProducerDestination implements ProducerDestination {
|
||||
private final String name;
|
||||
private final SubscribableChannel channel;
|
||||
|
||||
SpringIntegrationProducerDestination(String name, SubscribableChannel channel) {
|
||||
this.name = name;
|
||||
this.channel = channel;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getNameForPartition(int partition) {
|
||||
return this.getName() + partition;
|
||||
}
|
||||
|
||||
public SubscribableChannel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getName() {
|
||||
return this.name;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
/*
|
||||
* Copyright 2017 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.integration;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedTransferQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
* Implementation of binder endpoint that represents the target destination
|
||||
* (e.g., destination which receives messages sent to Processor.OUTPUT)
|
||||
* <br>
|
||||
* You can interact with it by calling {@link #receive()} operation.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
public class TargetDestination extends AbstractDestination {
|
||||
|
||||
private BlockingQueue<Message<?>> messages;
|
||||
|
||||
/**
|
||||
* Allows to access {@link Message}s received by this {@link TargetDestination}.
|
||||
* @param timeout how long to wait before giving up
|
||||
*/
|
||||
public Message<?> receive(long timeout) {
|
||||
try {
|
||||
return this.messages.poll(timeout, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Allows to access {@link Message}s received by this {@link TargetDestination}.
|
||||
*/
|
||||
public Message<?> receive() {
|
||||
return this.receive(0);
|
||||
}
|
||||
|
||||
@Override
|
||||
void afterChannelIsSet() {
|
||||
this.messages = new LinkedTransferQueue<>();
|
||||
this.getChannel().subscribe(message -> messages.offer(message));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user