From 61bce5272270b8f4e6b6ff8bed8c4be85450ae1d Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 24 Apr 2018 15:25:14 -0400 Subject: [PATCH] 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 --- .../stream/binder/ConsumerProperties.java | 20 ++++++++ .../cloud/stream/binding/BindingService.java | 34 ++++++++----- .../stream/binding/BindingServiceTests.java | 48 +++++++++++++++++++ 3 files changed, 89 insertions(+), 13 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java index a673aa96d..b21d5f13b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java +++ b/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; + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index 3b202f4a9..3633c74e5 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/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 Collection> bindConsumer(T input, String inputName) { - String bindingTarget = this.bindingServiceProperties - .getBindingDestination(inputName); - String[] bindingTargets = StringUtils - .commaDelimitedListToStringArray(bindingTarget); Collection> bindings = new ArrayList<>(); Binder binder = (Binder) getBinder( inputName, input.getClass()); @@ -106,19 +103,30 @@ public class BindingService { BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties); consumerProperties = extendedConsumerProperties; } + validate(consumerProperties); - for (String target : bindingTargets) { - Binding 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 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>(bindings)); + this.consumerBindings.put(inputName, new ArrayList<>(bindings)); return bindings; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index 15488dbe3..58d12b43d 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/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 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 mockBinding1 = Mockito.mock(Binding.class); + + when(binder.bindConsumer(eq("foo,bar"), isNull(), same(inputChannel), + any(ConsumerProperties.class))).thenReturn(mockBinding1); + + Collection> bindings = service.bindConsumer(inputChannel, + "input"); + assertThat(bindings).hasSize(1); + + Iterator> iterator = bindings.iterator(); + Binding 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 {