From 6f1f246dfba422d728f86644aab4be6a97c5ad73 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 30 Nov 2015 13:22:28 -0500 Subject: [PATCH] GH-211: PubSub Group Default to Random UUID Resolves #211 Add a default UUID to the `BindingProperties.group` so a pubsub binding is unique unless the group is specifically set. --- .../binder/rabbit/RabbitBinderTests.java | 3 +- .../stream/binding/ChannelBindingService.java | 20 +++-- .../stream/config/BindingProperties.java | 4 +- .../binding/ChannelBindingServiceTests.java | 76 +++++++++++++++++++ 4 files changed, 95 insertions(+), 8 deletions(-) create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java 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 dbcc5f225..1042f2ed5 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 @@ -447,8 +447,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { assertTrue(n < 100); binder.unbindConsumer("durabletest.0", moduleInputChannel); - binder.unbindPubSubConsumers("durabletest.0", "tgroup1"); - binder.unbindPubSubConsumers("durabletest.0", "tgroup2"); + binder.unbindPubSubConsumers("durabletest.0", "tgroup"); assertNotNull(admin.getQueueProperties(TEST_PREFIX + "tgroup.durabletest.0.dlq")); admin.deleteQueue(TEST_PREFIX + "tgroup.durabletest.0.dlq"); admin.deleteQueue(TEST_PREFIX + "tgroup.durabletest.0"); 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 231397b8e..ac22d826c 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 @@ -32,6 +32,7 @@ import org.springframework.util.StringUtils; * @author Dave Syer * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Gary Russell */ public class ChannelBindingService { @@ -47,11 +48,8 @@ public class ChannelBindingService { public void bindConsumer(MessageChannel inputChannel, String inputChannelName) { String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(inputChannelName); if (BinderUtils.isChannelPubSub(channelBindingTarget)) { - BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindings() - .get(inputChannelName); - String group = bindingProperties == null ? null : bindingProperties.getGroup(); this.binder.bindPubSubConsumer(removePrefix(channelBindingTarget), - inputChannel, group, + inputChannel, consumerGroup(inputChannelName), this.channelBindingServiceProperties.getConsumerProperties(inputChannelName)); } else { @@ -78,10 +76,22 @@ public class ChannelBindingService { } public void unbindConsumers(String inputChannelName) { - this.binder.unbindConsumers(inputChannelName); + if (BinderUtils.isChannelPubSub(this.channelBindingServiceProperties.getBindingDestination(inputChannelName))) { + this.binder.unbindPubSubConsumers(inputChannelName, consumerGroup(inputChannelName)); + } + else { + this.binder.unbindConsumers(inputChannelName); + } } public void unbindProducers(String outputChannelName) { this.binder.unbindProducers(outputChannelName); } + + private String consumerGroup(String inputChannelName) { + BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindings() + .get(inputChannelName); + return bindingProperties == null ? null : bindingProperties.getGroup(); + } + } 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 a0373cf92..23fb216ae 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,6 +16,8 @@ package org.springframework.cloud.stream.config; +import java.util.UUID; + import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonInclude.Include; @@ -43,7 +45,7 @@ public class BindingProperties { private String partitionSelectorExpression; - private String group; + private String group = UUID.randomUUID().toString(); private String contentType; 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 new file mode 100644 index 000000000..73e7756a3 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java @@ -0,0 +1,76 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.cloud.stream.binding; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.util.HashMap; +import java.util.Map; + +import org.junit.Test; + +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.messaging.MessageChannel; + +/** + * @author Gary Russell + * + */ +public class ChannelBindingServiceTests { + + @Test + public void testSimple() { + ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); + Map bindings = new HashMap<>(); + BindingProperties props = new BindingProperties(); + props.setDestination("foo"); + String name = "foo"; + bindings.put(name, props); + properties.setBindings(bindings); + @SuppressWarnings("unchecked") + Binder binder = mock(Binder.class); + ChannelBindingService service = new ChannelBindingService(properties, binder); + MessageChannel inputChannel = new DirectChannel(); + service.bindConsumer(inputChannel, name); + service.unbindConsumers(name); + verify(binder).bindConsumer(name, inputChannel, properties.getConsumerProperties(name)); + verify(binder).unbindConsumers(name); + } + + @Test + public void testPubSub() { + ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); + Map bindings = new HashMap<>(); + BindingProperties props = new BindingProperties(); + props.setDestination("topic:foo"); + String name = "foo"; + bindings.put(name, props); + properties.setBindings(bindings); + @SuppressWarnings("unchecked") + Binder binder = mock(Binder.class); + ChannelBindingService service = new ChannelBindingService(properties, binder); + MessageChannel inputChannel = new DirectChannel(); + service.bindConsumer(inputChannel, name); + service.unbindConsumers(name); + verify(binder).bindPubSubConsumer(name, inputChannel, props.getGroup(), properties.getConsumerProperties(name)); + verify(binder).unbindPubSubConsumers(name, props.getGroup()); + } + +}