SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow binding to multiple input destinations

Browse Source 
More integration tests in AbstractBinderTests
Use embedded Kafka in the tests
This commit is contained in:
Soby Chacko
2016-02-12 15:22:02 -05:00
committed by Ilayaperumal Gopinathan
parent 1216b4b797
commit 76c64e3763
5 changed files with 190 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

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

43
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java
View File

@@ -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<MessageChannel> binder = getBinder();
DirectChannel moduleOutputChannel1 = new DirectChannel();
DirectChannel moduleOutputChannel2 = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Binding<MessageChannel> producerBinding1 = binder.bindProducer("foo.x", moduleOutputChannel1, null);
Binding<MessageChannel> producerBinding2 = binder.bindProducer("foo.y", moduleOutputChannel2, null);
Binding<MessageChannel> consumerBinding1 = binder.bindConsumer("foo.x", "test", moduleInputChannel, null);
Binding<MessageChannel> 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));
}
}

45
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
View File

@@ -106,6 +106,50 @@ public abstract class AbstractBinderTests {
binder.unbind(consumerBinding);
}
@Test
public void testSendAndReceiveMutipleTopics() throws Exception {
Binder<MessageChannel> binder = getBinder();
DirectChannel moduleOutputChannel1 = new DirectChannel();
DirectChannel moduleOutputChannel2 = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Binding<MessageChannel> producerBinding1 = binder.bindProducer("foo.x", moduleOutputChannel1, null);
Binding<MessageChannel> producerBinding2 = binder.bindProducer("foo.y", moduleOutputChannel2, null);
Binding<MessageChannel> consumerBinding1 = binder.bindConsumer("foo.x", "test", moduleInputChannel, null);
Binding<MessageChannel> 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<MessageChannel> binder = getBinder();
@@ -145,6 +189,7 @@ public abstract class AbstractBinderTests {
if (testBinder != null) {
testBinder.cleanup();
}
System.clearProperty("SCS_KAFKA_TEST_EMBEDDED");
}
/**

35
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java
View File

@@ -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<String, Binding<MessageChannel>> producerBindings = new HashMap<>();
private final Map<String, Binding<MessageChannel>> consumerBindings = new HashMap<>();
private final Map<String, List<Binding<MessageChannel>>> consumerBindings = new HashMap<>();
public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties,
BinderFactory<MessageChannel> binderFactory) {
@@ -57,13 +63,22 @@ public class ChannelBindingService {
this.binderFactory = binderFactory;
}
public Binding<MessageChannel> bindConsumer(MessageChannel inputChannel, String inputChannelName) {
public Collection<Binding<MessageChannel>> bindConsumer(MessageChannel inputChannel, String inputChannelName) {
String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(inputChannelName);
String[] channelBindingTargets = StringUtils.commaDelimitedListToStringArray(channelBindingTarget);
List<Binding<MessageChannel>> bindings = new ArrayList<>();
Binder<MessageChannel> binder = getBinderForChannel(inputChannelName);
Binding<MessageChannel> 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<MessageChannel> binding = binder.bindConsumer(target, consumerGroup, inputChannel,
consumerProperties);
bindings.add(binding);
}
this.consumerBindings.put(inputChannelName, bindings);
return bindings;
}
public Binding<MessageChannel> bindProducer(MessageChannel outputChannel, String outputChannelName) {
@@ -77,9 +92,11 @@ public class ChannelBindingService {
public void unbindConsumers(String inputChannelName) {
Binder<MessageChannel> binder = getBinderForChannel(inputChannelName);
Binding<MessageChannel> binding = this.consumerBindings.remove(inputChannelName);
if (binding != null) {
binder.unbind(binding);
List<Binding<MessageChannel>> bindings = this.consumerBindings.remove(inputChannelName);
if (bindings != null && !CollectionUtils.isEmpty(bindings)) {
for (Binding<MessageChannel> binding : bindings) {
binder.unbind(binding);
}
}
else if (log.isWarnEnabled()) {
log.warn("Trying to unbind channel '" + inputChannelName + "', but no binding found.");

87
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java
View File

@@ -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<String, BindingProperties> bindings = new HashMap<>();
Map<String, BindingProperties> 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<MessageChannel> 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<MessageChannel> binding = service.bindConsumer(inputChannel, name);
Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel, inputChannelName);
assertThat(bindings.size(), is(1));
Binding<MessageChannel> 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<String, BindingProperties> bindingProperties = new HashMap<>();
BindingProperties props = new BindingProperties();
props.setDestination("foo,bar");
final String inputChannelName = "input";
bindingProperties.put(inputChannelName, props);
properties.setBindings(bindingProperties);
DefaultBinderFactory<MessageChannel> binderFactory =
new DefaultBinderFactory<>(Collections.singletonMap("mock",
new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}),
new Properties(), true)));
Binder<MessageChannel> binder = binderFactory.getBinder("mock");
ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding1 = Binding.forConsumer("foo", null, Mockito.mock(AbstractEndpoint.class),
inputChannel, null);
Binding<MessageChannel> 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<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel, "input");
assertThat(bindings.size(), is(2));
Iterator<Binding<MessageChannel>> iterator = bindings.iterator();
Binding<MessageChannel> binding1 = iterator.next();
Binding<MessageChannel> 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<String, BindingProperties> bindings = new HashMap<>();
Map<String, BindingProperties> 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<MessageChannel> 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<MessageChannel> binding = service.bindConsumer(inputChannel, name);
Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel, inputChannelName);
assertThat(bindings.size(), is(1));
Binding<MessageChannel> 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<MessageChannel> binder = binderFactory.getBinder("mock");
MessageChannel inputChannel = new DirectChannel();
ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
Binding<MessageChannel> mockBinding = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class),
inputChannel, null);