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

Binder API simplification

Browse Source 
Resolves #330

- removes `unbind` from the binder and moves it to the `Binding` instance itself;
- `Binding` is now an interface with a default implementation provided by SCS
- Removed all methods from Binder except for unbind()
- Removed old and unused code
- Removed circular reference between default Binder and Binding implementations
- Removed Binder type
This commit is contained in:
Marius Bogoevici
2016-02-09 17:53:23 -05:00
committed by Mark Fisher
parent 73b4b6fb3e
commit 6c880726b8
28 changed files with 455 additions and 900 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -36,13 +36,14 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.BinderException;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultBinding;
import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.MessageValues;
import org.springframework.http.MediaType;
import org.springframework.integration.channel.FixedSubscriberChannel;
@@ -127,7 +128,7 @@ import scala.collection.Seq;
* @author Mark Fisher
* @author Soby Chacko
*/
public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel> {
public static final ByteArraySerializer BYTE_ARRAY_SERIALIZER = new ByteArraySerializer();
@@ -489,10 +490,8 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
consumer.setBeanFactory(this.getBeanFactory());
consumer.setBeanName("outbound." + name);
consumer.afterPropertiesSet();
Binding<MessageChannel> producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer,
producerPropertiesAccessor);
addBinding(producerBinding);
producerBinding.start();
DefaultBinding<MessageChannel> producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer, producerPropertiesAccessor);
consumer.start();
return producerBinding;
}
catch (Exception e) {
@@ -644,9 +643,8 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
String groupedName = groupedName(name, group);
edc.setBeanName("inbound." + groupedName);
Binding<MessageChannel> consumerBinding = Binding.forConsumer(name, group, edc, moduleInputChannel, accessor);
addBinding(consumerBinding);
consumerBinding.start();
DefaultBinding<MessageChannel> consumerBinding = new DefaultBinding<>(name, group, moduleInputChannel, edc, accessor);
edc.start();
return consumerBinding;
}

36
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -174,8 +174,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertArrayEquals(ratherBigPayload, (byte[]) inbound.getPayload());
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
}
@@ -205,8 +205,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory().getPartitions(
"foo" + uniqueBindingId + ".0");
assertThat(partitions, hasSize(10));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
@@ -238,8 +238,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory().getPartitions(
"foo" + uniqueBindingId + ".0");
assertThat(partitions, hasSize(6));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
@@ -270,8 +270,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory().getPartitions(
"foo" + uniqueBindingId + ".0");
assertThat(partitions, hasSize(6));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
@@ -302,8 +302,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory().getPartitions(
"foo" + uniqueBindingId + ".0");
assertThat(partitions, hasSize(5));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
@@ -334,8 +334,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory().getPartitions(
"foo" + uniqueBindingId + ".0");
assertThat(partitions, hasSize(5));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
@@ -422,7 +422,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
binder.unbind(consumerBinding);
consumerBinding.unbind();
String testPayload3 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload3.getBytes()));
@@ -438,8 +438,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Message<byte[]> receivedMessage6 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage6, not(nullValue()));
assertThat(new String(receivedMessage6.getPayload()), equalTo(testPayload3));
binder.unbind(consumerBinding);
binder.unbind(producerBinding);
consumerBinding.unbind();
producerBinding.unbind();
}
@Test
@@ -468,7 +468,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
binder.unbind(consumerBinding);
consumerBinding.unbind();
String testPayload3 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload3.getBytes()));
@@ -478,7 +478,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Message<byte[]> receivedMessage3 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage3, not(nullValue()));
assertThat(new String(receivedMessage3.getPayload()), equalTo(testPayload3));
binder.unbind(consumerBinding);
binder.unbind(producerBinding);
consumerBinding.unbind();
producerBinding.unbind();
}
}

47
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java
View File

