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

Binder simplification

Browse Source 
* removed pub/sub methods from binder
* consumer group is now a parameter of the remaining bindConsumer method

* remove DynamicProducer from Binder
* Move logic to create the dynamic channel to the channel resolver.
* Return bindings from bind methods and use them for unbinding
* Suffix for dlq

Move All Rabbit Binder CleanUp to Test Bindera

More RabbitMQ Binder Test Cleanup

Clean up declarations for remaining tests.

removed BinderUtils

use Redis ZSET for consumer groups

copyright dates

AutoBindDLQ: Single DLQ Per Group When Partitioned

Configure a single DLQ for each group for all partitions.

Add DLX Exchange binding for each original queue routing key, including the partition.

Fix DLQ Binding (Producer Side)

Option was not allowed and the routing key was wrong.

Add test to verify producers can be bound before consumers.

`autoBindDLQ` must be set (or reset) on both sides for success.
This commit is contained in:
Mark Fisher
2015-12-21 15:48:48 -05:00
committed by Marius Bogoevici
parent 60e44b530d
commit c3758b9dc0
46 changed files with 1052 additions and 2758 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2015 the original author or authors.
* Copyright 2014-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.
@@ -18,7 +18,9 @@ package org.springframework.cloud.stream.binder.redis;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -31,6 +33,8 @@ 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;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.channel.DirectChannel;
@@ -38,9 +42,7 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter;
import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler;
import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -56,6 +58,7 @@ import org.springframework.util.StringUtils;
/**
* A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by Redis.
*
* @author Mark Fisher
* @author Gary Russell
* @author David Turanski
@@ -65,71 +68,30 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
private static final String ERROR_HEADER = "errorKey";
private static final String CONSUMER_GROUPS_KEY_PREFIX = "groups.";
private static final SpelExpressionParser parser = new SpelExpressionParser();
private final String[] headersToMap;
/**
* Retry only.
*/
private static final Set<Object> SUPPORTED_PUBSUB_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(CONSUMER_STANDARD_PROPERTIES)
.addAll(CONSUMER_RETRY_PROPERTIES)
.build();
private final RedisOperations<String, String> redisOperations;
/**
* Retry + concurrency.
*/
private static final Set<Object> SUPPORTED_NAMED_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(CONSUMER_STANDARD_PROPERTIES)
.addAll(CONSUMER_RETRY_PROPERTIES)
.add(BinderPropertyKeys.CONCURRENCY)
.build();
/**
* Named + partitioning.
* Retry + concurrency + partitioning.
*/
private static final Set<Object> SUPPORTED_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(SUPPORTED_NAMED_CONSUMER_PROPERTIES)
.add(BinderPropertyKeys.PARTITION_INDEX)
.build();
/**
* Retry + concurrency (request).
*/
private static final Set<Object> SUPPORTED_REPLYING_CONSUMER_PROPERTIES = new SetBuilder()
// request
.addAll(CONSUMER_STANDARD_PROPERTIES)
.addAll(CONSUMER_RETRY_PROPERTIES)
.add(BinderPropertyKeys.CONCURRENCY)
.add(BinderPropertyKeys.PARTITION_INDEX)
.build();
/**
* None.
*/
private static final Set<Object> SUPPORTED_PUBSUB_PRODUCER_PROPERTIES = PRODUCER_STANDARD_PROPERTIES;
/**
* None.
*/
private static final Set<Object> SUPPORTED_NAMED_PRODUCER_PROPERTIES = PRODUCER_STANDARD_PROPERTIES;
/**
* Partitioning.
*/
private static final Set<Object> SUPPORTED_PRODUCER_PROPERTIES = new SetBuilder()
.addAll(PRODUCER_PARTITIONING_PROPERTIES)
.addAll(PRODUCER_STANDARD_PROPERTIES)
.add(BinderPropertyKeys.DIRECT_BINDING_ALLOWED)
.build();
/**
* Retry, concurrency (reply).
*/
private static final Set<Object> SUPPORTED_REQUESTING_PRODUCER_PROPERTIES = new SetBuilder()
// reply
.addAll(CONSUMER_RETRY_PROPERTIES)
.add(BinderPropertyKeys.CONCURRENCY)
.build();
private final RedisConnectionFactory connectionFactory;
@@ -143,11 +105,12 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
this(connectionFactory, new String[0]);
}
public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory,
String... headersToMap) {
public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory, String... headersToMap) {
Assert.notNull(connectionFactory, "connectionFactory must not be null");
this.connectionFactory = connectionFactory;
StringRedisTemplate template = new StringRedisTemplate(connectionFactory);
template.afterPropertiesSet();
this.redisOperations = template;
if (headersToMap != null && headersToMap.length > 0) {
String[] combinedHeadersToMap =
Arrays.copyOfRange(BinderHeaders.STANDARD_HEADERS, 0, BinderHeaders.STANDARD_HEADERS.length
@@ -159,7 +122,6 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
else {
this.headersToMap = BinderHeaders.STANDARD_HEADERS;
}
this.errorAdapter = new RedisQueueOutboundChannelAdapter(
parser.parseExpression("headers['" + ERROR_HEADER + "']"), connectionFactory);
}
@@ -173,23 +135,16 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
}
@Override
public void bindConsumer(final String name, MessageChannel moduleInputChannel, Properties properties) {
if (name.startsWith(P2P_NAMED_CHANNEL_TYPE_PREFIX)) {
validateConsumerProperties(name, properties, SUPPORTED_NAMED_CONSUMER_PROPERTIES);
}
else {
validateConsumerProperties(name, properties, SUPPORTED_CONSUMER_PROPERTIES);
}
protected Binding<MessageChannel> doBindConsumer(final String name, String group, MessageChannel moduleInputChannel, Properties properties) {
RedisPropertiesAccessor accessor = new RedisPropertiesAccessor(properties);
String queueName = "queue." + name;
String queueName = groupedName(name, group);
validateConsumerProperties(queueName, properties, SUPPORTED_CONSUMER_PROPERTIES);
int partitionIndex = accessor.getPartitionIndex();
if (partitionIndex >= 0) {
queueName += "-" + partitionIndex;
}
MessageProducerSupport adapter = createInboundAdapter(accessor, queueName);
doRegisterConsumer(name, name + (partitionIndex >= 0 ? "-" + partitionIndex : ""), moduleInputChannel, adapter,
accessor);
bindExistingProducerDirectlyIfPossible(name, moduleInputChannel);
return doRegisterConsumer(name, group, queueName, moduleInputChannel, adapter, accessor);
}
private MessageProducerSupport createInboundAdapter(RedisPropertiesAccessor accessor, String queueName) {
@@ -209,37 +164,25 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
return adapter;
}
@Override
public void bindPubSubConsumer(final String name, MessageChannel moduleInputChannel, String group,
Properties properties) {
if (logger.isInfoEnabled()) {
logger.info("declaring pubsub for inbound: " + name);
}
validateConsumerProperties(name, properties, SUPPORTED_PUBSUB_CONSUMER_PROPERTIES);
RedisInboundChannelAdapter adapter = new RedisInboundChannelAdapter(this.connectionFactory);
adapter.setBeanFactory(this.getBeanFactory());
adapter.setSerializer(null);
adapter.setTopics(applyPubSub(name));
doRegisterConsumer(name, name, moduleInputChannel, adapter, new RedisPropertiesAccessor(properties));
}
private void doRegisterConsumer(String bindingName, String channelName, MessageChannel moduleInputChannel,
private Binding<MessageChannel> doRegisterConsumer(String bindingName, String group, String channelName, MessageChannel moduleInputChannel,
MessageProducerSupport adapter, RedisPropertiesAccessor properties) {
DirectChannel bridgeToModuleChannel = new DirectChannel();
bridgeToModuleChannel.setBeanFactory(this.getBeanFactory());
bridgeToModuleChannel.setBeanName(channelName + ".bridge");
MessageChannel bridgeInputChannel = addRetryIfNeeded(channelName, bridgeToModuleChannel, properties);
adapter.setOutputChannel(bridgeInputChannel);
adapter.setBeanName("inbound." + bindingName);
adapter.setBeanName("inbound." + channelName);
adapter.afterPropertiesSet();
Binding consumerBinding = Binding.forConsumer(bindingName, adapter, moduleInputChannel, properties);
Binding<MessageChannel> consumerBinding = Binding.forConsumer(channelName, group, adapter, moduleInputChannel, properties);
addBinding(consumerBinding);
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();
return consumerBinding;
}
/**
@@ -300,101 +243,49 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
}
@Override
public void bindProducer(final String name, MessageChannel moduleOutputChannel,
Properties properties) {
Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
if (name.startsWith(P2P_NAMED_CHANNEL_TYPE_PREFIX)) {
validateProducerProperties(name, properties, SUPPORTED_NAMED_PRODUCER_PROPERTIES);
}
else {
validateProducerProperties(name, properties, SUPPORTED_PRODUCER_PROPERTIES);
}
RedisPropertiesAccessor accessor = new RedisPropertiesAccessor(properties);
if (!bindNewProducerDirectlyIfPossible(name, (SubscribableChannel) moduleOutputChannel, accessor)) {
String partitionKeyExtractorClass = accessor.getPartitionKeyExtractorClass();
Expression partitionKeyExpression = accessor.getPartitionKeyExpression();
RedisQueueOutboundChannelAdapter queue;
String queueName = "queue." + name;
if (partitionKeyExpression == null && !StringUtils.hasText(partitionKeyExtractorClass)) {
queue = new RedisQueueOutboundChannelAdapter(queueName, this.connectionFactory);
}
else {
queue = new RedisQueueOutboundChannelAdapter(
parser.parseExpression(buildPartitionRoutingExpression(queueName)), this.connectionFactory);
}
queue.setIntegrationEvaluationContext(this.evaluationContext);
queue.setBeanFactory(this.getBeanFactory());
queue.afterPropertiesSet();
doRegisterProducer(name, moduleOutputChannel, queue, accessor);
protected void afterUnbind(Binding<MessageChannel> binding) {
if (Binding.Type.consumer.equals(binding.getType())) {
String key = CONSUMER_GROUPS_KEY_PREFIX + binding.getName();
this.redisOperations.boundZSetOps(key).incrementScore(binding.getGroup(), -1);
}
}
@Override
public void bindPubSubProducer(final String name, MessageChannel moduleOutputChannel,
Properties properties) {
validateProducerProperties(name, properties, SUPPORTED_PUBSUB_PRODUCER_PROPERTIES);
RedisPublishingMessageHandler topic = new RedisPublishingMessageHandler(connectionFactory);
topic.setBeanFactory(this.getBeanFactory());
topic.setTopic(applyPubSub(name));
topic.afterPropertiesSet();
doRegisterProducer(name, moduleOutputChannel, topic, new RedisPropertiesAccessor(properties));
}
private void doRegisterProducer(final String name, MessageChannel moduleOutputChannel, MessageHandler delegate,
RedisPropertiesAccessor properties) {
this.doRegisterProducer(name, moduleOutputChannel, delegate, null, properties);
}
private void doRegisterProducer(final String name, MessageChannel moduleOutputChannel, MessageHandler delegate,
String replyTo, RedisPropertiesAccessor properties) {
public Binding<MessageChannel> bindProducer(final String name, MessageChannel moduleOutputChannel, Properties properties) {
Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
MessageHandler handler = new SendingHandler(delegate, replyTo, properties);
validateProducerProperties(name, properties, SUPPORTED_PRODUCER_PROPERTIES);
RedisPropertiesAccessor accessor = new RedisPropertiesAccessor(properties);
return doRegisterProducer(name, moduleOutputChannel, accessor);
}
private RedisQueueOutboundChannelAdapter createProducerEndpoint(String name, RedisPropertiesAccessor accessor) {
String partitionKeyExtractorClass = accessor.getPartitionKeyExtractorClass();
Expression partitionKeyExpression = accessor.getPartitionKeyExpression();
RedisQueueOutboundChannelAdapter queue;
if (partitionKeyExpression == null && !StringUtils.hasText(partitionKeyExtractorClass)) {
queue = new RedisQueueOutboundChannelAdapter(name, this.connectionFactory);
}
else {
queue = new RedisQueueOutboundChannelAdapter(
parser.parseExpression(buildPartitionRoutingExpression(name)), this.connectionFactory);
}
queue.setIntegrationEvaluationContext(this.evaluationContext);
queue.setBeanFactory(this.getBeanFactory());
queue.afterPropertiesSet();
return queue;
}
private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel, RedisPropertiesAccessor properties) {
Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
MessageHandler handler = new SendingHandler(name, properties);
EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel, handler);
consumer.setBeanFactory(this.getBeanFactory());
consumer.setBeanName("outbound." + name);
consumer.afterPropertiesSet();
Binding producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer, properties);
Binding<MessageChannel> producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer, properties);
addBinding(producerBinding);
producerBinding.start();
}
@Override
public void bindRequestor(String name, MessageChannel requests, MessageChannel replies,
Properties properties) {
if (logger.isInfoEnabled()) {
logger.info("binding requestor: " + name);
}
Assert.isInstanceOf(SubscribableChannel.class, requests);
validateProducerProperties(name, properties, SUPPORTED_REQUESTING_PRODUCER_PROPERTIES);
RedisQueueOutboundChannelAdapter queue = new RedisQueueOutboundChannelAdapter("queue." + applyRequests(name),
this.connectionFactory);
queue.setBeanFactory(this.getBeanFactory());
queue.afterPropertiesSet();
String replyQueueName = name + ".replies." + this.getIdGenerator().generateId();
RedisPropertiesAccessor accessor = new RedisPropertiesAccessor(properties);
this.doRegisterProducer(name, requests, queue, replyQueueName, accessor);
MessageProducerSupport adapter = createInboundAdapter(accessor, replyQueueName);
this.doRegisterConsumer(name, name, replies, adapter, accessor);
}
@Override
public void bindReplier(String name, MessageChannel requests, MessageChannel replies,
Properties properties) {
if (logger.isInfoEnabled()) {
logger.info("binding replier: " + name);
}
validateConsumerProperties(name, properties, SUPPORTED_REPLYING_CONSUMER_PROPERTIES);
RedisPropertiesAccessor accessor = new RedisPropertiesAccessor(properties);
MessageProducerSupport adapter = createInboundAdapter(accessor, "queue." + applyRequests(name));
this.doRegisterConsumer(name, name, requests, adapter, accessor);
RedisQueueOutboundChannelAdapter replyQueue = new RedisQueueOutboundChannelAdapter(
RedisMessageChannelBinder.parser.parseExpression("headers['" + BinderHeaders.REPLY_TO + "']"),
this.connectionFactory);
replyQueue.setBeanFactory(this.getBeanFactory());
replyQueue.setIntegrationEvaluationContext(this.evaluationContext);
replyQueue.afterPropertiesSet();
this.doRegisterProducer(name, replies, replyQueue, accessor);
return producerBinding;
}
@Override
@@ -404,37 +295,48 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
private class SendingHandler extends AbstractMessageHandler {
private final MessageHandler delegate;
private final String replyTo;
private final String bindingName;
private final PartitioningMetadata partitioningMetadata;
private final RedisPropertiesAccessor accessor;
private SendingHandler(MessageHandler delegate, String replyTo, RedisPropertiesAccessor properties) {
this.delegate = delegate;
this.replyTo = replyTo;
private final Map<String, RedisQueueOutboundChannelAdapter> adapters = new HashMap<>();
private SendingHandler(String bindingName, RedisPropertiesAccessor properties) {
this.bindingName = bindingName;
this.accessor = properties;
this.partitioningMetadata = new PartitioningMetadata(properties, properties.getNextModuleCount());
this.setBeanFactory(RedisMessageChannelBinder.this.getBeanFactory());
refreshChannelAdapters();
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
MessageValues transformed = serializePayloadIfNecessary(message);
if (replyTo != null) {
transformed.put(BinderHeaders.REPLY_TO, this.replyTo);
}
if (this.partitioningMetadata.isPartitionedModule()) {
transformed.put(PARTITION_HEADER, determinePartition(message, this.partitioningMetadata));
}
byte[] messageToSend = embeddedHeadersMessageConverter.embedHeaders(transformed,
RedisMessageChannelBinder.this.headersToMap);
delegate.handleMessage(MessageBuilder.withPayload(messageToSend).copyHeaders(transformed).build());
refreshChannelAdapters();
for (RedisQueueOutboundChannelAdapter adapter : adapters.values()) {
adapter.handleMessage((MessageBuilder.withPayload(messageToSend).copyHeaders(transformed).build()));
}
}
private void refreshChannelAdapters() {
Set<String> groups = redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + bindingName).rangeByScore(1, Double.MAX_VALUE);
for (String group : groups) {
if (!adapters.containsKey(group)) {
String channel = String.format("%s.%s", this.bindingName, group);
adapters.put(group, createProducerEndpoint(channel, accessor));
}
}
}
}
private class ReceivingHandler extends AbstractReplyProducingMessageHandler {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2015 the original author or authors.
* 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.
@@ -16,7 +16,6 @@
package org.springframework.cloud.stream.binder.redis;
import static org.hamcrest.Matchers.allOf;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
@@ -24,17 +23,15 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
@@ -52,6 +49,7 @@ import org.springframework.expression.Expression;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
@@ -60,6 +58,7 @@ import org.springframework.retry.support.RetryTemplate;
/**
* @author Gary Russell
* @author David Turanski
* @author Mark Fisher
*/
public class RedisBinderTests extends PartitionCapableBinderTests {
@@ -86,34 +85,20 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
return true;
}
@Override
@Test
@Ignore("https://github.com/spring-cloud/spring-cloud-stream/issues/247")
public void testSendAndReceivePubSub() throws Exception {
//TimeUnit.SECONDS.sleep(2);
super.testSendAndReceivePubSub();
}
@Before
public void setup() {
createTemplate().boundListOps("queue.direct.0").trim(1, 0);
}
@Test
public void testConsumerProperties() throws Exception {
Binder<MessageChannel> binder = getBinder();
Properties properties = new Properties();
properties.put("maxAttempts", "1"); // disable retry
binder.bindConsumer("props.0", new DirectChannel(), properties);
Binding<MessageChannel> binding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties);
@SuppressWarnings("unchecked")
List<Binding> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
List<Binding<MessageChannel>> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindings.size());
AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
assertEquals(binding, bindings.get(0));
AbstractEndpoint endpoint = binding.getEndpoint();
assertThat(endpoint, instanceOf(RedisQueueMessageDrivenEndpoint.class));
assertSame(DirectChannel.class, TestUtils.getPropertyValue(endpoint, "outputChannel").getClass());
binder.unbindConsumers("props.0");
binder.unbind(binding);
assertEquals(0, bindings.size());
properties.put("backOffInitialInterval", "2000");
@@ -123,48 +108,34 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
properties.put("maxAttempts", "23");
properties.put("partitionIndex", 0);
binder.bindConsumer("props.0", new DirectChannel(), properties);
binding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties);
assertEquals(1, bindings.size());
endpoint = bindings.get(0).getEndpoint();
assertEquals(binding, bindings.get(0));
endpoint = binding.getEndpoint();
verifyConsumer(endpoint);
try {
binder.bindPubSubConsumer("dummy", null, null, properties);
fail("Expected exception");
}
catch (IllegalArgumentException e) {
assertThat(e.getMessage(), allOf(
containsString(getClassUnderTestName() + " does not support consumer properties: "),
containsString("partitionIndex"),
containsString("concurrency"),
containsString(" for dummy.")));
}
try {
binder.bindConsumer("queue:dummy", null, properties);
fail("Expected exception");
}
catch (IllegalArgumentException e) {
assertEquals(getClassUnderTestName() + " does not support consumer property: partitionIndex for queue:dummy.",
e.getMessage());
}
binder.unbindConsumers("props.0");
binder.unbind(binding);
assertEquals(0, bindings.size());
}
@Test
public void testProducerProperties() throws Exception {
Binder<MessageChannel> binder = getBinder();
binder.bindProducer("props.0", new DirectChannel(), null);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), null);
Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(), null);
@SuppressWarnings("unchecked")
List<Binding> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindings.size());
AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
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);
RedisQueueOutboundChannelAdapter adapter = adapters.get("test");
assertEquals(
"queue.props.0",
TestUtils.getPropertyValue(endpoint, "handler.delegate.queueNameExpression", Expression.class).getExpressionString());
binder.unbindProducers("props.0");
assertEquals(0, bindings.size());
"props.0.test",
TestUtils.getPropertyValue(adapter, "queueNameExpression", Expression.class).getExpressionString());
binder.unbind(producerBinding);
assertEquals(1, bindings.size());
Properties properties = new Properties();
properties.put("partitionKeyExpression", "'foo'");
@@ -173,138 +144,16 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
properties.put("partitionSelectorClass", "foo");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
binder.bindProducer("props.0", new DirectChannel(), properties);
assertEquals(1, bindings.size());
endpoint = bindings.get(0).getEndpoint();
assertEquals(
"'queue.props.0-' + headers['partition']",
TestUtils.getPropertyValue(endpoint, "handler.delegate.queueNameExpression", Expression.class).getExpressionString());
try {
binder.bindPubSubProducer("dummy", null, properties);
fail("Expected exception");
}
catch (IllegalArgumentException e) {
assertThat(e.getMessage(), allOf(
containsString(getClassUnderTestName() + " does not support producer properties: "),
containsString("partitionSelectorExpression"),
containsString("partitionKeyExtractorClass"),
containsString("partitionKeyExpression"),
containsString("partitionSelectorClass")));
assertThat(e.getMessage(), containsString("for dummy."));
}
try {
binder.bindProducer("queue:dummy", new DirectChannel(), properties);
fail("Expected exception");
}
catch (IllegalArgumentException e) {
assertThat(e.getMessage(), allOf(
containsString(getClassUnderTestName() + " does not support producer properties: "),
containsString("partitionSelectorExpression"),
containsString("partitionKeyExtractorClass"),
containsString("partitionKeyExpression"),
containsString("partitionSelectorClass")));
assertThat(e.getMessage(), containsString("for queue:dummy."));
}
binder.unbindProducers("props.0");
assertEquals(0, bindings.size());
}
@Test
public void testRequestReplyRequestorProperties() throws Exception {
Binder<MessageChannel> binder = getBinder();
Properties properties = new Properties();
properties.put("backOffInitialInterval", "2000");
properties.put("backOffMaxInterval", "20000");
properties.put("backOffMultiplier", "5.0");
properties.put("concurrency", "2");
properties.put("maxAttempts", "23");
binder.bindRequestor("props.0", new DirectChannel(), new DirectChannel(), properties);
@SuppressWarnings("unchecked")
List<Binding> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties);
assertEquals(2, bindings.size());
AbstractEndpoint endpoint = bindings.get(0).getEndpoint(); // producer
endpoint = bindings.get(1).getEndpoint();
adapter = (RedisQueueOutboundChannelAdapter) TestUtils.getPropertyValue(endpoint, "handler.adapters", Map.class).get("test");
assertEquals(
"queue.props.0.requests",
TestUtils.getPropertyValue(endpoint, "handler.delegate.queueNameExpression", Expression.class).getExpressionString());
"'props.0.test-' + headers['partition']",
TestUtils.getPropertyValue(adapter, "queueNameExpression", Expression.class).getExpressionString());
endpoint = bindings.get(1).getEndpoint(); // consumer
verifyConsumer(endpoint);
properties.put("partitionKeyExpression", "'foo'");
properties.put("partitionKeyExtractorClass", "foo");
properties.put("partitionSelectorExpression", "0");
properties.put("partitionSelectorClass", "foo");
properties.put("partitionIndex", "0");
try {
binder.bindRequestor("dummy", new DirectChannel(), new DirectChannel(), properties);
fail("Expected exception");
}
catch (IllegalArgumentException e) {
assertThat(e.getMessage(), allOf(
containsString(getClassUnderTestName() + " does not support producer properties: "),
containsString("partitionSelectorExpression"),
containsString("partitionKeyExtractorClass"),
containsString("partitionKeyExpression"),
containsString("partitionSelectorClass")));
assertThat(e.getMessage(), allOf(containsString("partitionIndex"), containsString("for dummy.")));
}
binder.unbindConsumers("props.0");
binder.unbindProducers("props.0");
assertEquals(0, bindings.size());
}
@Test
public void testRequestReplyReplierProperties() throws Exception {
Binder<MessageChannel> binder = getBinder();
Properties properties = new Properties();
properties.put("backOffInitialInterval", "2000");
properties.put("backOffMaxInterval", "20000");
properties.put("backOffMultiplier", "5.0");
properties.put("concurrency", "2");
properties.put("maxAttempts", "23");
binder.bindReplier("props.0", new DirectChannel(), new DirectChannel(), properties);
@SuppressWarnings("unchecked")
List<Binding> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(2, bindings.size());
AbstractEndpoint endpoint = bindings.get(1).getEndpoint(); // producer
assertEquals(
"headers['replyTo']",
TestUtils.getPropertyValue(endpoint, "handler.delegate.queueNameExpression", Expression.class).getExpressionString());
endpoint = bindings.get(0).getEndpoint(); // consumer
verifyConsumer(endpoint);
properties.put("partitionKeyExpression", "'foo'");
properties.put("partitionKeyExtractorClass", "foo");
properties.put("partitionSelectorExpression", "0");
properties.put("partitionSelectorClass", "foo");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
properties.put("partitionIndex", "0");
try {
binder.bindReplier("dummy", new DirectChannel(), new DirectChannel(), properties);
fail("Expected exception");
}
catch (IllegalArgumentException e) {
assertThat(e.getMessage(), allOf(
containsString(getClassUnderTestName() + " does not support consumer properties: "),
containsString("partitionSelectorExpression"),
containsString("partitionKeyExtractorClass"),
containsString("partitionKeyExpression"),
containsString("partitionSelectorClass")));
assertThat(e.getMessage(), allOf(containsString("partitionIndex"), containsString("for dummy.")));
}
binder.unbindConsumers("props.0");
binder.unbindProducers("props.0");
binder.unbind(producerBinding);
binder.unbind(consumerBinding);
assertEquals(0, bindings.size());
}
@@ -335,12 +184,13 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
props.put("maxAttempts", 2);
props.put("backOffInitialInterval", 100);
props.put("backOffMultiplier", "1.0");
binder.bindConsumer("retry.0", new DirectChannel(), props); // no subscriber
Binding<MessageChannel> consumerBinding = binder.bindConsumer("retry.0", "test", new DirectChannel(), props); // no subscriber
channel.send(new GenericMessage<String>("foo"));
RedisTemplate<String, Object> template = createTemplate();
Object rightPop = template.boundListOps("ERRORS:retry.0").rightPop(5, TimeUnit.SECONDS);
Object rightPop = template.boundListOps("ERRORS:retry.0.test").rightPop(5, TimeUnit.SECONDS);
assertNotNull(rightPop);
assertThat(new String((byte[]) rightPop), containsString("foo"));
binder.unbind(consumerBinding);
}
@Test
@@ -352,10 +202,6 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
assertTrue(headers.contains("bar"));
}
@Override @Ignore("https://github.com/spring-cloud/spring-cloud-stream/issues/247")
public void createInboundPubSubBeforeOutboundPubSub() throws Exception {
}
private RedisTemplate<String, Object> createTemplate() {
if (this.redisTemplate != null) {
return this.redisTemplate;
@@ -370,13 +216,15 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
}
@Override
@SuppressWarnings("unchecked")
protected String getEndpointRouting(AbstractEndpoint endpoint) {
return TestUtils.getPropertyValue(endpoint, "handler.delegate.queueNameExpression", Expression.class).getExpressionString();
Map<String, RedisQueueOutboundChannelAdapter> adapters = TestUtils.getPropertyValue(endpoint, "handler.adapters", Map.class);
return TestUtils.getPropertyValue(adapters.values().iterator().next(), "queueNameExpression", Expression.class).getExpressionString();
}
@Override
protected String getPubSubEndpointRouting(AbstractEndpoint endpoint) {
return TestUtils.getPropertyValue(endpoint, "handler.delegate.topicExpression", Expression.class).getExpressionString();
protected String getExpectedRoutingBaseDestination(String name, String group) {
return name + "." + group;
}
@Override
@@ -391,7 +239,7 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
@Override
public Object receive(boolean expectNull) throws Exception {
byte[] bytes = (byte[]) template.boundListOps("queue." + queue).rightPop(50, TimeUnit.MILLISECONDS);
byte[] bytes = (byte[]) template.boundListOps(queue).rightPop(50, TimeUnit.MILLISECONDS);
if (bytes == null) {
return null;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-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,18 +16,16 @@
package org.springframework.cloud.stream.binder.redis;
import org.springframework.cloud.stream.binder.AbstractTestBinder;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.integration.channel.DefaultHeaderChannelRegistry;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.cloud.stream.binder.AbstractTestBinder;
/**
* Test support class for {@link RedisMessageChannelBinder}.