From 192aa79baa881e01f197ebf1a58651de4b4a0106 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Tue, 26 Jan 2016 19:13:34 -0500 Subject: [PATCH] Remove default groups, support default pubsub with empty group Addressing PR comments --- .../kafka/KafkaMessageChannelBinder.java | 6 +- .../rabbit/RabbitMessageChannelBinder.java | 66 +++++++++++++------ .../binder/rabbit/RabbitBinderTests.java | 18 +++-- .../binder/rabbit/RabbitTestBinder.java | 5 +- .../redis/RedisMessageChannelBinder.java | 8 ++- .../binder/PartitionCapableBinderTests.java | 53 +++++++++++++++ .../stream/annotation/EnableBinding.java | 1 + .../cloud/stream/binder/Binder.java | 3 +- .../cloud/stream/binder/Binding.java | 22 ++++--- ... => DefaultBindingPropertiesAccessor.java} | 5 +- .../binder/MessageChannelBinderSupport.java | 19 +++--- .../stream/binding/ChannelBindingService.java | 21 +++++- .../stream/config/BindingProperties.java | 10 +-- .../local/LocalMessageChannelBinder.java | 4 +- .../binding/ChannelBindingServiceTests.java | 16 +++++ 15 files changed, 193 insertions(+), 64 deletions(-) rename spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/{AbstractBindingPropertiesAccessor.java => DefaultBindingPropertiesAccessor.java} (98%) diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 249f7f13c..9cc69a56e 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -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); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 864d4c7bf..7cd952f3c 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -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 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 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 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). diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 7ecff915b..c3cb521b6 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -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 consumerBinding = binder.bindConsumer("dlqtest", null, moduleInputChannel, properties); + Binding 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 input0Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input0, properties); - Binding defaultConsumerBinding1 = binder.bindConsumer("partDLQ.0", null, new QueueChannel(), properties); + Binding defaultConsumerBinding1 = + binder.bindConsumer("partDLQ.0", "default", new QueueChannel(), properties); properties.put("partitionIndex", "1"); DirectChannel input1 = new DirectChannel(); input1.setBeanName("test.input1DLQ"); Binding input1Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input1, properties); - Binding defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", null, new QueueChannel(), properties); + Binding 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 input0Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input0, properties); - Binding defaultConsumerBinding1 = binder.bindConsumer("partDLQ.1", null, new QueueChannel(), properties); + Binding defaultConsumerBinding1 = binder.bindConsumer("partDLQ.1", "defaultConsumer", new QueueChannel(), properties); properties.put("partitionIndex", "1"); DirectChannel input1 = new DirectChannel(); input1.setBeanName("test.input1DLQ"); Binding input1Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input1, properties); - Binding defaultConsumerBinding2 = binder.bindConsumer("partDLQ.1", null, new QueueChannel(), properties); + Binding 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() { diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 65651b88d..653e0c755 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -69,10 +69,11 @@ public class RabbitTestBinder extends AbstractTestBinder 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 diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java index 6ebe5af68..05f705af3 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java @@ -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 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); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java index e3c2ad6fb..7b7e9e452 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java @@ -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 binder = getBinder(); + DirectChannel output = new DirectChannel(); + Properties properties = new Properties(); + Binding producerBinding = binder.bindProducer("defaultGroup.0", output, properties); + + QueueChannel input1 = new QueueChannel(); + Binding binding1 = binder.bindConsumer("defaultGroup.0", null, input1, properties); + + QueueChannel input2 = new QueueChannel(); + Binding binding2 = binder.bindConsumer("defaultGroup.0", null, input2, properties); + + String testPayload1 = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload1.getBytes())); + + Message receivedMessage1 = (Message) input1.receive(1000); + assertThat(receivedMessage1, not(nullValue())); + assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload1)); + + Message receivedMessage2 = (Message) 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) input1.receive(1000); + assertThat(receivedMessage1, not(nullValue())); + assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload2)); + receivedMessage1 = (Message) input1.receive(1000); + assertThat(receivedMessage1, not(nullValue())); + assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload3)); + + receivedMessage2 = (Message) 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 binder = getBinder(); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java index 18e02eda2..847161835 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java @@ -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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java index fdd4b66dc..52eaa693a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java @@ -35,7 +35,8 @@ public interface Binder { * 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 null or empty String, the "default" group will be used) + * in the same group (a null 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 */ diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java index f8e9884fa..90f469cb6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java @@ -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 implements Lifecycle { - public static enum Type { + public enum Type { producer, consumer } @@ -45,10 +47,10 @@ public class Binding 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 implements Lifecycle { } public static Binding forConsumer(String name, String group, AbstractEndpoint adapterFromBinder, T inputTarget, - AbstractBindingPropertiesAccessor properties) { - return new Binding(name, group, inputTarget, adapterFromBinder, Type.consumer, properties); + DefaultBindingPropertiesAccessor properties) { + return new Binding<>(name, group, inputTarget, adapterFromBinder, Type.consumer, properties); } public static Binding forProducer(String name, T outputTarget, AbstractEndpoint adapterToBinder, - AbstractBindingPropertiesAccessor properties) { - return new Binding(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 implements Lifecycle { return type; } - public AbstractBindingPropertiesAccessor getPropertiesAccessor() { + public DefaultBindingPropertiesAccessor getPropertiesAccessor() { return properties; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBindingPropertiesAccessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBindingPropertiesAccessor.java similarity index 98% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBindingPropertiesAccessor.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBindingPropertiesAccessor.java index 55ad48f6f..3df26354b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBindingPropertiesAccessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBindingPropertiesAccessor.java @@ -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(); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java index b3b040da2..3eee0ac1d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java @@ -82,11 +82,6 @@ public abstract class MessageChannelBinderSupport protected static final String PARTITION_HEADER = "partition"; - /** - * Default group name (used if null 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> payloadTypeCache = new ConcurrentHashMap<>(); @@ -368,7 +363,13 @@ public abstract class MessageChannelBinderSupport @Override public final Binding 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(); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java index bb24a6a57..fa448b80b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java @@ -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 binderFactory; private final ChannelBindingServiceProperties channelBindingServiceProperties; @@ -72,12 +77,24 @@ public class ChannelBindingService { public void unbindConsumers(String inputChannelName) { Binder binder = getBinderForChannel(inputChannelName); - binder.unbind(this.consumerBindings.remove(inputChannelName)); + Binding 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 binder = getBinderForChannel(outputChannelName); - binder.unbind(this.producerBindings.remove(outputChannelName)); + Binding 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 getBinderForChannel(String channelName) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index b4408e243..f6fb87923 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -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 diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java index 4001d515e..e75acdf25 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java @@ -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); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java index 761462cbf..41f761af8 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java @@ -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 binder = binderFactory.getBinder("mock"); ChannelBindingService service = new ChannelBindingService(properties, binderFactory); MessageChannel inputChannel = new DirectChannel(); + Binding mockBinding = Binding.forConsumer("foo", null, Mockito.mock(AbstractEndpoint.class), + inputChannel, null); + Mockito.when(binder.bindConsumer("foo", null, inputChannel, new Properties())) + .thenReturn(mockBinding); Binding 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 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 binder = binderFactory.getBinder("mock"); ChannelBindingService service = new ChannelBindingService(properties, binderFactory); MessageChannel inputChannel = new DirectChannel(); + Binding mockBinding = Binding.forConsumer("foo", "fooGroup", Mockito.mock(AbstractEndpoint.class), + inputChannel, null); + Mockito.when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties())) + .thenReturn(mockBinding); Binding 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);