@@ -23,10 +23,8 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import org.junit.Ignore;
@@ -36,7 +34,6 @@ import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.TestUtils;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -72,10 +69,6 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, properties);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindings.size());
properties.clear();
properties.put("concurrency", "2");
properties.put("count","3");
@@ -109,10 +102,10 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
((byte[]) receive2.getPayload())[0]),
containsInAnyOrder((byte)0, (byte)1, (byte)2));
binder.unbind(input0Binding);
binder.unbind(input1Binding);
binder.unbind(input2Binding);
binder.unbind(outputBinding);
input0Binding.unbind();
input1Binding.unbind();
input2Binding.unbind();
outputBinding.unbind();
}
@Test
@@ -128,11 +121,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("part.0", output, properties);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindings.size());
try {
AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
AbstractEndpoint endpoint = extractEndpoint(outputBinding);
assertThat(getEndpointRouting(endpoint), containsString("part.0-' + headers['partition']"));
}
catch (UnsupportedOperationException ignored) {
@@ -179,10 +169,10 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
((byte[]) receive2.getPayload())[0]),
containsInAnyOrder((byte)0, (byte)1, (byte)2));
binder.unbind(input0Binding);
binder.unbind(input1Binding);
binder.unbind(input2Binding);
binder.unbind(outputBinding);
input0Binding.unbind();
input1Binding.unbind();
input2Binding.unbind();
outputBinding.unbind();
}
@Test
@@ -200,8 +190,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertEquals("foo", new String((byte[])inbound.getPayload()));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
// Ignored, since raw mode does not support headers
@@ -251,7 +241,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
assertEquals("foo", new String((byte[]) tapped2.getPayload()));
}
// delete one tap stream is deleted
binder.unbind(input3Binding);
input3Binding.unbind();
Message<?> message2 = MessageBuilder.withPayload("bar".getBytes()).build();
moduleOutputChannel.send(message2);
@@ -267,11 +257,14 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
assertNotNull(receive(module3InputChannel));
// clean up
binder.unbind(input1Binding);
binder.unbind(input2Binding);
binder.unbind(input3Binding);
binder.unbind(producerBinding);
assertTrue(getBindings(binder).isEmpty());
input1Binding.unbind();
input2Binding.unbind();
input3Binding.unbind();
producerBinding.unbind();
assertFalse(extractEndpoint(input1Binding).isRunning());
assertFalse(extractEndpoint(input2Binding).isRunning());
assertFalse(extractEndpoint(input3Binding).isRunning());
assertFalse(extractEndpoint(producerBinding).isRunning());
}
private void assertMessageReceive(QueueChannel moduleInputChannel, String payload) {

6
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java
View File

@@ -26,7 +26,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.binder.BindingCleaner;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
@@ -102,7 +102,7 @@ public class RabbitBindingCleaner implements BindingCleaner {
private List<String> findStreamQueues(String adminUri, String vhost, String binderPrefix, String stream,
RestTemplate restTemplate) {
String queueNamePrefix = adjustPrefix(MessageChannelBinderSupport.applyPrefix(binderPrefix, stream));
String queueNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, stream));
List<Map<String, Object>> queues = listAllQueues(adminUri, vhost, restTemplate);
List<String> removedQueues = new ArrayList<>();
for (Map<String, Object> queue : queues) {
@@ -147,7 +147,7 @@ public class RabbitBindingCleaner implements BindingCleaner {
.pathSegment("exchanges", "{vhost}")
.buildAndExpand(vhost).encode().toUri();
List<Map<String, Object>> exchanges = restTemplate.getForObject(uri, List.class);
String exchangeNamePrefix = adjustPrefix(MessageChannelBinderSupport.applyPrefix(binderPrefix, entity));
String exchangeNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, entity));
for (Map<String, Object> exchange : exchanges) {
String exchangeName = (String) exchange.get("name");
if (exchangeName.startsWith(exchangeNamePrefix)) {

47
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -59,13 +59,13 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultBinding;
import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.MessageValues;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.GenericApplicationContext;
@@ -108,7 +108,7 @@ import com.rabbitmq.client.Envelope;
* @author David Turanski
* @author Marius Bogoevici
*/
public class RabbitMessageChannelBinder extends MessageChannelBinderSupport implements DisposableBean {
public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel> {
public static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
= new AnonymousQueue.Base64UrlNamingStrategy("anonymous.");
@@ -455,9 +455,9 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
return args;
}
private Binding<MessageChannel> doRegisterConsumer(String name, String group, MessageChannel moduleInputChannel, Queue queue,
RabbitPropertiesAccessor properties) {
Binding<MessageChannel> consumerBinding = null;
private Binding<MessageChannel> doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue,
final RabbitPropertiesAccessor properties) {
DefaultBinding<MessageChannel> consumerBinding = null;
// Fix for XD-2503
// Temporarily overrides the thread context classloader with the one where the SimpleMessageListenerContainer
// is defined
@@ -511,14 +511,19 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
mapper.setReplyHeaderNames(properties.getReplyHeaderPattens(this.defaultReplyHeaderPatterns));
adapter.setHeaderMapper(mapper);
adapter.afterPropertiesSet();
consumerBinding = Binding.forConsumer(name, group, adapter, moduleInputChannel, properties);
addBinding(consumerBinding);
consumerBinding = new DefaultBinding<MessageChannel>(name, group, moduleInputChannel, adapter, properties) {
@Override
protected void afterUnbind() {
cleanAutoDeclareContext(properties.getPrefix(defaultPrefix), name);
}
};
ReceivingHandler convertingBridge = new ReceivingHandler();
convertingBridge.setOutputChannel(moduleInputChannel);
convertingBridge.setBeanName(name + ".convert.bridge");
convertingBridge.afterPropertiesSet();
bridgeToModuleChannel.subscribe(convertingBridge);
consumerBinding.start();
adapter.start();
}
finally {
Thread.currentThread().setContextClassLoader(originalClassloader);
@@ -635,9 +640,9 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
consumer.setBeanFactory(getBeanFactory());
consumer.setBeanName("outbound." + name);
consumer.afterPropertiesSet();
Binding<MessageChannel> producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer, properties);
addBinding(producerBinding);
producerBinding.start();
DefaultBinding<MessageChannel> producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer, properties);
consumer.start();
return producerBinding;
}
@@ -714,13 +719,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
return prefix + DEAD_LETTER_EXCHANGE;
}
@Override
protected void afterUnbind(Binding<MessageChannel> binding) {
if (Binding.Type.consumer.equals(binding.getType())) {
cleanAutoDeclareContext(binding);
}
}
private void addToAutoDeclareContext(String name, Object bean) {
synchronized (this.autoDeclareContext) {
if (!this.autoDeclareContext.containsBean(name)) {
@@ -729,11 +727,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
}
}
private void cleanAutoDeclareContext(Binding<MessageChannel> binding) {
Assert.isTrue(binding.getPropertiesAccessor() instanceof RabbitPropertiesAccessor,
"Binding was not created by this binder");
String prefix = ((RabbitPropertiesAccessor) binding.getPropertiesAccessor()).getPrefix(this.defaultPrefix);
String name = binding.getName();
public void cleanAutoDeclareContext(String prefix, String name) {
synchronized(this.autoDeclareContext) {
removeSingleton(applyPrefix(prefix,name) + ".binding");
removeSingleton(applyPrefix(prefix,name));
@@ -752,11 +746,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
}
}
@Override
public void destroy() {
stopBindings();
}
@Override
public void doManualAck(LinkedList<MessageHeaders> messageHeadersList) {
Iterator<MessageHeaders> iterator = messageHeadersList.iterator();

14
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java
View File

@@ -38,7 +38,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
@@ -69,8 +69,8 @@ public class RabbitBinderCleanerTests {
CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource();
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
for (int i = 0; i < 5; i++) {
String queue1Name = MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream1 + ".default." + i);
String queue2Name = MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream2 + ".default." + i);
String queue1Name = AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".default." + i);
String queue2Name = AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".default." + i);
if (firstQueue == null) {
firstQueue = queue1Name;
}
@@ -86,7 +86,7 @@ public class RabbitBinderCleanerTests {
template.put(uri, new AmqpQueue(false, true));
uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}")
.buildAndExpand("/", MessageChannelBinderSupport.constructDLQName(queue1Name)).encode().toUri();
.buildAndExpand("/", AbstractBinder.constructDLQName(queue1Name)).encode().toUri();
template.put(uri, new AmqpQueue(false, true));
TopicExchange exchange = new TopicExchange(queue1Name);
rabbitAdmin.declareExchange(exchange);
@@ -96,21 +96,21 @@ public class RabbitBinderCleanerTests {
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue2Name)).to(exchange).with(queue2Name));
}
final TopicExchange topic1 = new TopicExchange(
MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream1 + ".foo.bar"));
AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".foo.bar"));
rabbitAdmin.declareExchange(topic1);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(firstQueue)).to(topic1).with("#"));
String foreignQueue = UUID.randomUUID().toString();
rabbitAdmin.declareQueue(new Queue(foreignQueue));
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(foreignQueue)).to(topic1).with("#"));
final TopicExchange topic2 = new TopicExchange(
MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream2 + ".foo.bar"));
AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".foo.bar"));
rabbitAdmin.declareExchange(topic2);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(firstQueue)).to(topic2).with("#"));
new RabbitTemplate(connectionFactory).execute(new ChannelCallback<Void>() {
@Override
public Void doInRabbit(Channel channel) throws Exception {
String queueName = MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream1 + ".default." + 4);
String queueName = AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".default." + 4);
String consumerTag = channel.basicConsume(queueName, new DefaultConsumer(channel));
try {
waitForConsumerStateNot(queueName, 0);

89
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -122,8 +122,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
});
moduleOutputChannel.send(message);
assertTrue(latch.await(10, TimeUnit.SECONDS));
binder.unbind(consumerBinding);
binder.unbind(producerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
@@ -132,10 +132,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
Properties properties = new Properties();
properties.put("transacted", "true"); // test transacted with defaults; not allowed with ackmode NONE
Binding<MessageChannel> consumerBinding = binder.bindConsumer("props.0", null, new DirectChannel(), properties);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindings.size());
AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
AbstractEndpoint endpoint = extractEndpoint(consumerBinding);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertEquals(AcknowledgeMode.AUTO, container.getAcknowledgeMode());
@@ -152,8 +149,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
assertEquals(1000L, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval"));
assertEquals(10000L, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval"));
assertEquals(2.0, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier"));
binder.unbind(consumerBinding);
assertEquals(0, bindings.size());
consumerBinding.unbind();
assertFalse(endpoint.isRunning());
properties = new Properties();
properties.put("ackMode", "NONE");
@@ -171,16 +168,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put("partitionIndex", 0);
consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindingsNow = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindingsNow.size());
endpoint = bindingsNow.get(0).getEndpoint();
endpoint = extractEndpoint(consumerBinding);
container = verifyContainer(endpoint);
assertEquals("foo.props.0.test", container.getQueueNames()[0]);
binder.unbind(consumerBinding);
assertEquals(0, bindingsNow.size());
consumerBinding.unbind();
assertFalse(endpoint.isRunning());
}
@Test
@@ -188,17 +182,15 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
Binder<MessageChannel> binder = getBinder();
Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(), null);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindings.size());
AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
AbstractEndpoint endpoint = extractEndpoint(producerBinding);
MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode",
MessageDeliveryMode.class);
assertEquals(MessageDeliveryMode.PERSISTENT, mode);
List<?> requestHeaders = TestUtils.getPropertyValue(endpoint,
"handler.delegate.headerMapper.requestHeaderMatcher.strategies", List.class);
assertEquals(2, requestHeaders.size());
binder.unbind(producerBinding);
assertEquals(0, bindings.size());
producerBinding.unbind();
assertFalse(endpoint.isRunning());
Properties properties = new Properties();
properties.put("prefix", "foo.");
@@ -211,8 +203,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties);
assertEquals(1, bindings.size());
endpoint = bindings.get(0).getEndpoint();
endpoint = extractEndpoint(producerBinding);
assertEquals(
"'props.0-' + headers['partition']",
TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression",
@@ -222,8 +213,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
assertEquals(MessageDeliveryMode.NON_PERSISTENT, mode);
verifyFooRequestProducer(endpoint);
binder.unbind(producerBinding);
assertEquals(0, bindings.size());
producerBinding.unbind();
assertFalse(endpoint.isRunning());
}
@Test
@@ -264,7 +255,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
}
assertTrue(n < 100);
binder.unbind(consumerBinding);
consumerBinding.unbind();
assertNotNull(admin.getQueueProperties(TEST_PREFIX + "durabletest.0.tgroup.dlq"));
}
@@ -291,7 +282,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("nondurabletest.0", "tgroup", moduleInputChannel, properties);
binder.unbind(consumerBinding);
consumerBinding.unbind();
assertNull(admin.getQueueProperties(TEST_PREFIX + "nondurabletest.0.dlq"));
}
@@ -330,7 +321,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
}
assertTrue(n < 100);
binder.unbind(consumerBinding);
consumerBinding.unbind();
ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.autoDeclareContext",
ApplicationContext.class);
@@ -419,11 +410,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
assertNotNull(received);
assertEquals(0, received.getMessageProperties().getHeaders().get("partition"));
binder.unbind(input0Binding);
binder.unbind(input1Binding);
binder.unbind(defaultConsumerBinding1);
binder.unbind(defaultConsumerBinding2);
binder.unbind(outputBinding);
input0Binding.unbind();
input1Binding.unbind();
defaultConsumerBinding1.unbind();
defaultConsumerBinding2.unbind();
outputBinding.unbind();
}
@Test
@@ -505,11 +496,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
assertNotNull(received);
assertEquals(0, received.getMessageProperties().getHeaders().get("partition"));
binder.unbind(input0Binding);
binder.unbind(input1Binding);
binder.unbind(defaultConsumerBinding1);
binder.unbind(defaultConsumerBinding2);
binder.unbind(outputBinding);
input0Binding.unbind();
input1Binding.unbind();
defaultConsumerBinding1.unbind();
defaultConsumerBinding2.unbind();
outputBinding.unbind();
}
@Test
@@ -557,7 +548,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
}
assertTrue(n < 100);
binder.unbind(consumerBinding);
consumerBinding.unbind();
}
@SuppressWarnings("unchecked")
@@ -613,8 +604,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
assertEquals("bar", new String(in.getPayload()));
assertNull(in.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
/*
@@ -700,16 +691,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
assertNotNull(message);
assertEquals("1", message.getPayload());
binder.unbind(late0ProducerBinding);
binder.unbind(late0ConsumerBinding);
binder.unbind(partlate0ProducerBinding);
binder.unbind(partlate0Consumer0Binding);
binder.unbind(partlate0Consumer1Binding);
binder.unbind(noDlqProducerBinding);
binder.unbind(noDlqConsumerBinding);
binder.unbind(pubSubProducerBinding);
binder.unbind(nonDurableConsumerBinding);
binder.unbind(durableConsumerBinding);
late0ProducerBinding.unbind();
late0ConsumerBinding.unbind();
partlate0ProducerBinding.unbind();
partlate0Consumer0Binding.unbind();
partlate0Consumer1Binding.unbind();
noDlqProducerBinding.unbind();
noDlqConsumerBinding.unbind();
pubSubProducerBinding.unbind();
nonDurableConsumerBinding.unbind();
durableConsumerBinding.unbind();
binder.cleanup();

45
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java
View File

@@ -25,13 +25,13 @@ import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultBinding;
import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.MessageValues;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisOperations;
@@ -65,11 +65,11 @@ import org.springframework.util.StringUtils;
* @author David Turanski
* @author Jennifer Hickey
*/
public class RedisMessageChannelBinder extends MessageChannelBinderSupport implements DisposableBean {
public class RedisMessageChannelBinder extends AbstractBinder<MessageChannel> {
private static final String ERROR_HEADER = "errorKey";
private static final String CONSUMER_GROUPS_KEY_PREFIX = "groups.";
static final String CONSUMER_GROUPS_KEY_PREFIX = "groups.";
private static final SpelExpressionParser parser = new SpelExpressionParser();
@@ -170,7 +170,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
}
private Binding<MessageChannel> doRegisterConsumer(String bindingName, String group, String channelName, MessageChannel moduleInputChannel,
MessageProducerSupport adapter, RedisPropertiesAccessor properties) {
MessageProducerSupport adapter, final RedisPropertiesAccessor properties) {
DirectChannel bridgeToModuleChannel = new DirectChannel();
bridgeToModuleChannel.setBeanFactory(this.getBeanFactory());
bridgeToModuleChannel.setBeanName(channelName + ".bridge");
@@ -178,15 +178,21 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
adapter.setOutputChannel(bridgeInputChannel);
adapter.setBeanName("inbound." + channelName);
adapter.afterPropertiesSet();
Binding<MessageChannel> consumerBinding = Binding.forConsumer(channelName, group, adapter, moduleInputChannel, properties);
addBinding(consumerBinding);
DefaultBinding<MessageChannel> consumerBinding = new DefaultBinding<MessageChannel>(channelName, group, moduleInputChannel, adapter, properties) {
@Override
protected void afterUnbind() {
String key = RedisMessageChannelBinder.CONSUMER_GROUPS_KEY_PREFIX + getName();
RedisMessageChannelBinder.this.redisOperations.boundZSetOps(key).incrementScore(getGroup(), -1);
}
};
ReceivingHandler convertingBridge = new ReceivingHandler();
convertingBridge.setOutputChannel(moduleInputChannel);
convertingBridge.setBeanName(channelName + ".bridge.handler");
convertingBridge.afterPropertiesSet();
bridgeToModuleChannel.subscribe(convertingBridge);
this.redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + bindingName).incrementScore(group, 1);
consumerBinding.start();
adapter.start();
return consumerBinding;
}
@@ -247,17 +253,6 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
}
}
@Override
protected void afterUnbind(Binding<MessageChannel> binding) {
if (Binding.Type.consumer.equals(binding.getType())) {
String key = CONSUMER_GROUPS_KEY_PREFIX + binding.getName();
boolean durable = binding.getPropertiesAccessor().isDurable(defaultDurableSubscription);
if (!durable) {
this.redisOperations.boundZSetOps(key).incrementScore(binding.getGroup(), -1);
}
}
}
@Override
public Binding<MessageChannel> bindProducer(final String name, MessageChannel moduleOutputChannel, Properties properties) {
Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
@@ -290,17 +285,11 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
consumer.setBeanFactory(this.getBeanFactory());
consumer.setBeanName("outbound." + name);
consumer.afterPropertiesSet();
Binding<MessageChannel> producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer, properties);
addBinding(producerBinding);
producerBinding.start();
DefaultBinding<MessageChannel> producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer, properties);
consumer.start();
return producerBinding;
}
@Override
public void destroy() {
stopBindings();
}
private class SendingHandler extends AbstractMessageHandler {
private final String bindingName;

45
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.redis;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
@@ -91,15 +92,11 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
Properties properties = new Properties();
properties.put("maxAttempts", "1"); // disable retry
Binding<MessageChannel> binding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindings.size());
assertEquals(binding, bindings.get(0));
AbstractEndpoint endpoint = binding.getEndpoint();
AbstractEndpoint endpoint = extractEndpoint(binding);
assertThat(endpoint, instanceOf(RedisQueueMessageDrivenEndpoint.class));
assertSame(DirectChannel.class, TestUtils.getPropertyValue(endpoint, "outputChannel").getClass());
binder.unbind(binding);
assertEquals(0, bindings.size());
binding.unbind();
assertFalse(endpoint.isRunning());
properties.put("backOffInitialInterval", "2000");
properties.put("backOffMaxInterval", "20000");
@@ -109,13 +106,11 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
properties.put("partitionIndex", 0);
binding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties);
assertEquals(1, bindings.size());
assertEquals(binding, bindings.get(0));
endpoint = binding.getEndpoint();
endpoint = extractEndpoint(binding);
verifyConsumer(endpoint);
binder.unbind(binding);
assertEquals(0, bindings.size());
binding.unbind();
assertFalse(endpoint.isRunning());
}
@Test
@@ -123,19 +118,15 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
Binder<MessageChannel> binder = getBinder();
Binding<MessageChannel> consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), null);
Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(), null);
AbstractEndpoint producerEndpoint = extractEndpoint(producerBinding);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(2, bindings.size());
assertEquals(producerBinding, bindings.get(1));
AbstractEndpoint endpoint = producerBinding.getEndpoint();
@SuppressWarnings("unchecked")
Map<String, RedisQueueOutboundChannelAdapter> adapters = TestUtils.getPropertyValue(endpoint, "handler.adapters", Map.class);
Map<String, RedisQueueOutboundChannelAdapter> adapters = TestUtils.getPropertyValue(producerEndpoint, "handler.adapters", Map.class);
RedisQueueOutboundChannelAdapter adapter = adapters.get("test");
assertEquals(
"props.0.test",
TestUtils.getPropertyValue(adapter, "queueNameExpression", Expression.class).getExpressionString());
binder.unbind(producerBinding);
assertEquals(1, bindings.size());
producerBinding.unbind();
assertFalse(producerEndpoint.isRunning());
Properties properties = new Properties();
properties.put("partitionKeyExpression", "'foo'");
@@ -145,16 +136,16 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties);
assertEquals(2, bindings.size());
endpoint = bindings.get(1).getEndpoint();
adapter = (RedisQueueOutboundChannelAdapter) TestUtils.getPropertyValue(endpoint, "handler.adapters", Map.class).get("test");
producerEndpoint = extractEndpoint(producerBinding);
adapter = (RedisQueueOutboundChannelAdapter) TestUtils.getPropertyValue(producerEndpoint, "handler.adapters", Map.class).get("test");
assertEquals(
"'props.0.test-' + headers['partition']",
TestUtils.getPropertyValue(adapter, "queueNameExpression", Expression.class).getExpressionString());
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
assertEquals(0, bindings.size());
producerBinding.unbind();
consumerBinding.unbind();
assertFalse(producerEndpoint.isRunning());
assertFalse(extractEndpoint(consumerBinding).isRunning());
}
private void verifyConsumer(AbstractEndpoint endpoint) {
@@ -190,7 +181,7 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
Object rightPop = template.boundListOps("ERRORS:retry.0.test").rightPop(5, TimeUnit.SECONDS);
assertNotNull(rightPop);
assertThat(new String((byte[]) rightPop), containsString("foo"));
binder.unbind(consumerBinding);
consumerBinding.unbind();
}
@Test

