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

Remove default groups, support default pubsub with empty group

Browse Source 
Addressing PR comments
This commit is contained in:
Marius Bogoevici
2016-01-26 19:13:34 -05:00
parent c3758b9dc0
commit 192aa79baa
15 changed files with 193 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -37,7 +37,7 @@ 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.AbstractBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderException;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
@@ -448,7 +448,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
// but multiple instances of this binding will each get all messages
// PubSub consumers reset at the latest time, which allows them to receive only messages sent after
// they've been bound
String consumerGroup = group == null ? UUID.randomUUID().toString() : group;
String consumerGroup = group == null ? "anonymous." + UUID.randomUUID().toString() : group;
return createKafkaConsumer(name, inputChannel, properties, consumerGroup, OffsetRequest.LatestTime());
}
@@ -732,7 +732,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
}
}
private class KafkaPropertiesAccessor extends AbstractBindingPropertiesAccessor {
private class KafkaPropertiesAccessor extends DefaultBindingPropertiesAccessor {
public KafkaPropertiesAccessor(Properties properties) {
super(properties);

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

@@ -26,12 +26,17 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import org.aopalliance.aop.Advice;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.UncategorizedAmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AnonymousQueue;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
@@ -60,9 +65,9 @@ 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.AbstractBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
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;
@@ -91,10 +96,6 @@ import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
/**
* A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ.
*
@@ -104,9 +105,13 @@ import com.rabbitmq.client.Envelope;
* @author Gunnar Hillert
* @author Ilayaperumal Gopinathan
* @author David Turanski
* @author Marius Bogoevici
*/
public class RabbitMessageChannelBinder extends MessageChannelBinderSupport implements DisposableBean {
public static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
= new AnonymousQueue.Base64UrlNamingStrategy("anonymous.");
private static final AcknowledgeMode DEFAULT_ACKNOWLEDGE_MODE = AcknowledgeMode.AUTO;
private static final MessageDeliveryMode DEFAULT_DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT;
@@ -389,35 +394,44 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
@Override
public Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel, Properties properties) {
String baseQueueName = groupedName(name, group);
boolean anonymousConsumer = !StringUtils.hasText(group);
String baseQueueName = anonymousConsumer ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
: groupedName(name, group);
if (this.logger.isInfoEnabled()) {
this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name);
}
RabbitPropertiesAccessor accessor = new RabbitPropertiesAccessor(properties);
validateConsumerProperties(baseQueueName, properties, SUPPORTED_CONSUMER_PROPERTIES);
RabbitPropertiesAccessor accessor = new RabbitPropertiesAccessor(properties);
String prefix = accessor.getPrefix(this.defaultPrefix);
String exchangeName = applyPrefix(prefix, name);
TopicExchange exchange = new TopicExchange(exchangeName);
declareExchange(exchangeName, exchange);
String queueName = applyPrefix(prefix, baseQueueName);
int partitionIndex = accessor.getPartitionIndex();
if (partitionIndex >= 0) {
String partitionSuffix = "-" + partitionIndex;
queueName += partitionSuffix;
}
boolean partitioned = !anonymousConsumer && accessor.getPartitionIndex() >= 0;
boolean durable = !anonymousConsumer && accessor.isDurable(this.defaultDurableSubscription);
Queue queue;
boolean durable = accessor.isDurable(this.defaultDurableSubscription);
if (durable) {
queue = new Queue(queueName, true, false, false, queueArgs(accessor, queueName));
if (anonymousConsumer) {
queue = new Queue(queueName, false, true, true);
}
else {
queue = new Queue(queueName, false, false, true);
if (partitioned) {
String partitionSuffix = "-" + accessor.getPartitionIndex();
queueName += partitionSuffix;
}
if (durable) {
queue = new Queue(queueName, true, false, false, queueArgs(accessor, queueName));
}
else {
queue = new Queue(queueName, false, false, true);
}
}
declareQueue(queueName, queue);
if (partitionIndex >= 0) {
String bindingKey = String.format("%s-%d", name, partitionIndex);
if (partitioned) {
String bindingKey = String.format("%s-%d", name, accessor.getPartitionIndex());
declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(bindingKey));
}
else {
@@ -428,6 +442,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
autoBindDLQ(applyPrefix(prefix, baseQueueName), queueName, accessor);
}
return binding;
}
private Map<String, Object> queueArgs(RabbitPropertiesAccessor accessor, String queueName) {
@@ -658,6 +673,15 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
}
}
catch (UncategorizedAmqpException e) {
if (e.getCause() instanceof NullPointerException) {
// Temporary fix for https://jira.spring.io/browse/AMQP-565
// TODO remove once Spring AMQP is upgraded beyond 1.5.4
}
else {
throw e;
}
}
addToAutoDeclareContext(beanName, queue);
}
@@ -692,7 +716,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
@Override
protected void afterUnbind(Binding<MessageChannel> binding) {
if (Binding.Type.consumer.equals(binding.getType())) {
cleanAutoDeclareContext(groupedName(binding.getName(), binding.getGroup()));
cleanAutoDeclareContext(binding.getName());
}
}
@@ -824,7 +848,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
* Property accessor for the RabbitBinder. Refer to the Spring-AMQP documentation for information on the
* specific properties.
*/
private static class RabbitPropertiesAccessor extends AbstractBindingPropertiesAccessor {
private static class RabbitPropertiesAccessor extends DefaultBindingPropertiesAccessor {
/**
* The acknowledge mode (i.e. NONE, MANUAL, AUTO).

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

@@ -138,7 +138,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertEquals(AcknowledgeMode.AUTO, container.getAcknowledgeMode());
assertEquals(RabbitMessageChannelBinder.DEFAULT_RABBIT_PREFIX + "props.0.default", container.getQueueNames()[0]);
assertThat(container.getQueueNames()[0],
startsWith(RabbitMessageChannelBinder.DEFAULT_RABBIT_PREFIX));
assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class));
assertEquals(1, TestUtils.getPropertyValue(container, "concurrentConsumers"));
assertNull(TestUtils.getPropertyValue(container, "maxConcurrentConsumers"));
@@ -301,6 +302,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put("autoBindDLQ", "true");
properties.put("maxAttempts", "1"); // disable retry
properties.put("requeue", "false");
properties.put("durableSubscription","true");
DirectChannel moduleInputChannel = new DirectChannel();
moduleInputChannel.setBeanName("dlqTest");
moduleInputChannel.subscribe(new MessageHandler() {
@@ -311,7 +313,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqtest", null, moduleInputChannel, properties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqtest", "default", moduleInputChannel, properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo");
@@ -339,15 +341,17 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put("maxAttempts", "1"); // disable retry
properties.put("requeue", "false");
properties.put("partitionIndex", "0");
properties.put("durableSubscription","true");
DirectChannel input0 = new DirectChannel();
input0.setBeanName("test.input0DLQ");
Binding<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input0, properties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partDLQ.0", null, new QueueChannel(), properties);
Binding<MessageChannel> defaultConsumerBinding1 =
binder.bindConsumer("partDLQ.0", "default", new QueueChannel(), properties);
properties.put("partitionIndex", "1");
DirectChannel input1 = new DirectChannel();
input1.setBeanName("test.input1DLQ");
Binding<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input1, properties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", null, new QueueChannel(), properties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", "default", new QueueChannel(), properties);
properties.clear();
properties.put("prefix", "bindertest.");
@@ -434,15 +438,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put("maxAttempts", "1"); // disable retry
properties.put("requeue", "false");
properties.put("partitionIndex", "0");
properties.put(BinderPropertyKeys.DURABLE,"true");
DirectChannel input0 = new DirectChannel();
input0.setBeanName("test.input0DLQ");
Binding<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input0, properties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partDLQ.1", null, new QueueChannel(), properties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partDLQ.1", "defaultConsumer", new QueueChannel(), properties);
properties.put("partitionIndex", "1");
DirectChannel input1 = new DirectChannel();
input1.setBeanName("test.input1DLQ");
Binding<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input1, properties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.1", null, new QueueChannel(), properties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.1", "defaultConsumer", new QueueChannel(), properties);
final CountDownLatch latch0 = new CountDownLatch(1);
input0.subscribe(new MessageHandler() {
@@ -516,6 +521,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put("republishToDLQ", "true");
properties.put("maxAttempts", "1"); // disable retry
properties.put("requeue", "false");
properties.put("durableSubscription", "true");
DirectChannel moduleInputChannel = new DirectChannel();
moduleInputChannel.setBeanName("dlqPubTest");
moduleInputChannel.subscribe(new MessageHandler() {

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

@@ -69,10 +69,11 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel, Properties properties) {
this.queues.add(prefix(properties) + name + (group == null ? ".default" : "." + group));
if (group != null) {
this.queues.add(prefix(properties) + name + ("." + group));
}
this.exchanges.add(prefix(properties) + name);
return super.bindConsumer(name, group, moduleInputChannel, properties);
}
@Override

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

@@ -23,9 +23,10 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.cloud.stream.binder.AbstractBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
@@ -136,6 +137,9 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
@Override
protected Binding<MessageChannel> doBindConsumer(final String name, String group, MessageChannel moduleInputChannel, Properties properties) {
if (!StringUtils.hasText(group)) {
group = "anonymous." + UUID.randomUUID().toString();
}
RedisPropertiesAccessor accessor = new RedisPropertiesAccessor(properties);
String queueName = groupedName(name, group);
validateConsumerProperties(queueName, properties, SUPPORTED_CONSUMER_PROPERTIES);
@@ -368,7 +372,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
}
private static class RedisPropertiesAccessor extends AbstractBindingPropertiesAccessor {
private static class RedisPropertiesAccessor extends DefaultBindingPropertiesAccessor {
public RedisPropertiesAccessor(Properties properties) {
super(properties);

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

@@ -21,6 +21,8 @@ import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasProperty;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
@@ -28,6 +30,7 @@ import static org.junit.Assert.assertThat;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.UUID;
import org.hamcrest.CustomMatcher;
import org.hamcrest.Matcher;
@@ -51,6 +54,56 @@ import org.springframework.messaging.support.GenericMessage;
*/
abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
@Test
@SuppressWarnings("unchecked")
public void testAnonymousGroup() throws Exception {
Binder<MessageChannel> binder = getBinder();
DirectChannel output = new DirectChannel();
Properties properties = new Properties();
Binding<MessageChannel> producerBinding = binder.bindProducer("defaultGroup.0", output, properties);
QueueChannel input1 = new QueueChannel();
Binding<MessageChannel> binding1 = binder.bindConsumer("defaultGroup.0", null, input1, properties);
QueueChannel input2 = new QueueChannel();
Binding<MessageChannel> binding2 = binder.bindConsumer("defaultGroup.0", null, input2, properties);
String testPayload1 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload1.getBytes()));
Message<byte[]> receivedMessage1 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload1));
Message<byte[]> receivedMessage2 = (Message<byte[]>) input2.receive(1000);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload1));
binder.unbind(binding2);
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
binding2 = binder.bindConsumer("defaultGroup.0", null, input2, properties);
String testPayload3 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload3.getBytes()));
receivedMessage1 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload2));
receivedMessage1 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload3));
receivedMessage2 = (Message<byte[]>) input2.receive(1000);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload3));
binder.unbind(producerBinding);
binder.unbind(binding1);
binder.unbind(binding2);
}
@Test
public void testBadProperties() throws Exception {
Binder<MessageChannel> binder = getBinder();

1
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java
View File

@@ -34,6 +34,7 @@ import org.springframework.integration.config.EnableIntegration;
/**
* Enables the binding of inputs and outputs to a broker, according to the list
* of interfaces passed as value to the annotation.
*
* @author Dave Syer
* @author Marius Bogoevici
* @author David Turanski

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

@@ -35,7 +35,8 @@ public interface Binder<T> {
* Bind a message consumer on a channel
* @param name the logical identity of the message source
* @param group the consumer group to which this consumer belongs - subscriptions are shared among consumers
* in the same group (if <code>null</code> or empty String, the "default" group will be used)
* in the same group (a <code>null</code> or empty String, must be treated as an anonymous group that doesn't share
* the subscription with any other consumer)
* @param inboundBindTarget the module interface to be bound as a consumer
* @param properties arbitrary String key/value pairs that will be used in the binding
*/

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

@@ -21,17 +21,19 @@ import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.util.Assert;
/**
* Represents a binding between a channel and an adapter endpoint that connects via a Binder. The binding
* 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
* input channel. A producer binding represents a connection from an output channel to an adapter.
* input. A producer binding represents a connection from an output to an adapter.
*
* @author Jennifer Hickey
* @author Mark Fisher
* @author Gary Russell
* @author Marius Bogoevici
* @see org.springframework.cloud.stream.annotation.EnableBinding
*/
public class Binding<T> implements Lifecycle {
public static enum Type {
public enum Type {
producer, consumer
}
@@ -45,10 +47,10 @@ public class Binding<T> implements Lifecycle {
private final Type type;
private final AbstractBindingPropertiesAccessor properties;
private final DefaultBindingPropertiesAccessor properties;
private Binding(String name, String group, T target, AbstractEndpoint endpoint, Type type,
AbstractBindingPropertiesAccessor properties) {
DefaultBindingPropertiesAccessor properties) {
Assert.notNull(target, "target must not be null");
Assert.notNull(endpoint, "endpoint must not be null");
this.name = name;
@@ -60,13 +62,13 @@ public class Binding<T> implements Lifecycle {
}
public static <T> Binding<T> forConsumer(String name, String group, AbstractEndpoint adapterFromBinder, T inputTarget,
AbstractBindingPropertiesAccessor properties) {
return new Binding<T>(name, group, inputTarget, adapterFromBinder, Type.consumer, properties);
DefaultBindingPropertiesAccessor properties) {
return new Binding<>(name, group, inputTarget, adapterFromBinder, Type.consumer, properties);
}
public static <T> Binding<T> forProducer(String name, T outputTarget, AbstractEndpoint adapterToBinder,
AbstractBindingPropertiesAccessor properties) {
return new Binding<T>(name, null, outputTarget, adapterToBinder, Type.producer, properties);
DefaultBindingPropertiesAccessor properties) {
return new Binding<>(name, null, outputTarget, adapterToBinder, Type.producer, properties);
}
public String getName() {
@@ -89,7 +91,7 @@ public class Binding<T> implements Lifecycle {
return type;
}
public AbstractBindingPropertiesAccessor getPropertiesAccessor() {
public DefaultBindingPropertiesAccessor getPropertiesAccessor() {
return properties;
}

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBindingPropertiesAccessor.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBindingPropertiesAccessor.java
View File

@@ -28,14 +28,15 @@ import org.springframework.util.StringUtils;
* are defined here.
*
* @author Gary Russell
* @author Marius Bogoevici
*/
public abstract class AbstractBindingPropertiesAccessor {
public class DefaultBindingPropertiesAccessor {
private static final SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
private final Properties properties;
public AbstractBindingPropertiesAccessor(Properties properties) {
public DefaultBindingPropertiesAccessor(Properties properties) {
if (properties == null) {
this.properties = new Properties();
}

19
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java
View File

@@ -82,11 +82,6 @@ public abstract class MessageChannelBinderSupport
protected static final String PARTITION_HEADER = "partition";
/**
* Default group name (used if <code>null</code> or empty String is provided).
*/
protected static final String DEFAULT_CONSUMER_GROUP = "default";
/**
* The delimiter between a group and index when constructing a binder consumer/producer.
*/
@@ -206,7 +201,7 @@ public abstract class MessageChannelBinderSupport
protected volatile boolean defaultCompress = false;
protected volatile boolean defaultDurableSubscription = true;
protected volatile boolean defaultDurableSubscription = false;
// Payload type cache
private volatile Map<String, Class<?>> payloadTypeCache = new ConcurrentHashMap<>();
@@ -368,7 +363,13 @@ public abstract class MessageChannelBinderSupport
@Override
public final Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel inputChannel, Properties properties) {
group = (StringUtils.hasText(group)) ? group : DEFAULT_CONSUMER_GROUP;
DefaultBindingPropertiesAccessor accessor = new DefaultBindingPropertiesAccessor(properties);
if (StringUtils.isEmpty(group)) {
Assert.isTrue(!accessor.getProperty(BinderPropertyKeys.DURABLE, defaultDurableSubscription),
"A consumer group is required for a durable subscription");
Assert.isTrue(accessor.getPartitionIndex() < 0,
"A consumer group is required for a partitioned subscription");
}
return doBindConsumer(name, group, inputChannel, properties);
}
@@ -692,7 +693,7 @@ public abstract class MessageChannelBinderSupport
* @param properties The properties.
* @return The retry template, or null if retry is not enabled.
*/
protected RetryTemplate buildRetryTemplateIfRetryEnabled(AbstractBindingPropertiesAccessor properties) {
protected RetryTemplate buildRetryTemplateIfRetryEnabled(DefaultBindingPropertiesAccessor properties) {
int maxAttempts = properties.getMaxAttempts(this.defaultMaxAttempts);
if (maxAttempts > 1) {
RetryTemplate template = new RetryTemplate();
@@ -740,7 +741,7 @@ public abstract class MessageChannelBinderSupport
private final int partitionCount;
public PartitioningMetadata(AbstractBindingPropertiesAccessor properties, int partitionCount) {
public PartitioningMetadata(DefaultBindingPropertiesAccessor properties, int partitionCount) {
this.partitionCount = partitionCount;
this.partitionKeyExtractorClass = properties.getPartitionKeyExtractorClass();
this.partitionKeyExpression = properties.getPartitionKeyExpression();

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

@@ -19,6 +19,9 @@ package org.springframework.cloud.stream.binding;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.Binding;
@@ -38,6 +41,8 @@ import org.springframework.messaging.MessageChannel;
*/
public class ChannelBindingService {
private final Log log = LogFactory.getLog(ChannelBindingService.class);
private BinderFactory<MessageChannel> binderFactory;
private final ChannelBindingServiceProperties channelBindingServiceProperties;
@@ -72,12 +77,24 @@ public class ChannelBindingService {
public void unbindConsumers(String inputChannelName) {
Binder<MessageChannel> binder = getBinderForChannel(inputChannelName);
binder.unbind(this.consumerBindings.remove(inputChannelName));
Binding<MessageChannel> binding = this.consumerBindings.remove(inputChannelName);
if (binding != null) {
binder.unbind(binding);
}
else if (log.isWarnEnabled()) {
log.warn("Trying to unbind channel '" + inputChannelName + "', but no binding found.");
}
}
public void unbindProducers(String outputChannelName) {
Binder<MessageChannel> binder = getBinderForChannel(outputChannelName);
binder.unbind(this.producerBindings.remove(outputChannelName));
Binding<MessageChannel> binding = this.producerBindings.remove(outputChannelName);
if (binding != null) {
binder.unbind(binding);
}
else if (log.isWarnEnabled()) {
log.warn("Trying to unbind channel '" + outputChannelName + "', but no binding found.");
}
}
private Binder<MessageChannel> getBinderForChannel(String channelName) {

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

@@ -16,8 +16,6 @@
package org.springframework.cloud.stream.config;
import java.util.UUID;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonInclude.Include;
@@ -39,9 +37,13 @@ public class BindingProperties {
private String destination;
/**
* Unique name that the binding belongs to.
* Unique name that the binding belongs to (applies to consumers only). Multiple consumers within the same group
* share the subscription. A null or empty String value indicates an anonymous group that is not shared.
*
* @see org.springframework.cloud.stream.binder.Binder#bindConsumer(java.lang.String, java.lang.String,
* java.lang.Object, java.util.Properties)
*/
private String group = UUID.randomUUID().toString();
private String group;
// Properties for both inbound/outbound

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

@@ -19,7 +19,7 @@ package org.springframework.cloud.stream.binder.local;
import java.util.Collection;
import java.util.Properties;
import org.springframework.cloud.stream.binder.AbstractBindingPropertiesAccessor;
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;
@@ -248,7 +248,7 @@ public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
return getApplicationContext().getBean(name, requiredType);
}
private static class LocalBindingPropertiesAccessor extends AbstractBindingPropertiesAccessor {
private static class LocalBindingPropertiesAccessor extends DefaultBindingPropertiesAccessor {
public LocalBindingPropertiesAccessor(Properties properties) {
super(properties);

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binding;
import static org.hamcrest.CoreMatchers.sameInstance;
import static org.mockito.Mockito.verify;
import java.util.Collections;
@@ -23,7 +24,9 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderConfiguration;
@@ -34,11 +37,13 @@ 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;
/**
* @author Gary Russell
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class ChannelBindingServiceTests {
@@ -58,7 +63,12 @@ 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);
Mockito.when(binder.bindConsumer("foo", null, inputChannel, new Properties()))
.thenReturn(mockBinding);
Binding<MessageChannel> binding = service.bindConsumer(inputChannel, name);
Assert.assertThat(binding, sameInstance(mockBinding));
service.unbindConsumers(name);
verify(binder).bindConsumer(name, props.getGroup(), inputChannel, properties.getConsumerProperties(name));
verify(binder).unbind(binding);
@@ -71,6 +81,7 @@ public class ChannelBindingServiceTests {
Map<String, BindingProperties> bindings = new HashMap<>();
BindingProperties props = new BindingProperties();
props.setDestination("foo");
props.setGroup("fooGroup");
String name = "foo";
bindings.put(name, props);
properties.setBindings(bindings);
@@ -81,7 +92,12 @@ 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);
Mockito.when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties()))
.thenReturn(mockBinding);
Binding<MessageChannel> binding = service.bindConsumer(inputChannel, name);
Assert.assertThat(binding, sameInstance(mockBinding));
service.unbindConsumers(name);
verify(binder).bindConsumer(name, props.getGroup(), inputChannel, properties.getConsumerProperties(name));
verify(binder).unbind(binding);