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.

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");

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();
+	}
+
 }

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;
 

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<String, BindingProperties> bindings = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		props.setDestination("foo");
+		String name = "foo";
+		bindings.put(name, props);
+		properties.setBindings(bindings);
+		@SuppressWarnings("unchecked")
+		Binder<MessageChannel> 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<String, BindingProperties> bindings = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		props.setDestination("topic:foo");
+		String name = "foo";
+		bindings.put(name, props);
+		properties.setBindings(bindings);
+		@SuppressWarnings("unchecked")
+		Binder<MessageChannel> 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());
+	}
+
+}