52
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
View File

@@ -20,23 +20,22 @@ import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasProperty;
import static org.hamcrest.collection.IsArrayContainingInAnyOrder.arrayContainingInAnyOrder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import org.junit.After;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.http.MediaType;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -78,16 +77,16 @@ public abstract class AbstractBinderTests {
Binding<MessageChannel> foo1ProducerBinding = binder.bindProducer("foo.1", new DirectChannel(), null);
Binding<MessageChannel> foo1ConsumerBinding = binder.bindConsumer("foo.1", "test", new DirectChannel(), null);
Binding<MessageChannel> foo2ProducerBinding = binder.bindProducer("foo.2", new DirectChannel(), null);
Collection<?> bindings = getBindings(binder);
assertEquals(5, bindings.size());
binder.unbind(foo0ProducerBinding);
assertEquals(4, bindings.size());
binder.unbind(foo0ConsumerBinding);
binder.unbind(foo1ProducerBinding);
assertEquals(2, bindings.size());
binder.unbind(foo1ConsumerBinding);
binder.unbind(foo2ProducerBinding);
assertTrue(bindings.isEmpty());
foo0ProducerBinding.unbind();
assertFalse(TestUtils.getPropertyValue(foo0ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning());
foo0ConsumerBinding.unbind();
foo1ProducerBinding.unbind();
assertFalse(TestUtils.getPropertyValue(foo0ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning());
assertFalse(TestUtils.getPropertyValue(foo1ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning());
foo1ConsumerBinding.unbind();
foo2ProducerBinding.unbind();
assertFalse(TestUtils.getPropertyValue(foo1ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning());
assertFalse(TestUtils.getPropertyValue(foo2ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning());
}
@Test
@@ -107,8 +106,8 @@ public abstract class AbstractBinderTests {
assertEquals("foo", inbound.getPayload());
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertEquals("foo/bar", inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
@@ -148,11 +147,11 @@ public abstract class AbstractBinderTests {
hasProperty("payload", equalTo(testPayload2.getBytes()))));
binder.unbind(producerBinding1);
binder.unbind(consumerBinding1);
producerBinding1.unbind();
producerBinding2.unbind();
binder.unbind(producerBinding2);
binder.unbind(consumerBinding2);
consumerBinding1.unbind();
consumerBinding2.unbind();
}
@Test
@@ -171,21 +170,10 @@ public abstract class AbstractBinderTests {
assertEquals("foo", inbound.getPayload());
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertNull(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
producerBinding.unbind();
consumerBinding.unbind();
}
protected Collection<?> getBindings(Binder<MessageChannel> testBinder) {
if (testBinder instanceof AbstractTestBinder) {
return getBindingsFromBinder(((AbstractTestBinder<?>) testBinder).getCoreBinder());
}
return Collections.EMPTY_LIST;
}
protected Collection<?> getBindingsFromBinder(Binder<MessageChannel> binder) {
DirectFieldAccessor accessor = new DirectFieldAccessor(binder);
return (List<?>) accessor.getPropertyValue("bindings");
}
protected abstract Binder<MessageChannel> getBinder() throws Exception;

7
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java
View File

@@ -29,7 +29,7 @@ import org.springframework.messaging.MessageChannel;
* @author Gary Russell
* @author Mark Fisher
*/
public abstract class AbstractTestBinder<C extends MessageChannelBinderSupport> implements Binder<MessageChannel> {
public abstract class AbstractTestBinder<C extends AbstractBinder> implements Binder<MessageChannel> {
protected Set<String> queues = new HashSet<String>();
@@ -63,11 +63,6 @@ public abstract class AbstractTestBinder<C extends MessageChannelBinderSupport>
public abstract void cleanup();
@Override
public void unbind(Binding<MessageChannel> binding) {
binder.unbind(binding);
}
public C getBinder() {
return this.binder;
}

40
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java
View File

@@ -28,7 +28,6 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.UUID;
@@ -37,6 +36,7 @@ import org.hamcrest.Matcher;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -80,7 +80,7 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload1));
binder.unbind(binding2);
binding2.unbind();
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
@@ -100,9 +100,9 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload3));
binder.unbind(producerBinding);
binder.unbind(binding1);
binder.unbind(binding2);
producerBinding.unbind();
binding1.unbind();
binding2.unbind();
}
@Test
@@ -163,11 +163,8 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("part.0", output, producerProperties);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(4, bindings.size());
try {
AbstractEndpoint endpoint = bindings.get(3).getEndpoint();
AbstractEndpoint endpoint = extractEndpoint(outputBinding);
assertThat(getEndpointRouting(endpoint), containsString(
getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']"));
}
@@ -227,10 +224,10 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
containsOur3Messages);
}
binder.unbind(input0Binding);
binder.unbind(input1Binding);
binder.unbind(input2Binding);
binder.unbind(outputBinding);
input0Binding.unbind();
input1Binding.unbind();
input2Binding.unbind();
outputBinding.unbind();
}
@Test
@@ -261,11 +258,8 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, producerProperties);
@SuppressWarnings("unchecked")
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(4, bindings.size());
if (usesExplicitRouting()) {
AbstractEndpoint endpoint = bindings.get(3).getEndpoint();
AbstractEndpoint endpoint = extractEndpoint(outputBinding);
assertThat(getEndpointRouting(endpoint), containsString(
getExpectedRoutingBaseDestination("partJ.0", "test") + "-' + headers['partition']"));
}
@@ -295,10 +289,10 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
containsInAnyOrder(0, 1, 2));
}
binder.unbind(input0Binding);
binder.unbind(input1Binding);
binder.unbind(input2Binding);
binder.unbind(outputBinding);
input0Binding.unbind();
input1Binding.unbind();
input2Binding.unbind();
outputBinding.unbind();
}
/**
@@ -333,4 +327,8 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
protected abstract String getClassUnderTestName();
protected AbstractEndpoint extractEndpoint(Binding<MessageChannel> binding) {
DirectFieldAccessor accessor = new DirectFieldAccessor(binding);
return (AbstractEndpoint) accessor.getPropertyValue("endpoint");
}
}

9
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java
View File

@@ -25,12 +25,10 @@ import java.util.Collections;
import java.util.List;
import java.util.Properties;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Registration;
import org.junit.Before;
import org.junit.Test;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport.JavaClassMimeTypeConversion;
import org.springframework.cloud.stream.binder.AbstractBinder.JavaClassMimeTypeConversion;
import org.springframework.cloud.stream.tuple.DefaultTuple;
import org.springframework.cloud.stream.tuple.Tuple;
import org.springframework.cloud.stream.tuple.TupleBuilder;
@@ -46,6 +44,9 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Registration;
/**
* @author Gary Russell
* @author David Turanski
@@ -265,7 +266,7 @@ public class MessageChannelBinderSupportTests {
}
public class TestMessageChannelBinder extends MessageChannelBinderSupport {
public class TestMessageChannelBinder extends AbstractBinder<MessageChannel> {
@Override
protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel channel, Properties properties) {

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

@@ -53,7 +53,7 @@ public class TestSupportBinder implements Binder<MessageChannel> {
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel inboundBindTarget, Properties properties) {
return null;
return new TestBinding(name, inboundBindTarget, messageCollector);
}
/**
@@ -69,14 +69,7 @@ public class TestSupportBinder implements Binder<MessageChannel> {
}
});
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());
}
return new TestBinding(name, outboundBindTarget, messageCollector);
}
public MessageCollector messageCollector() {
@@ -114,4 +107,26 @@ public class TestSupportBinder implements Binder<MessageChannel> {
return queue;
}
}
/**
* @author Marius Bogoevici
*/
public static class TestBinding implements Binding<MessageChannel> {
private final MessageChannel target;
private final MessageCollectorImpl messageCollector;
private String name;
public TestBinding(String name, MessageChannel target, MessageCollectorImpl messageCollector) {
this.name = name;
this.target = target;
this.messageCollector = messageCollector;
}
@Override
public void unbind() {
messageCollector.unregister(target);
}
}
}

