Native multiple input binding changes

If the binder implementation enables native multiple input binding, skip individual
consumer binding per binding target (for example - from a comma separated targets
provided as input destination property).

By default, native multiple input binding is disabled at the core framework level
and the binder implementation explicitly needs to override that to enable it.

Fixes #1365

Chaning the boolean flag for multiple input binding to multiplex

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java

@@ -123,6 +123,18 @@ public class ConsumerProperties {
 	 */
 	private boolean useNativeDecoding;
 
+	/**
+	 * When set to true, the underlying binder will natively multiplex destinations on the same input binding.
+	 * For example, in the case of a comma separated multiple destinations, the core framework will skip binding
+	 * them individually if this is set to true, but delegate that responsibility to the binder.
+	 *
+	 * By default this property is set to `false` and the binder will individually bind each destinations in case
+	 * of a comma separated multi destination list. The individual binder implementations that need to support multiple
+	 * input bindings natively (multiplex) can enable this property. Under normal circumstances, the end users are
+	 * not expected to enable or disable this property directly.
+	 */
+	private boolean multiplex;
+
 	@Min(value = 1, message = "Concurrency should be greater than zero.")
 	public int getConcurrency() {
 		return concurrency;
@@ -209,4 +221,12 @@ public class ConsumerProperties {
 	public void setUseNativeDecoding(boolean useNativeDecoding) {
 		this.useNativeDecoding = useNativeDecoding;
 	}
+
+	public boolean isMultiplex() {
+		return multiplex;
+	}
+
+	public void setMultiplex(boolean multiplex) {
+		this.multiplex = multiplex;
+	}
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java

@@ -54,6 +54,7 @@ import org.springframework.validation.beanvalidation.CustomValidatorBean;
  * @author Ilayaperumal Gopinathan
  * @author Gary Russell
  * @author Janne Valkealahti
+ * @author Soby Chacko
  */
 public class BindingService {
 
@@ -89,10 +90,6 @@ public class BindingService {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public <T> Collection<Binding<T>> bindConsumer(T input, String inputName) {
-		String bindingTarget = this.bindingServiceProperties
-				.getBindingDestination(inputName);
-		String[] bindingTargets = StringUtils
-				.commaDelimitedListToStringArray(bindingTarget);
 		Collection<Binding<T>> bindings = new ArrayList<>();
 		Binder<T, ConsumerProperties, ?> binder = (Binder<T, ConsumerProperties, ?>) getBinder(
 				inputName, input.getClass());
@@ -106,19 +103,30 @@ public class BindingService {
 			BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties);
 			consumerProperties = extendedConsumerProperties;
 		}
+
 		validate(consumerProperties);
-		for (String target : bindingTargets) {
-			Binding<T> binding;
-			if (input instanceof PollableSource) {
-				binding = doBindPollableConsumer(input, inputName, binder, consumerProperties, target);
+
+		String bindingTarget = this.bindingServiceProperties
+				.getBindingDestination(inputName);
+
+		if (consumerProperties.isMultiplex()) {
+			bindings.add(doBindConsumer(input, inputName, binder, consumerProperties, bindingTarget));
+		}
+		else {
+			String[] bindingTargets = StringUtils
+					.commaDelimitedListToStringArray(bindingTarget);
+			for (String target : bindingTargets) {
+				Binding<T> binding;
+				if (input instanceof PollableSource) {
+					binding = doBindPollableConsumer(input, inputName, binder, consumerProperties, target);
+				} else {
+					binding = doBindConsumer(input, inputName, binder, consumerProperties, target);
+				}
+				bindings.add(binding);
 			}
-			else {
-				binding = doBindConsumer(input, inputName, binder, consumerProperties, target);
-			}
-			bindings.add(binding);
 		}
 		bindings = Collections.unmodifiableCollection(bindings);
-		this.consumerBindings.put(inputName, new ArrayList<Binding<?>>(bindings));
+		this.consumerBindings.put(inputName, new ArrayList<>(bindings));
 		return bindings;
 	}
 

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java

@@ -81,6 +81,7 @@ import static org.mockito.Mockito.when;
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
  * @author Janne Valkealahti
+ * @author Soby Chacko
  */
 public class BindingServiceTests {
 
@@ -163,6 +164,53 @@ public class BindingServiceTests {
 		binderFactory.destroy();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testConsumerBindingWhenMultiplexingIsEnabled() throws Exception {
+		BindingServiceProperties properties = new BindingServiceProperties();
+		Map<String, BindingProperties> bindingProperties = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		props.setDestination("foo,bar");
+
+		ConsumerProperties consumer = properties.getConsumerProperties("input");
+		consumer.setMultiplex(true);
+		props.setConsumer(consumer);
+
+		final String inputChannelName = "input";
+		bindingProperties.put(inputChannelName, props);
+
+		properties.setBindings(bindingProperties);
+
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
+
+		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
+		BindingService service = new BindingService(properties,
+				binderFactory);
+		MessageChannel inputChannel = new DirectChannel();
+
+		Binding<MessageChannel> mockBinding1 = Mockito.mock(Binding.class);
+
+		when(binder.bindConsumer(eq("foo,bar"), isNull(), same(inputChannel),
+				any(ConsumerProperties.class))).thenReturn(mockBinding1);
+
+		Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
+				"input");
+		assertThat(bindings).hasSize(1);
+
+		Iterator<Binding<MessageChannel>> iterator = bindings.iterator();
+		Binding<MessageChannel> binding1 = iterator.next();
+
+		assertThat(binding1).isSameAs(mockBinding1);
+
+		service.unbindConsumers("input");
+
+		verify(binder).bindConsumer(eq("foo,bar"), isNull(), same(inputChannel),
+				any(ConsumerProperties.class));
+		verify(binding1).unbind();
+
+		binderFactory.destroy();
+	}
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	public void testExplicitGroup() throws Exception {