From 76c64e3763053e04eee8e376534cc132563c47ac Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 12 Feb 2016 15:22:02 -0500 Subject: [PATCH] Allow binding to multiple input destinations More integration tests in AbstractBinderTests Use embedded Kafka in the tests --- .../stream/binder/kafka/KafkaBinderTests.java | 5 +- .../binder/kafka/RawModeKafkaBinderTests.java | 43 +++++++++ .../stream/binder/AbstractBinderTests.java | 45 ++++++++++ .../stream/binding/ChannelBindingService.java | 35 ++++++-- .../binding/ChannelBindingServiceTests.java | 87 +++++++++++++++---- 5 files changed, 190 insertions(+), 25 deletions(-) diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index c3b29242b..70b2a9483 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -70,6 +70,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); + static { + System.setProperty("SCS_KAFKA_TEST_EMBEDDED", "true"); + } + @ClassRule public static KafkaTestSupport kafkaTestSupport = new KafkaTestSupport(); @@ -88,7 +92,6 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { return binder; } - @Before public void init() { String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER"); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java index f4777e824..fa4e5d39a 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java @@ -33,6 +33,7 @@ import org.junit.Ignore; import org.junit.Test; import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.BinderPropertyKeys; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.TestUtils; @@ -273,4 +274,46 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { assertTrue(getBindings(binder).isEmpty()); } + @Test + @Override + public void testSendAndReceiveMutipleTopics() throws Exception { + Binder binder = getBinder(); + + DirectChannel moduleOutputChannel1 = new DirectChannel(); + DirectChannel moduleOutputChannel2 = new DirectChannel(); + + QueueChannel moduleInputChannel = new QueueChannel(); + + Binding producerBinding1 = binder.bindProducer("foo.x", moduleOutputChannel1, null); + Binding producerBinding2 = binder.bindProducer("foo.y", moduleOutputChannel2, null); + + Binding consumerBinding1 = binder.bindConsumer("foo.x", "test", moduleInputChannel, null); + Binding consumerBinding2 = binder.bindConsumer("foo.y", "test", moduleInputChannel, null); + + Message message1 = MessageBuilder.withPayload("foo-x-payload".getBytes()).build(); + Message message2 = MessageBuilder.withPayload("foo-y-payload".getBytes()).build(); + + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel1.send(message1); + Thread.sleep(50); + moduleOutputChannel2.send(message2); + + assertMessageReceive(moduleInputChannel, "foo-x-payload"); + assertMessageReceive(moduleInputChannel, "foo-y-payload"); + + binder.unbind(producerBinding1); + binder.unbind(consumerBinding1); + + binder.unbind(producerBinding2); + binder.unbind(consumerBinding2); + } + + private void assertMessageReceive(QueueChannel moduleInputChannel, String payload) { + Message inbound = receive(moduleInputChannel); + assertNotNull(inbound); + assertEquals(payload, new String((byte[])inbound.getPayload())); + assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)); + } + } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index 6fa615ac3..b9c75f88b 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -106,6 +106,50 @@ public abstract class AbstractBinderTests { binder.unbind(consumerBinding); } + @Test + public void testSendAndReceiveMutipleTopics() throws Exception { + Binder binder = getBinder(); + + DirectChannel moduleOutputChannel1 = new DirectChannel(); + DirectChannel moduleOutputChannel2 = new DirectChannel(); + + QueueChannel moduleInputChannel = new QueueChannel(); + + Binding producerBinding1 = binder.bindProducer("foo.x", moduleOutputChannel1, null); + Binding producerBinding2 = binder.bindProducer("foo.y", moduleOutputChannel2, null); + + Binding consumerBinding1 = binder.bindConsumer("foo.x", "test", moduleInputChannel, null); + Binding consumerBinding2 = binder.bindConsumer("foo.y", "test", moduleInputChannel, null); + + Message message1 = MessageBuilder.withPayload("foo-x-payload").setHeader(MessageHeaders.CONTENT_TYPE, + "foo/bar").build(); + Message message2 = MessageBuilder.withPayload("foo-y-payload").setHeader(MessageHeaders.CONTENT_TYPE, + "foo/bar").build(); + + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel1.send(message1); + Thread.sleep(50); + moduleOutputChannel2.send(message2); + + assertMessageReceive(moduleInputChannel, "foo-x-payload"); + assertMessageReceive(moduleInputChannel, "foo-y-payload"); + + binder.unbind(producerBinding1); + binder.unbind(consumerBinding1); + + binder.unbind(producerBinding2); + binder.unbind(consumerBinding2); + } + + private void assertMessageReceive(QueueChannel moduleInputChannel, String payload) { + Message inbound = receive(moduleInputChannel); + assertNotNull(inbound); + assertEquals(payload, inbound.getPayload()); + assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)); + assertEquals("foo/bar", inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)); + } + @Test public void testSendAndReceiveNoOriginalContentType() throws Exception { Binder binder = getBinder(); @@ -145,6 +189,7 @@ public abstract class AbstractBinderTests { if (testBinder != null) { testBinder.cleanup(); } + System.clearProperty("SCS_KAFKA_TEST_EMBEDDED"); } /** 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 fa448b80b..be40bc76f 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 @@ -16,8 +16,12 @@ package org.springframework.cloud.stream.binding; +import java.util.ArrayList; +import java.util.Collection; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Properties; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -28,6 +32,8 @@ import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.messaging.MessageChannel; +import org.springframework.util.CollectionUtils; +import org.springframework.util.StringUtils; /** * Handles the operations related to channel binding including binding of input/output channels by delegating @@ -49,7 +55,7 @@ public class ChannelBindingService { private final Map> producerBindings = new HashMap<>(); - private final Map> consumerBindings = new HashMap<>(); + private final Map>> consumerBindings = new HashMap<>(); public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties, BinderFactory binderFactory) { @@ -57,13 +63,22 @@ public class ChannelBindingService { this.binderFactory = binderFactory; } - public Binding bindConsumer(MessageChannel inputChannel, String inputChannelName) { + public Collection> bindConsumer(MessageChannel inputChannel, String inputChannelName) { String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(inputChannelName); + String[] channelBindingTargets = StringUtils.commaDelimitedListToStringArray(channelBindingTarget); + List> bindings = new ArrayList<>(); + Binder binder = getBinderForChannel(inputChannelName); - Binding binding = binder.bindConsumer(channelBindingTarget, consumerGroup(inputChannelName), inputChannel, - this.channelBindingServiceProperties.getConsumerProperties(inputChannelName)); - this.consumerBindings.put(inputChannelName, binding); - return binding; + String consumerGroup = consumerGroup(inputChannelName); + Properties consumerProperties = this.channelBindingServiceProperties.getConsumerProperties(inputChannelName); + + for (String target : channelBindingTargets) { + Binding binding = binder.bindConsumer(target, consumerGroup, inputChannel, + consumerProperties); + bindings.add(binding); + } + this.consumerBindings.put(inputChannelName, bindings); + return bindings; } public Binding bindProducer(MessageChannel outputChannel, String outputChannelName) { @@ -77,9 +92,11 @@ public class ChannelBindingService { public void unbindConsumers(String inputChannelName) { Binder binder = getBinderForChannel(inputChannelName); - Binding binding = this.consumerBindings.remove(inputChannelName); - if (binding != null) { - binder.unbind(binding); + List> bindings = this.consumerBindings.remove(inputChannelName); + if (bindings != null && !CollectionUtils.isEmpty(bindings)) { + for (Binding binding : bindings) { + binder.unbind(binding); + } } else if (log.isWarnEnabled()) { log.warn("Trying to unbind channel '" + inputChannelName + "', but no binding found."); 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 ecc2574be..ea5896975 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 @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binding; import static org.hamcrest.CoreMatchers.sameInstance; import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.core.Is.is; import static org.junit.Assert.assertThat; import static org.junit.Assert.fail; import static org.mockito.Matchers.any; @@ -29,8 +30,10 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.Iterator; import java.util.Map; import java.util.Properties; import java.util.concurrent.atomic.AtomicReference; @@ -65,12 +68,12 @@ public class ChannelBindingServiceTests { @Test public void testDefaultGroup() throws Exception { ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); - Map bindings = new HashMap<>(); + Map bindingProperties = new HashMap<>(); BindingProperties props = new BindingProperties(); props.setDestination("foo"); - String name = "foo"; - bindings.put(name, props); - properties.setBindings(bindings); + final String inputChannelName = "input"; + bindingProperties.put(inputChannelName, props); + properties.setBindings(bindingProperties); DefaultBinderFactory binderFactory = new DefaultBinderFactory<>(Collections.singletonMap("mock", new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), @@ -82,24 +85,77 @@ public class ChannelBindingServiceTests { inputChannel, null); when(binder.bindConsumer("foo", null, inputChannel, new Properties())) .thenReturn(mockBinding); - Binding binding = service.bindConsumer(inputChannel, name); + Collection> bindings = service.bindConsumer(inputChannel, inputChannelName); + assertThat(bindings.size(), is(1)); + Binding binding = bindings.iterator().next(); assertThat(binding, sameInstance(mockBinding)); - service.unbindConsumers(name); - verify(binder).bindConsumer(name, props.getGroup(), inputChannel, properties.getConsumerProperties(name)); + service.unbindConsumers(inputChannelName); + verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName)); verify(binder).unbind(binding); binderFactory.destroy(); } + @Test + public void testMultipleConsumerBindings() throws Exception { + ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); + Map bindingProperties = new HashMap<>(); + + BindingProperties props = new BindingProperties(); + props.setDestination("foo,bar"); + final String inputChannelName = "input"; + bindingProperties.put(inputChannelName, props); + + properties.setBindings(bindingProperties); + + DefaultBinderFactory binderFactory = + new DefaultBinderFactory<>(Collections.singletonMap("mock", + new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), + new Properties(), true))); + + Binder binder = binderFactory.getBinder("mock"); + ChannelBindingService service = new ChannelBindingService(properties, binderFactory); + MessageChannel inputChannel = new DirectChannel(); + + Binding mockBinding1 = Binding.forConsumer("foo", null, Mockito.mock(AbstractEndpoint.class), + inputChannel, null); + Binding mockBinding2 = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class), + inputChannel, null); + + when(binder.bindConsumer("foo", null, inputChannel, new Properties())) + .thenReturn(mockBinding1); + when(binder.bindConsumer("bar", null, inputChannel, new Properties())) + .thenReturn(mockBinding2); + + Collection> bindings = service.bindConsumer(inputChannel, "input"); + assertThat(bindings.size(), is(2)); + + Iterator> iterator = bindings.iterator(); + Binding binding1 = iterator.next(); + Binding binding2 = iterator.next(); + + assertThat(binding1, sameInstance(mockBinding1)); + assertThat(binding2, sameInstance(mockBinding2)); + + service.unbindConsumers("input"); + + verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName)); + verify(binder).bindConsumer("bar", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName)); + verify(binder).unbind(binding1); + verify(binder).unbind(binding2); + + binderFactory.destroy(); + } + @Test public void testExplicitGroup() throws Exception { ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); - Map bindings = new HashMap<>(); + Map bindingProperties = new HashMap<>(); BindingProperties props = new BindingProperties(); props.setDestination("foo"); props.setGroup("fooGroup"); - String name = "foo"; - bindings.put(name, props); - properties.setBindings(bindings); + final String inputChannelName = "input"; + bindingProperties.put(inputChannelName, props); + properties.setBindings(bindingProperties); DefaultBinderFactory binderFactory = new DefaultBinderFactory<>(Collections.singletonMap("mock", new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), @@ -111,11 +167,13 @@ public class ChannelBindingServiceTests { inputChannel, null); when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties())) .thenReturn(mockBinding); - Binding binding = service.bindConsumer(inputChannel, name); + Collection> bindings = service.bindConsumer(inputChannel, inputChannelName); + assertThat(bindings.size(), is(1)); + Binding binding = bindings.iterator().next(); assertThat(binding, sameInstance(mockBinding)); - service.unbindConsumers(name); - verify(binder).bindConsumer(name, props.getGroup(), inputChannel, properties.getConsumerProperties(name)); + service.unbindConsumers(inputChannelName); + verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName)); verify(binder).unbind(binding); binderFactory.destroy(); } @@ -131,7 +189,6 @@ public class ChannelBindingServiceTests { Binder binder = binderFactory.getBinder("mock"); MessageChannel inputChannel = new DirectChannel(); - ChannelBindingService service = new ChannelBindingService(properties, binderFactory); Binding mockBinding = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class), inputChannel, null);