196
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java
View File

@@ -16,7 +16,6 @@
package org.springframework.cloud.stream.binder;
import static org.springframework.util.MimeTypeUtils.ALL;
import static org.springframework.util.MimeTypeUtils.APPLICATION_OCTET_STREAM;
import static org.springframework.util.MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE;
import static org.springframework.util.MimeTypeUtils.TEXT_PLAIN;
@@ -25,12 +24,9 @@ import static org.springframework.util.MimeTypeUtils.TEXT_PLAIN_VALUE;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
@@ -42,25 +38,18 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
@@ -76,9 +65,9 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
* @author Ilayaperumal Gopinathan
* @author Mark Fisher
* @author Marius Bogoevici
*/
public abstract class MessageChannelBinderSupport
implements Binder<MessageChannel>, ApplicationContextAware, InitializingBean {
public abstract class AbstractBinder<T> implements ApplicationContextAware, InitializingBean, Binder<T> {
protected static final String PARTITION_HEADER = "partition";
@@ -95,8 +84,6 @@ public abstract class MessageChannelBinderSupport
private final StringConvertingContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver();
protected static final List<MimeType> MEDIATYPES_MEDIATYPE_ALL = Collections.singletonList(ALL);
private static final int DEFAULT_BACKOFF_INITIAL_INTERVAL = 1000;
private static final int DEFAULT_BACKOFF_MAX_INTERVAL = 10000;
@@ -156,27 +143,12 @@ public abstract class MessageChannelBinderSupport
BinderPropertyKeys.BATCH_BUFFER_LIMIT,
}));
private final List<Binding<MessageChannel>> bindings = Collections.synchronizedList(new ArrayList<Binding<MessageChannel>>());
private final IdGenerator idGenerator = new AlternativeJdkIdGenerator();
protected volatile EvaluationContext evaluationContext;
private volatile PartitionSelectorStrategy partitionSelector = new DefaultPartitionSelector();
/**
* Used in the canonical case, when the binding does not involve an alias name.
*/
protected final SharedChannelProvider<DirectChannel> directChannelProvider = new
SharedChannelProvider<DirectChannel>(
DirectChannel.class) {
@Override
protected DirectChannel createSharedChannel(String name) {
return new DirectChannel();
}
};
protected volatile long defaultBackOffInitialInterval = DEFAULT_BACKOFF_INITIAL_INTERVAL;
protected volatile long defaultBackOffMaxInterval = DEFAULT_BACKOFF_MAX_INTERVAL;
@@ -209,7 +181,7 @@ public abstract class MessageChannelBinderSupport
/**
* For binder implementations that support a prefix, apply the prefix to the name.
* @param prefix the prefix.
* @param name the name.
* @param name the name.
*/
public static String applyPrefix(String prefix, String name) {
return prefix + name;
@@ -277,7 +249,8 @@ public abstract class MessageChannelBinderSupport
}
/**
* Set the default retry back off max interval for this binder; can be overridden with consumer 'backOffMaxInterval'
* Set the default retry back off max interval for this binder; can be overridden with consumer
* 'backOffMaxInterval'
* property.
* @param defaultBackOffMaxInterval
*/
@@ -362,7 +335,7 @@ public abstract class MessageChannelBinderSupport
}
@Override
public final Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel inputChannel, Properties properties) {
public final Binding<T> bindConsumer(String name, String group, T target, Properties properties) {
DefaultBindingPropertiesAccessor accessor = new DefaultBindingPropertiesAccessor(properties);
if (StringUtils.isEmpty(group)) {
Assert.isTrue(!accessor.getProperty(BinderPropertyKeys.DURABLE, defaultDurableSubscription),
@@ -370,75 +343,14 @@ public abstract class MessageChannelBinderSupport
Assert.isTrue(accessor.getPartitionIndex() < 0,
"A consumer group is required for a partitioned subscription");
}
return doBindConsumer(name, group, inputChannel, properties);
return doBindConsumer(name, group, target, properties);
}
protected abstract Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel, Properties properties);
/**
* Create a producer for the named channel and bind it to the binder. Synchronized to avoid creating multiple
* instances.
* @param name The name.
* @param channelName The name of the channel to be created, and registered as bean.
* @param properties The properties.
* @return The channel.
*/
protected synchronized MessageChannel doBindDynamicProducer(String name, String channelName,
Properties properties) {
MessageChannel channel = this.directChannelProvider.lookupSharedChannel(channelName);
if (channel == null) {
try {
channel = this.directChannelProvider.createAndRegisterChannel(channelName);
bindProducer(name, channel, properties);
}
catch (RuntimeException e) {
destroyCreatedChannel(channelName, channel);
throw new BinderException(
"Failed to bind dynamic channel '" + name + "' with properties " + properties, e);
}
}
return channel;
}
private void destroyCreatedChannel(String name, MessageChannel channel) {
BeanFactory beanFactory = this.applicationContext.getBeanFactory();
if (beanFactory.containsBean(name)) {
if (beanFactory instanceof DefaultListableBeanFactory) {
((DefaultListableBeanFactory) beanFactory).destroySingleton(name);
}
}
}
@Override
public void unbind(Binding<MessageChannel> binding) {
binding.stop();
this.bindings.remove(binding);
afterUnbind(binding);
}
protected void afterUnbind(Binding<MessageChannel> binding) {
}
protected void addBinding(Binding<MessageChannel> binding) {
this.bindings.add(binding);
}
protected void stopBindings() {
for (Lifecycle bean : this.bindings) {
try {
bean.stop();
}
catch (Exception e) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("failed to stop adapter", e);
}
}
}
}
protected abstract Binding<T> doBindConsumer(String name, String group, T inputTarget, Properties properties);
/**
* Construct a name comprised of the name and group.
* @param name the name.
* @param name the name.
* @param group the group.
* @return the constructed name.
*/
@@ -552,10 +464,11 @@ public abstract class MessageChannelBinderSupport
* partition selector class is provided, it will be invoked to determine the partition. Otherwise, if the partition
* expression is not null, it is evaluated against the key and is expected to return an integer to which the modulo
* function will be applied, using the partitionCount as the divisor. If no partition expression is provided, the
* key will be passed to the binder partition strategy along with the partitionCount. The default partition strategy
* key will be passed to the binder partition strategy along with the partitionCount. The default partition
* strategy
* uses {@code key.hashCode()}, and the result will be the mod of that value.
* @param message the message.
* @param meta the partitioning metadata.
* @param meta the partitioning metadata.
* @return the partition.
*/
protected int determinePartition(Message<?> message, PartitioningMetadata meta) {
@@ -638,12 +551,14 @@ public abstract class MessageChannelBinderSupport
}
/**
* Validate the provided deployment properties for the consumer against those supported by this binder implementation.
* The consumer is that part of the binder that consumes messages from the underlying infrastructure and sends them to
* Validate the provided deployment properties for the consumer against those supported by this binder
* implementation.
* The consumer is that part of the binder that consumes messages from the underlying infrastructure and sends them
* to
* the next module. Consumer properties are used to configure the consumer.
* @param name The name.
* @param name The name.
* @param properties The properties.
* @param supported The supported properties.
* @param supported The supported properties.
*/
protected void validateConsumerProperties(String name, Properties properties, Set<Object> supported) {
if (properties != null) {
@@ -652,12 +567,14 @@ public abstract class MessageChannelBinderSupport
}
/**
* Validate the provided deployment properties for the producer against those supported by this binder implementation.
* When a module sends a message to the binder, the producer uses these properties while sending it to the underlying
* Validate the provided deployment properties for the producer against those supported by this binder
* implementation.
* When a module sends a message to the binder, the producer uses these properties while sending it to the
* underlying
* infrastructure.
* @param name The name.
* @param name The name.
* @param properties The properties.
* @param supported The supported properties.
* @param supported The supported properties.
*/
protected void validateProducerProperties(String name, Properties properties, Set<Object> supported) {
if (properties != null) {
@@ -759,56 +676,6 @@ public abstract class MessageChannelBinderSupport
}
/**
* Looks up or optionally creates a new channel to use.
* @author Eric Bottard
*/
protected abstract class SharedChannelProvider<T extends MessageChannel> {
private final Class<T> requiredType;
protected SharedChannelProvider(Class<T> clazz) {
this.requiredType = clazz;
}
public synchronized final T lookupOrCreateSharedChannel(String name) {
T channel = lookupSharedChannel(name);
if (channel == null) {
channel = createAndRegisterChannel(name);
}
return channel;
}
@SuppressWarnings("unchecked")
public T createAndRegisterChannel(String name) {
T channel = createSharedChannel(name);
ConfigurableListableBeanFactory beanFactory = MessageChannelBinderSupport.this.applicationContext.getBeanFactory();
beanFactory.registerSingleton(name, channel);
channel = (T) beanFactory.initializeBean(channel, name);
if (MessageChannelBinderSupport.this.logger.isDebugEnabled()) {
MessageChannelBinderSupport.this.logger.debug("Registered channel:" + name);
}
return channel;
}
protected abstract T createSharedChannel(String name);
public T lookupSharedChannel(String name) {
T channel = null;
if (MessageChannelBinderSupport.this.applicationContext.containsBean(name)) {
try {
channel = MessageChannelBinderSupport.this.applicationContext.getBean(name, this.requiredType);
}
catch (Exception e) {
throw new IllegalArgumentException("bean '" + name
+ "' is already registered but does not match the required type");
}
}
return channel;
}
}
/**
* Handles representing any java class as a {@link MimeType}.
* @author David Turanski
@@ -888,21 +755,6 @@ public abstract class MessageChannelBinderSupport
}
public static class DirectHandler implements MessageHandler {
private final MessageChannel outputChannel;
public DirectHandler(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
this.outputChannel.send(message);
}
}
/**
* Perform manual acknowledgement based on the metadata stored in the binder.
*/

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java
View File

@@ -22,6 +22,7 @@ import java.util.Properties;
* A strategy interface used to bind an app interface to a logical name. The name is intended to identify a
* logical consumer or producer of messages. This may be a queue, a channel adapter, another message channel, a Spring
* bean, etc.
*
* @author Mark Fisher
* @author David Turanski
* @author Gary Russell
@@ -50,10 +51,4 @@ public interface Binder<T> {
*/
Binding<T> bindProducer(String name, T outboundBindTarget, Properties properties);
/**
* Unbind the target component represented by the provided Binding and stop any active components.
* @param binding the Binding instance to unbind
*/
void unbind(Binding<T> binding);
}

150
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2016 the original author or authors.
* 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.
@@ -16,10 +16,6 @@
package org.springframework.cloud.stream.binder;
import org.springframework.context.Lifecycle;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.util.Assert;
/**
* Represents a binding between an input or output and an adapter endpoint that connects via a Binder. The binding
* could be for a consumer or a producer. A consumer binding represents a connection from an adapter to an
@@ -31,141 +27,11 @@ import org.springframework.util.Assert;
* @author Marius Bogoevici
* @see org.springframework.cloud.stream.annotation.EnableBinding
*/
public class Binding<T> implements Lifecycle {
public enum Type {
producer, consumer
}
private final String name;
private final String group;
private final T target;
private final AbstractEndpoint endpoint;
private final Type type;
private final DefaultBindingPropertiesAccessor properties;
private Binding(String name, String group, T target, AbstractEndpoint endpoint, Type type,
DefaultBindingPropertiesAccessor properties) {
Assert.notNull(target, "target must not be null");
Assert.notNull(endpoint, "endpoint must not be null");
this.name = name;
this.group = group;
this.target = target;
this.endpoint = endpoint;
this.type = type;
this.properties = properties;
}
public static <T> Binding<T> forConsumer(String name, String group, AbstractEndpoint adapterFromBinder, T inputTarget,
DefaultBindingPropertiesAccessor properties) {
return new Binding<>(name, group, inputTarget, adapterFromBinder, Type.consumer, properties);
}
public static <T> Binding<T> forProducer(String name, T outputTarget, AbstractEndpoint adapterToBinder,
DefaultBindingPropertiesAccessor properties) {
return new Binding<>(name, null, outputTarget, adapterToBinder, Type.producer, properties);
}
public String getName() {
return name;
}
public String getGroup() {
return group;
}
public T getTarget() {
return target;
}
public AbstractEndpoint getEndpoint() {
return endpoint;
}
public Type getType() {
return type;
}
public DefaultBindingPropertiesAccessor getPropertiesAccessor() {
return properties;
}
@Override
public void start() {
endpoint.start();
}
@Override
public void stop() {
endpoint.stop();
}
@Override
public boolean isRunning() {
return endpoint.isRunning();
}
@Override
public String toString() {
return type + " Binding [name=" + name + ", target=" + target + ", endpoint=" + endpoint.getComponentName()
+ "]";
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + ((endpoint == null) ? 0 : endpoint.hashCode());
result = prime * result + ((group == null) ? 0 : group.hashCode());
result = prime * result + ((name == null) ? 0 : name.hashCode());
result = prime * result + ((properties == null) ? 0 : properties.hashCode());
result = prime * result + ((target == null) ? 0 : target.hashCode());
result = prime * result + ((type == null) ? 0 : type.hashCode());
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
Binding<?> other = (Binding<?>) obj;
if (endpoint == null) {
if (other.endpoint != null)
return false;
} else if (!endpoint.equals(other.endpoint))
return false;
if (group == null) {
if (other.group != null)
return false;
} else if (!group.equals(other.group))
return false;
if (name == null) {
if (other.name != null)
return false;
} else if (!name.equals(other.name))
return false;
if (properties == null) {
if (other.properties != null)
return false;
} else if (!properties.equals(other.properties))
return false;
if (target == null) {
if (other.target != null)
return false;
} else if (!target.equals(other.target))
return false;
if (type != other.type)
return false;
return true;
}
public interface Binding<T> {
/**
* Unbinds the target component represented by this instance and stops any active components. Implementations must
* be idempotent. After this method is invoked, the target is not expected to receive any message, this instance
* should be discarded, and a new Binding should be created instead.
*/
void unbind();
}

83
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java Normal file
View File

@@ -0,0 +1,83 @@
/*
* Copyright 2013-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.binder;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.util.Assert;
/**
* Default implementation for a {@link Binding}.
*
* @author Jennifer Hickey
* @author Mark Fisher
* @author Gary Russell
* @author Marius Bogoevici
* @see org.springframework.cloud.stream.annotation.EnableBinding
*/
public class DefaultBinding<T> implements Binding<T> {
private final String name;
private final String group;
private final T target;
private final AbstractEndpoint endpoint;
private final DefaultBindingPropertiesAccessor properties;
public DefaultBinding(String name, String group, T target, AbstractEndpoint endpoint,
DefaultBindingPropertiesAccessor properties) {
Assert.notNull(target, "target must not be null");
Assert.notNull(endpoint, "endpoint must not be null");
this.name = name;
this.group = group;
this.target = target;
this.endpoint = endpoint;
this.properties = properties;
}
public String getName() {
return name;
}
public String getGroup() {
return group;
}
@Override
public final void unbind() {
endpoint.stop();
afterUnbind();
}
protected void afterUnbind() {
}
public DefaultBindingPropertiesAccessor getPropertiesAccessor() {
return properties;
}
@Override
public String toString() {
return " Binding [name=" + name + ", target=" + target + ", endpoint=" + endpoint.getComponentName()
+ "]";
}
}

40
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DirectHandler.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* 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.binder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
/**
* @author Marius Bogoevici
*/
public class DirectHandler implements MessageHandler {
private final MessageChannel outputChannel;
public DirectHandler(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
this.outputChannel.send(message);
}
}

14
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java
View File

@@ -39,7 +39,7 @@ import org.springframework.cloud.stream.aggregate.SharedChannelRegistry;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.DirectHandler;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -180,7 +180,7 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
// handle the special case where the shared channel is of a different nature
// (i.e. pollable vs subscribable) than the target channel
if (isPollable(sharedChannel.getClass())) {
bridgePollableToSubscribableChannel(sharedChannel, new DirectChannel());
bridgePollableToSubscribableChannel((PollableChannel) sharedChannel, new DirectChannel());
}
else {
bridgeSubscribableToPollableChannel((SubscribableChannel) sharedChannel, new QueueChannel());
@@ -191,19 +191,19 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
return PollableChannel.class.equals(channelType);
}
private void bridgeSubscribableToPollableChannel(SubscribableChannel sharedChannel, MessageChannel inputChannel) {
sharedChannel.subscribe(new MessageChannelBinderSupport.DirectHandler(inputChannel));
private void bridgeSubscribableToPollableChannel(SubscribableChannel sharedChannel, PollableChannel pollableChannel) {
sharedChannel.subscribe(new DirectHandler(pollableChannel));
}
private void bridgePollableToSubscribableChannel(MessageChannel pollableChannel,
MessageChannel subscribableChannel) {
private void bridgePollableToSubscribableChannel(PollableChannel pollableChannel,
SubscribableChannel subscribableChannel) {
ConsumerEndpointFactoryBean consumerEndpointFactoryBean = new ConsumerEndpointFactoryBean();
consumerEndpointFactoryBean.setInputChannel(pollableChannel);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(this.pollableBridgeDefaultFrequency));
consumerEndpointFactoryBean.setPollerMetadata(pollerMetadata);
consumerEndpointFactoryBean
.setHandler(new MessageChannelBinderSupport.DirectHandler(
.setHandler(new DirectHandler(
subscribableChannel));
consumerEndpointFactoryBean.setBeanFactory(this.beanFactory);
try {

6
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java
View File

@@ -91,11 +91,10 @@ public class ChannelBindingService {
}
public void unbindConsumers(String inputChannelName) {
Binder<MessageChannel> binder = getBinderForChannel(inputChannelName);
List<Binding<MessageChannel>> bindings = this.consumerBindings.remove(inputChannelName);
if (bindings != null && !CollectionUtils.isEmpty(bindings)) {
for (Binding<MessageChannel> binding : bindings) {
binder.unbind(binding);
binding.unbind();
}
}
else if (log.isWarnEnabled()) {
@@ -104,10 +103,9 @@ public class ChannelBindingService {
}
public void unbindProducers(String outputChannelName) {
Binder<MessageChannel> binder = getBinderForChannel(outputChannelName);
Binding<MessageChannel> binding = this.producerBindings.remove(outputChannelName);
if (binding != null) {
binder.unbind(binding);
binding.unbind();
}
else if (log.isWarnEnabled()) {
log.warn("Trying to unbind channel '" + outputChannelName + "', but no binding found.");

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

@@ -32,6 +32,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -41,13 +42,11 @@ import org.mockito.Mockito;
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;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.utils.IntegrationUtils;
@@ -55,8 +54,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.messaging.SubscribableChannel;
/**
* @author Mark Fisher
@@ -68,13 +66,11 @@ public class BinderAwareChannelResolverTests {
private volatile BinderAwareChannelResolver resolver;
private volatile LocalMessageChannelBinder binder;
private volatile Binder<MessageChannel> binder;
@Before
public void setupContext() throws Exception {
this.binder = new LocalMessageChannelBinder();
this.binder.setApplicationContext(context);
this.binder.afterPropertiesSet();
this.binder = new TestBinder();
this.resolver = new BinderAwareChannelResolver(new BinderFactory<MessageChannel>() {
@Override
public Binder<MessageChannel> getBinder(String configurationName) {
@@ -85,14 +81,9 @@ public class BinderAwareChannelResolverTests {
context.getBeanFactory().registerSingleton("channelResolver",
this.resolver);
context.registerSingleton("other", DirectChannel.class);
context.registerSingleton("taskScheduler", ThreadPoolTaskScheduler.class);
context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
DefaultMessageBuilderFactory.class);
context.refresh();
PollerMetadata poller = new PollerMetadata();
poller.setTrigger(new PeriodicTrigger(1000));
binder.setPoller(poller);
}
@Test
@@ -156,4 +147,58 @@ public class BinderAwareChannelResolverTests {
assertSame(resolved, beanFactory.getBean("someTransport:foo"));
}
/**
* A simple test binder that creates queues for the destinations. Ignores groups.
*/
private class TestBinder implements Binder<MessageChannel> {
private final Map<String, DirectChannel> destinations = new ConcurrentHashMap<>();
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel inboundBindTarget,
Properties 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(inboundBindTarget, name, directHandler);
}
@Override
public Binding<MessageChannel> bindProducer(String name, MessageChannel outboundBindTarget, Properties 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);
return new TestBinding(outboundBindTarget, name, directHandler);
}
private class TestBinding implements Binding<MessageChannel> {
private final MessageChannel target;
private final String name;
private final DirectHandler directHandler;
public TestBinding(MessageChannel outboundChannel, String name, DirectHandler directHandler) {
this.target = outboundChannel;
this.name = name;
this.directHandler = directHandler;
}
@Override
public void unbind() {
destinations.get(name).unsubscribe(directHandler);
}
}
}
}

259
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java
View File

@@ -1,259 +0,0 @@
/*
* Copyright 2013-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.binder.local;
import java.util.Collection;
import java.util.Properties;
import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
/**
* A simple implementation of {@link org.springframework.cloud.stream.binder.Binder} for in-process use. For inbound and outbound, creates a
* {@link DirectChannel} or a {@link QueueChannel} depending on whether the binding is aliased or not then bridges the
* passed {@link MessageChannel} to the channel which is registered in the given application context. If that channel
* does not yet exist, it will be created.
*
* @author David Turanski
* @author Mark Fisher
* @author Gary Russell
* @author Jennifer Hickey
* @author Ilayaperumal Gopinathan
* @since 1.0
*/
public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
public static final String THREAD_NAME_PREFIX = "binder.local-";
private static final int DEFAULT_EXECUTOR_CORE_POOL_SIZE = 0;
private static final int DEFAULT_EXECUTOR_MAX_POOL_SIZE = 200;
private static final int DEFAULT_EXECUTOR_QUEUE_SIZE = Integer.MAX_VALUE;
private static final int DEFAULT_EXECUTOR_KEEPALIVE_SECONDS = 60;
private volatile PollerMetadata poller;
private final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
private volatile int executorCorePoolSize = DEFAULT_EXECUTOR_CORE_POOL_SIZE;
private volatile int executorMaxPoolSize = DEFAULT_EXECUTOR_MAX_POOL_SIZE;
private volatile int executorQueueSize = DEFAULT_EXECUTOR_QUEUE_SIZE;
private volatile int executorKeepAliveSeconds = DEFAULT_EXECUTOR_KEEPALIVE_SECONDS;
private final SharedChannelProvider<PublishSubscribeChannel> pubsubChannelProvider = new SharedChannelProvider<PublishSubscribeChannel>(
PublishSubscribeChannel.class) {
@Override
protected PublishSubscribeChannel createSharedChannel(String name) {
PublishSubscribeChannel publishSubscribeChannel = new PublishSubscribeChannel(executor);
publishSubscribeChannel.setIgnoreFailures(true);
return publishSubscribeChannel;
}
};
/**
* Set the poller to use when QueueChannels are used.
*/
public void setPoller(PollerMetadata poller) {
this.poller = poller;
}
/**
* Set the {@link ThreadPoolTaskExecutor}} core pool size to limit the number of concurrent
* threads. The executor is used for PubSub operations.
* Default: 0 (threads created on demand until maxPoolSize).
* @param executorCorePoolSize the pool size.
*/
public void setExecutorCorePoolSize(int executorCorePoolSize) {
this.executorCorePoolSize = executorCorePoolSize;
}
/**
* Set the {@link ThreadPoolTaskExecutor}} max pool size to limit the number of concurrent
* threads. The executor is used for PubSub operations.
* Default: 200.
* @param executorMaxPoolSize the pool size.
*/
public void setExecutorMaxPoolSize(int executorMaxPoolSize) {
this.executorMaxPoolSize = executorMaxPoolSize;
}
/**
* Set the {@link ThreadPoolTaskExecutor}} queue size to limit the number of concurrent
* threads. The executor is used for PubSub operations.
* Default: {@link Integer#MAX_VALUE}.
* @param executorQueueSize the queue size.
*/
public void setExecutorQueueSize(int executorQueueSize) {
this.executorQueueSize = executorQueueSize;
}
/**
* Set the {@link ThreadPoolTaskExecutor}} keep alive seconds.
* The executor is used for PubSub operations.
* @param executorKeepAliveSeconds the keep alive seconds.
*/
public void setExecutorKeepAliveSeconds(int executorKeepAliveSeconds) {
this.executorKeepAliveSeconds = executorKeepAliveSeconds;
}
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
this.executor.setCorePoolSize(this.executorCorePoolSize);
this.executor.setMaxPoolSize(this.executorMaxPoolSize);
this.executor.setQueueCapacity(this.executorQueueSize);
this.executor.setKeepAliveSeconds(this.executorKeepAliveSeconds);
this.executor.setThreadNamePrefix(THREAD_NAME_PREFIX);
this.executor.initialize();
}
@Override
protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel moduleInputChannel,
Properties properties) {
validateConsumerProperties(name, properties, CONSUMER_STANDARD_PROPERTIES);
return doRegisterConsumer(name, moduleInputChannel, this.pubsubChannelProvider, properties);
}
private Binding<MessageChannel> doRegisterConsumer(String name, MessageChannel moduleInputChannel,
SharedChannelProvider<?> channelProvider, Properties properties) {
Assert.hasText(name, "a valid name is required to register an inbound channel");
Assert.notNull(moduleInputChannel, "channel must not be null");
MessageChannel registeredChannel = channelProvider.lookupOrCreateSharedChannel("localbinder." + name);
bridge(name, registeredChannel, moduleInputChannel,
"inbound." + ((NamedComponent) registeredChannel).getComponentName(),
new LocalBindingPropertiesAccessor(properties));
// TODO: ?
return null;
}
/**
* Looks up or creates a DirectChannel with the given name and creates a bridge to that channel from the provided
* channel instance.
*/
@Override
public Binding<MessageChannel> bindProducer(String name, MessageChannel moduleOutputChannel, Properties properties) {
validateConsumerProperties(name, properties, PRODUCER_STANDARD_PROPERTIES);
return doRegisterProducer(name, moduleOutputChannel, this.pubsubChannelProvider, properties);
}
private Binding<MessageChannel> doRegisterProducer(String name, MessageChannel moduleOutputChannel,
SharedChannelProvider<?> channelProvider, Properties properties) {
Assert.hasText(name, "a valid name is required to register an outbound channel");
Assert.notNull(moduleOutputChannel, "channel must not be null");
MessageChannel registeredChannel = channelProvider.lookupOrCreateSharedChannel("localbinder." + name);
bridge(name, moduleOutputChannel, registeredChannel,
"outbound." + ((NamedComponent) registeredChannel).getComponentName(),
new LocalBindingPropertiesAccessor(properties));
// TODO: ?
return null;
}
@Override
public void unbind(Binding<MessageChannel> binding) {
}
protected BridgeHandler bridge(String name, MessageChannel from, MessageChannel to, String bridgeName,
LocalBindingPropertiesAccessor properties) {
return bridge(name, from, to, bridgeName, null, properties);
}
protected BridgeHandler bridge(String name, MessageChannel from, MessageChannel to, String bridgeName,
final Collection<MimeType> acceptedMimeTypes, LocalBindingPropertiesAccessor properties) {
final boolean isInbound = bridgeName.startsWith("inbound.");
BridgeHandler handler = new BridgeHandler() {
@Override
protected boolean shouldCopyRequestHeaders() {
return false;
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return requestMessage;
}
};
handler.setBeanFactory(getBeanFactory());
handler.setOutputChannel(to);
handler.setBeanName(bridgeName);
handler.afterPropertiesSet();
// Usage of a CEFB allows to handle both Subscribable & Pollable channels the same way
ConsumerEndpointFactoryBean cefb = new ConsumerEndpointFactoryBean();
cefb.setInputChannel(from);
cefb.setHandler(handler);
cefb.setBeanFactory(getBeanFactory());
if (from instanceof PollableChannel) {
cefb.setPollerMetadata(poller);
}
try {
cefb.afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
try {
cefb.getObject().setComponentName(handler.getComponentName());
Binding<MessageChannel> binding = isInbound ? Binding.forConsumer(name, null, cefb.getObject(), to, properties)
: Binding.forProducer(name, from, cefb.getObject(), properties);
addBinding(binding);
binding.start();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return handler;
}
protected <T> T getBean(String name, Class<T> requiredType) {
return getApplicationContext().getBean(name, requiredType);
}
private static class LocalBindingPropertiesAccessor extends DefaultBindingPropertiesAccessor {
public LocalBindingPropertiesAccessor(Properties properties) {
super(properties);
}
}
}

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java
View File

@@ -47,8 +47,4 @@ public class StubBinder1 implements Binder<Object> {
return null;
}
@Override
public void unbind(Binding<Object> binding) {
}
}

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java
View File

@@ -44,8 +44,4 @@ public class StubBinder2 implements Binder<Object> {
return null;
}
@Override
public void unbind(Binding<Object> binding) {
}
}

1
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java
View File

@@ -17,7 +17,6 @@
package org.springframework.cloud.stream.binder.stub2;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.stub2.StubBinder2;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

@@ -54,7 +54,6 @@ import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
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;
@@ -81,8 +80,8 @@ public class ChannelBindingServiceTests {
Binder<MessageChannel> binder = binderFactory.getBinder("mock");
ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding = Binding.forConsumer("foo", null, Mockito.mock(AbstractEndpoint.class),
inputChannel, null);
@SuppressWarnings("unchecked")
Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
when(binder.bindConsumer("foo", null, inputChannel, new Properties()))
.thenReturn(mockBinding);
Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel, inputChannelName);
@@ -91,7 +90,7 @@ public class ChannelBindingServiceTests {
assertThat(binding, sameInstance(mockBinding));
service.unbindConsumers(inputChannelName);
verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName));
verify(binder).unbind(binding);
verify(binding).unbind();
binderFactory.destroy();
}
@@ -116,10 +115,10 @@ public class ChannelBindingServiceTests {
ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding1 = Binding.forConsumer("foo", null, Mockito.mock(AbstractEndpoint.class),
inputChannel, null);
Binding<MessageChannel> mockBinding2 = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class),
inputChannel, null);
@SuppressWarnings("unchecked")
Binding<MessageChannel> mockBinding1 = Mockito.mock(Binding.class);
@SuppressWarnings("unchecked")
Binding<MessageChannel> mockBinding2 = Mockito.mock(Binding.class);
when(binder.bindConsumer("foo", null, inputChannel, new Properties()))
.thenReturn(mockBinding1);
@@ -140,8 +139,8 @@ public class ChannelBindingServiceTests {
verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName));
verify(binder).bindConsumer("bar", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName));
verify(binder).unbind(binding1);
verify(binder).unbind(binding2);
verify(binding1).unbind();
verify(binding2).unbind();
binderFactory.destroy();
}
@@ -163,8 +162,8 @@ public class ChannelBindingServiceTests {
Binder<MessageChannel> binder = binderFactory.getBinder("mock");
ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding = Binding.forConsumer("foo", "fooGroup", Mockito.mock(AbstractEndpoint.class),
inputChannel, null);
@SuppressWarnings("unchecked")
Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties()))
.thenReturn(mockBinding);
Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel, inputChannelName);
@@ -174,7 +173,7 @@ public class ChannelBindingServiceTests {
service.unbindConsumers(inputChannelName);
verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName));
verify(binder).unbind(binding);
verify(binding).unbind();
binderFactory.destroy();
}
@@ -189,9 +188,9 @@ public class ChannelBindingServiceTests {
Binder<MessageChannel> binder = binderFactory.getBinder("mock");
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class),
inputChannel, null);
@SuppressWarnings("unchecked")
Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
@SuppressWarnings("unchecked")
final AtomicReference<MessageChannel> dynamic = new AtomicReference<>();
when(binder.bindProducer(
matches("mock:bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(mockBinding);

3
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java
View File

@@ -17,14 +17,11 @@
package org.springframework.cloud.stream.utils;
import java.util.Collections;
import java.util.Properties;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderConfiguration;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.BinderType;
import org.springframework.cloud.stream.binder.BinderTypeRegistry;
import org.springframework.cloud.stream.binder.DefaultBinderFactory;
import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;