Renamed MessageBus to Binder
- Renamed MessageBus to Binder - Parameterized Binder interface - Restored exclusion of spring-xd-codec
This commit is contained in:
David Turanski
committed by
Marius Bogoevici
Marius Bogoevici
parent
11e9f269db
commit
35b74b00b3
@@ -0,0 +1,311 @@
|
||||
/*
|
||||
* Copyright 2013-2014 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.binder;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Properties;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.channel.interceptor.WireTap;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.cloud.stream.binder.Binder.Capability;
|
||||
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
|
||||
import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author David Turanski
|
||||
*/
|
||||
public abstract class AbstractBinderTests {
|
||||
|
||||
protected static final Collection<MediaType> ALL = Collections.singletonList(MediaType.ALL);
|
||||
|
||||
protected AbstractTestBinder<?> testBinder;
|
||||
|
||||
@Test
|
||||
public void testClean() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
binder.bindProducer("foo.0", new DirectChannel(), null);
|
||||
binder.bindConsumer("foo.0", new DirectChannel(), null);
|
||||
binder.bindProducer("foo.1", new DirectChannel(), null);
|
||||
binder.bindConsumer("foo.1", new DirectChannel(), null);
|
||||
binder.bindProducer("foo.2", new DirectChannel(), null);
|
||||
Collection<?> bindings = getBindings(binder);
|
||||
assertEquals(5, bindings.size());
|
||||
binder.unbindProducers("foo.0");
|
||||
assertEquals(4, bindings.size());
|
||||
binder.unbindConsumers("foo.0");
|
||||
binder.unbindProducers("foo.1");
|
||||
assertEquals(2, bindings.size());
|
||||
binder.unbindConsumers("foo.1");
|
||||
binder.unbindProducers("foo.2");
|
||||
assertTrue(bindings.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendAndReceive() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
binder.bindProducer("foo.0", moduleOutputChannel, null);
|
||||
binder.bindConsumer("foo.0", moduleInputChannel, null);
|
||||
Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE,
|
||||
"foo/bar").build();
|
||||
// Let the consumer actually bind to the producer before sending a msg
|
||||
binderBindUnbindLatency();
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = moduleInputChannel.receive(5000);
|
||||
assertNotNull(inbound);
|
||||
assertEquals("foo", inbound.getPayload());
|
||||
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
assertEquals("foo/bar", inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
|
||||
binder.unbindProducers("foo.0");
|
||||
binder.unbindConsumers("foo.0");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendAndReceiveNoOriginalContentType() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
binder.bindProducer("bar.0", moduleOutputChannel, null);
|
||||
binder.bindConsumer("bar.0", moduleInputChannel, null);
|
||||
binderBindUnbindLatency();
|
||||
|
||||
Message<?> message = MessageBuilder.withPayload("foo").build();
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = moduleInputChannel.receive(5000);
|
||||
assertNotNull(inbound);
|
||||
assertEquals("foo", inbound.getPayload());
|
||||
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
assertNull(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
|
||||
binder.unbindProducers("bar.0");
|
||||
binder.unbindConsumers("bar.0");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendAndReceivePubSub() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
// Test pub/sub by emulating how StreamPlugin handles taps
|
||||
DirectChannel tapChannel = new DirectChannel();
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
QueueChannel module2InputChannel = new QueueChannel();
|
||||
QueueChannel module3InputChannel = new QueueChannel();
|
||||
binder.bindProducer("baz.0", moduleOutputChannel, null);
|
||||
binder.bindConsumer("baz.0", moduleInputChannel, null);
|
||||
moduleOutputChannel.addInterceptor(new WireTap(tapChannel));
|
||||
binder.bindPubSubProducer("tap:baz.http", tapChannel, null);
|
||||
// A new module is using the tap as an input channel
|
||||
String fooTapName = binder.isCapable(Capability.DURABLE_PUBSUB) ? "foo.tap:baz.http" : "tap:baz.http";
|
||||
binder.bindPubSubConsumer(fooTapName, module2InputChannel, null);
|
||||
// Another new module is using tap as an input channel
|
||||
String barTapName = binder.isCapable(Capability.DURABLE_PUBSUB) ? "bar.tap:baz.http" : "tap:baz.http";
|
||||
binder.bindPubSubConsumer(barTapName, module3InputChannel, null);
|
||||
Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE,
|
||||
"foo/bar").build();
|
||||
boolean success = false;
|
||||
boolean retried = false;
|
||||
while (!success) {
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = moduleInputChannel.receive(5000);
|
||||
assertNotNull(inbound);
|
||||
assertEquals("foo", inbound.getPayload());
|
||||
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
assertEquals("foo/bar", inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
|
||||
Message<?> tapped1 = module2InputChannel.receive(5000);
|
||||
Message<?> tapped2 = module3InputChannel.receive(5000);
|
||||
if (tapped1 == null || tapped2 == null) {
|
||||
// listener may not have started
|
||||
assertFalse("Failed to receive tap after retry", retried);
|
||||
retried = true;
|
||||
continue;
|
||||
}
|
||||
success = true;
|
||||
assertEquals("foo", tapped1.getPayload());
|
||||
assertNull(tapped1.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
assertEquals("foo/bar", tapped1.getHeaders().get(MessageHeaders.CONTENT_TYPE));
|
||||
assertEquals("foo", tapped2.getPayload());
|
||||
assertNull(tapped2.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
assertEquals("foo/bar", tapped2.getHeaders().get(MessageHeaders.CONTENT_TYPE));
|
||||
}
|
||||
// delete one tap stream is deleted
|
||||
binder.unbindConsumer(barTapName, module3InputChannel);
|
||||
Message<?> message2 = MessageBuilder.withPayload("bar").setHeader(MessageHeaders.CONTENT_TYPE,
|
||||
"foo/bar").build();
|
||||
moduleOutputChannel.send(message2);
|
||||
|
||||
// other tap still receives messages
|
||||
Message<?> tapped = module2InputChannel.receive(5000);
|
||||
assertNotNull(tapped);
|
||||
|
||||
// Removed tap does not
|
||||
assertNull(module3InputChannel.receive(1000));
|
||||
|
||||
// when other tap stream is deleted
|
||||
binder.unbindConsumer(fooTapName, module2InputChannel);
|
||||
// Clean up as StreamPlugin would
|
||||
binder.unbindConsumer("baz.0", moduleInputChannel);
|
||||
binder.unbindProducer("baz.0", moduleOutputChannel);
|
||||
binder.unbindProducers("tap:baz.http");
|
||||
assertTrue(getBindings(binder).isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void createInboundPubSubBeforeOutboundPubSub() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
// Test pub/sub by emulating how StreamPlugin handles taps
|
||||
DirectChannel tapChannel = new DirectChannel();
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
QueueChannel module2InputChannel = new QueueChannel();
|
||||
QueueChannel module3InputChannel = new QueueChannel();
|
||||
// Create the tap first
|
||||
String fooTapName = binder.isCapable(Capability.DURABLE_PUBSUB) ? "foo.tap:baz.http" : "tap:baz.http";
|
||||
binder.bindPubSubConsumer(fooTapName, module2InputChannel, null);
|
||||
|
||||
// Then create the stream
|
||||
binder.bindProducer("baz.0", moduleOutputChannel, null);
|
||||
binder.bindConsumer("baz.0", moduleInputChannel, null);
|
||||
moduleOutputChannel.addInterceptor(new WireTap(tapChannel));
|
||||
binder.bindPubSubProducer("tap:baz.http", tapChannel, null);
|
||||
|
||||
// Another new module is using tap as an input channel
|
||||
String barTapName = binder.isCapable(Capability.DURABLE_PUBSUB) ? "bar.tap:baz.http" : "tap:baz.http";
|
||||
binder.bindPubSubConsumer(barTapName, module3InputChannel, null);
|
||||
Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE,
|
||||
"foo/bar").build();
|
||||
boolean success = false;
|
||||
boolean retried = false;
|
||||
while (!success) {
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = moduleInputChannel.receive(5000);
|
||||
assertNotNull(inbound);
|
||||
assertEquals("foo", inbound.getPayload());
|
||||
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
assertEquals("foo/bar", inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
|
||||
Message<?> tapped1 = module2InputChannel.receive(5000);
|
||||
Message<?> tapped2 = module3InputChannel.receive(5000);
|
||||
if (tapped1 == null || tapped2 == null) {
|
||||
// listener may not have started
|
||||
assertFalse("Failed to receive tap after retry", retried);
|
||||
retried = true;
|
||||
continue;
|
||||
}
|
||||
success = true;
|
||||
assertEquals("foo", tapped1.getPayload());
|
||||
assertNull(tapped1.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
assertEquals("foo/bar", tapped1.getHeaders().get(MessageHeaders.CONTENT_TYPE));
|
||||
assertEquals("foo", tapped2.getPayload());
|
||||
assertNull(tapped2.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
assertEquals("foo/bar", tapped2.getHeaders().get(MessageHeaders.CONTENT_TYPE));
|
||||
}
|
||||
// delete one tap stream is deleted
|
||||
binder.unbindConsumer(barTapName, module3InputChannel);
|
||||
Message<?> message2 = MessageBuilder.withPayload("bar").setHeader(MessageHeaders.CONTENT_TYPE,
|
||||
"foo/bar").build();
|
||||
moduleOutputChannel.send(message2);
|
||||
|
||||
// other tap still receives messages
|
||||
Message<?> tapped = module2InputChannel.receive(5000);
|
||||
assertNotNull(tapped);
|
||||
|
||||
// Removed tap does not
|
||||
assertNull(module3InputChannel.receive(1000));
|
||||
|
||||
// when other tap stream is deleted
|
||||
binder.unbindConsumer(fooTapName, module2InputChannel);
|
||||
// Clean up as StreamPlugin would
|
||||
binder.unbindConsumer("baz.0", moduleInputChannel);
|
||||
binder.unbindProducer("baz.0", moduleOutputChannel);
|
||||
binder.unbindProducers("tap:baz.http");
|
||||
assertTrue(getBindings(binder).isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBadDynamic() throws Exception {
|
||||
Properties properties = new Properties();
|
||||
properties.setProperty(BinderProperties.PARTITION_KEY_EXPRESSION, "'foo'");
|
||||
Binder binder = getBinder();
|
||||
try {
|
||||
binder.bindDynamicProducer("queue:foo", properties);
|
||||
fail("Exception expected");
|
||||
}
|
||||
catch (BinderException mbe) {
|
||||
Assert.assertEquals("Failed to bind dynamic channel 'queue:foo' with properties " +
|
||||
"{partitionKeyExpression='foo'}",
|
||||
mbe.getMessage());
|
||||
if (binder instanceof AbstractTestBinder) {
|
||||
binder = ((AbstractTestBinder) binder).getCoreBinder();
|
||||
}
|
||||
assertFalse(((MessageChannelBinderSupport) binder).getApplicationContext().containsBean("queue:foo"));
|
||||
}
|
||||
}
|
||||
|
||||
protected Collection<?> getBindings(Binder testBinder) {
|
||||
if (testBinder instanceof AbstractTestBinder) {
|
||||
return getBindingsFromBinder(((AbstractTestBinder) testBinder).getCoreBinder());
|
||||
}
|
||||
return Collections.EMPTY_LIST;
|
||||
}
|
||||
|
||||
protected Collection<?> getBindingsFromBinder(Binder binder) {
|
||||
DirectFieldAccessor accessor = new DirectFieldAccessor(binder);
|
||||
return (List<?>) accessor.getPropertyValue("bindings");
|
||||
}
|
||||
|
||||
@SuppressWarnings({"unchecked", "rawtypes"})
|
||||
protected MultiTypeCodec<Object> getCodec() {
|
||||
return new PojoCodec();
|
||||
}
|
||||
|
||||
protected abstract Binder getBinder() throws Exception;
|
||||
|
||||
@After
|
||||
public void cleanup() {
|
||||
if (testBinder != null) {
|
||||
testBinder.cleanup();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* If appropriate, let the binder middleware settle down a bit while binding/unbinding actually happens.
|
||||
*/
|
||||
protected void binderBindUnbindLatency() throws InterruptedException {
|
||||
// default none
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,134 @@
|
||||
/*
|
||||
* Copyright 2013 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.binder;
|
||||
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.junit.Assume;
|
||||
import org.junit.Rule;
|
||||
import org.junit.rules.TestRule;
|
||||
import org.junit.runner.Description;
|
||||
import org.junit.runners.model.Statement;
|
||||
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
|
||||
/**
|
||||
* Abstract base class for JUnit {@link Rule}s that detect the presence of some external resource. If the resource is
|
||||
* indeed present, it will be available during the test lifecycle through {@link #getResource()}. If it is not, tests
|
||||
* will either fail or be skipped, depending on the value of system property {@value #XD_EXTERNAL_SERVERS_REQUIRED}.
|
||||
*
|
||||
* @author Eric Bottard
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public abstract class AbstractExternalResourceTestSupport<R> implements TestRule {
|
||||
|
||||
public static final String XD_EXTERNAL_SERVERS_REQUIRED = "XD_EXTERNAL_SERVERS_REQUIRED";
|
||||
|
||||
protected R resource;
|
||||
|
||||
private String resourceDescription;
|
||||
|
||||
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
|
||||
|
||||
protected AbstractExternalResourceTestSupport(String resourceDescription) {
|
||||
Assert.hasText(resourceDescription, "resourceDescription is required");
|
||||
this.resourceDescription = resourceDescription;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Statement apply(final Statement base, Description description) {
|
||||
try {
|
||||
obtainResource();
|
||||
}
|
||||
catch (Exception e) {
|
||||
maybeCleanup();
|
||||
|
||||
return failOrSkip(e);
|
||||
}
|
||||
|
||||
return new Statement() {
|
||||
|
||||
@Override
|
||||
public void evaluate() throws Throwable {
|
||||
try {
|
||||
base.evaluate();
|
||||
}
|
||||
finally {
|
||||
try {
|
||||
cleanupResource();
|
||||
}
|
||||
catch (Exception ignored) {
|
||||
logger.warn("Exception while trying to cleanup proper resource", ignored);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
private Statement failOrSkip(final Exception e) {
|
||||
String serversRequired = System.getenv(XD_EXTERNAL_SERVERS_REQUIRED);
|
||||
if ("true".equalsIgnoreCase(serversRequired)) {
|
||||
logger.error(resourceDescription + " IS REQUIRED BUT NOT AVAILABLE", e);
|
||||
fail(resourceDescription + " IS NOT AVAILABLE");
|
||||
// Never reached, here to satisfy method signature
|
||||
return null;
|
||||
}
|
||||
else {
|
||||
logger.error(resourceDescription + " IS NOT AVAILABLE, SKIPPING TESTS", e);
|
||||
return new Statement() {
|
||||
|
||||
@Override
|
||||
public void evaluate() throws Throwable {
|
||||
Assume.assumeTrue("Skipping test due to " + resourceDescription + " not being available " + e, false);
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
private void maybeCleanup() {
|
||||
if (resource != null) {
|
||||
try {
|
||||
cleanupResource();
|
||||
}
|
||||
catch (Exception ignored) {
|
||||
logger.warn("Exception while trying to cleanup failed resource", ignored);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public R getResource() {
|
||||
return resource;
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform cleanup of the {@link #resource} field, which is guaranteed to be non null.
|
||||
*
|
||||
* @throws Exception any exception thrown by this method will be logged and swallowed
|
||||
*/
|
||||
protected abstract void cleanupResource() throws Exception;
|
||||
|
||||
/**
|
||||
* Try to obtain and validate a resource. Implementors should either set the {@link #resource} field with a valid
|
||||
* resource and return normally, or throw an exception.
|
||||
*/
|
||||
protected abstract void obtainResource() throws Exception;
|
||||
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
/*
|
||||
* Copyright 2014-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.binder;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.Properties;
|
||||
import java.util.Set;
|
||||
|
||||
import org.springframework.expression.spel.support.StandardEvaluationContext;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
|
||||
|
||||
/**
|
||||
* Abstract class that adds test support for {@link Binder}.
|
||||
*
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public abstract class AbstractTestBinder<C extends MessageChannelBinderSupport> implements Binder<MessageChannel> {
|
||||
|
||||
protected Set<String> queues = new HashSet<String>();
|
||||
|
||||
protected Set<String> topics = new HashSet<String>();
|
||||
|
||||
private C binder;
|
||||
|
||||
public void setBinder(C binder) {
|
||||
binder.setIntegrationEvaluationContext(new StandardEvaluationContext());
|
||||
try {
|
||||
binder.afterPropertiesSet();
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new RuntimeException("Failed to initialize binder", e);
|
||||
}
|
||||
this.binder = binder;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindConsumer(String name, MessageChannel moduleInputChannel, Properties properties) {
|
||||
binder.bindConsumer(name, moduleInputChannel, properties);
|
||||
queues.add(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindPubSubConsumer(String name, MessageChannel inputChannel, Properties properties) {
|
||||
binder.bindPubSubConsumer(name, inputChannel, properties);
|
||||
addTopic(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindProducer(String name, MessageChannel moduleOutputChannel, Properties properties) {
|
||||
binder.bindProducer(name, moduleOutputChannel, properties);
|
||||
queues.add(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindPubSubProducer(String name, MessageChannel outputChannel, Properties properties) {
|
||||
binder.bindPubSubProducer(name, outputChannel, properties);
|
||||
addTopic(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindRequestor(String name, MessageChannel requests, MessageChannel replies,
|
||||
Properties properties) {
|
||||
binder.bindRequestor(name, requests, replies, properties);
|
||||
queues.add(name + ".requests");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindReplier(String name, MessageChannel requests, MessageChannel replies,
|
||||
Properties properties) {
|
||||
binder.bindReplier(name, requests, replies, properties);
|
||||
queues.add(name + ".requests");
|
||||
}
|
||||
|
||||
private void addTopic(String topicName) {
|
||||
topics.add("topic." + topicName);
|
||||
}
|
||||
|
||||
public C getCoreBinder() {
|
||||
return binder;
|
||||
}
|
||||
|
||||
public abstract void cleanup();
|
||||
|
||||
@Override
|
||||
public void unbindConsumers(String name) {
|
||||
binder.unbindConsumers(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbindProducers(String name) {
|
||||
binder.unbindProducers(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbindConsumer(String name, MessageChannel channel) {
|
||||
binder.unbindConsumer(name, channel);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbindProducer(String name, MessageChannel channel) {
|
||||
binder.unbindProducer(name, channel);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MessageChannel bindDynamicProducer(String name, Properties properties) {
|
||||
this.queues.add(name);
|
||||
return this.binder.bindDynamicProducer(name, properties);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MessageChannel bindDynamicPubSubProducer(String name, Properties properties) {
|
||||
this.topics.add(name);
|
||||
return this.binder.bindDynamicPubSubProducer(name, properties);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isCapable(Capability capability) {
|
||||
return this.binder.isCapable(capability);
|
||||
}
|
||||
|
||||
public Binder getBinder() {
|
||||
return this.binder;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
/*
|
||||
* Copyright 2014 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.binder;
|
||||
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.context.support.AbstractApplicationContext;
|
||||
import org.springframework.integration.support.DefaultMessageBuilderFactory;
|
||||
import org.springframework.integration.support.MessageBuilderFactory;
|
||||
import org.springframework.integration.support.utils.IntegrationUtils;
|
||||
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public class BinderTestUtils {
|
||||
|
||||
private static final MessageBuilderFactory mbf = new DefaultMessageBuilderFactory();
|
||||
|
||||
public static final AbstractApplicationContext MOCK_AC = mock(AbstractApplicationContext.class);
|
||||
|
||||
public static final ConfigurableListableBeanFactory MOCK_BF = mock(ConfigurableListableBeanFactory.class);
|
||||
|
||||
static {
|
||||
when(MOCK_BF.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
|
||||
MessageBuilderFactory.class)).thenReturn(mbf);
|
||||
when(MOCK_AC.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
|
||||
MessageBuilderFactory.class)).thenReturn(mbf);
|
||||
when(MOCK_AC.getBeanFactory()).thenReturn(MOCK_BF);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
/*
|
||||
* Copyright 2014 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.binder;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertSame;
|
||||
|
||||
import java.util.Properties;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
|
||||
/**
|
||||
* Tests for binders that use an external broker.
|
||||
*
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public abstract class BrokerBinderTests extends
|
||||
AbstractBinderTests {
|
||||
|
||||
@Test
|
||||
public void testDirectBinding() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
Properties properties = new Properties();
|
||||
properties.setProperty(BinderProperties.DIRECT_BINDING_ALLOWED, "true");
|
||||
|
||||
DirectChannel moduleInputChannel = new DirectChannel();
|
||||
moduleInputChannel.setBeanName("direct.input");
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
moduleOutputChannel.setBeanName("direct.output");
|
||||
binder.bindConsumer("direct.0", moduleInputChannel, null);
|
||||
binder.bindProducer("direct.0", moduleOutputChannel, properties);
|
||||
|
||||
final AtomicReference<Thread> caller = new AtomicReference<Thread>();
|
||||
final AtomicInteger count = new AtomicInteger();
|
||||
moduleInputChannel.subscribe(new MessageHandler() {
|
||||
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
caller.set(Thread.currentThread());
|
||||
count.incrementAndGet();
|
||||
}
|
||||
});
|
||||
|
||||
moduleOutputChannel.send(new GenericMessage<String>("foo"));
|
||||
moduleOutputChannel.send(new GenericMessage<String>("foo"));
|
||||
|
||||
assertNotNull(caller.get());
|
||||
assertSame(Thread.currentThread(), caller.get());
|
||||
assertEquals(2, count.get());
|
||||
assertNull(spyOn("direct.0").receive(true));
|
||||
|
||||
// Remove direct binding and bind producer to the binder
|
||||
binder.unbindConsumers("direct.0");
|
||||
binderBindUnbindLatency();
|
||||
|
||||
Spy spy = spyOn("direct.0");
|
||||
count.set(0);
|
||||
moduleOutputChannel.send(new GenericMessage<String>("bar"));
|
||||
moduleOutputChannel.send(new GenericMessage<String>("baz"));
|
||||
Object bar = spy.receive(false);
|
||||
assertEquals("bar", bar);
|
||||
Object baz = spy.receive(false);
|
||||
assertEquals("baz", baz);
|
||||
assertEquals(0, count.get());
|
||||
|
||||
// Unbind producer from binder and bind directly again
|
||||
caller.set(null);
|
||||
binder.bindConsumer("direct.0", moduleInputChannel, null);
|
||||
moduleOutputChannel.send(new GenericMessage<String>("foo"));
|
||||
moduleOutputChannel.send(new GenericMessage<String>("foo"));
|
||||
assertNotNull(caller.get());
|
||||
assertSame(Thread.currentThread(), caller.get());
|
||||
assertEquals(2, count.get());
|
||||
assertNull(spy.receive(true));
|
||||
|
||||
binder.unbindProducers("direct.0");
|
||||
binder.unbindConsumers("direct.0");
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new spy on the given 'queue'. This allows de-correlating the creation of
|
||||
* the 'connection' from its actual usage, which may be needed by some implementations to
|
||||
* see messages sent after connection creation.
|
||||
*/
|
||||
public abstract Spy spyOn(final String name);
|
||||
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,272 @@
|
||||
/*
|
||||
* Copyright 2014 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.binder;
|
||||
|
||||
import static org.hamcrest.Matchers.allOf;
|
||||
import static org.hamcrest.Matchers.containsInAnyOrder;
|
||||
import static org.hamcrest.Matchers.containsString;
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.hamcrest.Matchers.hasProperty;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Properties;
|
||||
|
||||
import org.hamcrest.CustomMatcher;
|
||||
import org.hamcrest.Matcher;
|
||||
import org.hamcrest.Matchers;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.IntegrationMessageHeaderAccessor;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.endpoint.AbstractEndpoint;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
|
||||
/**
|
||||
* Tests for binders that support partitioning.
|
||||
*
|
||||
* @author Gary Russell
|
||||
*/
|
||||
abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
|
||||
|
||||
@Test
|
||||
public void testBadProperties() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
Properties properties = new Properties();
|
||||
properties.put("foo", "bar");
|
||||
properties.put("baz", "qux");
|
||||
|
||||
DirectChannel output = new DirectChannel();
|
||||
try {
|
||||
binder.bindProducer("badprops.0", output, properties);
|
||||
}
|
||||
catch (IllegalArgumentException e) {
|
||||
assertThat(e.getMessage(), allOf(Matchers.containsString(getClassUnderTestName()
|
||||
+ " does not support producer "),
|
||||
containsString("foo"),
|
||||
containsString("baz"),
|
||||
containsString(" for badprops.0.")));
|
||||
}
|
||||
|
||||
properties.remove("baz");
|
||||
try {
|
||||
binder.bindConsumer("badprops.0", output, properties);
|
||||
}
|
||||
catch (IllegalArgumentException e) {
|
||||
assertThat(e.getMessage(), equalTo(getClassUnderTestName()
|
||||
+ " does not support consumer property: foo for badprops.0."));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPartitionedModuleSpEL() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
Properties properties = new Properties();
|
||||
properties.put("partitionKeyExpression", "payload");
|
||||
properties.put("partitionSelectorExpression", "hashCode()");
|
||||
properties.put(BinderProperties.NEXT_MODULE_COUNT, "3");
|
||||
properties.put(BinderProperties.NEXT_MODULE_CONCURRENCY, "2");
|
||||
|
||||
DirectChannel output = new DirectChannel();
|
||||
output.setBeanName("test.output");
|
||||
binder.bindProducer("part.0", output, properties);
|
||||
@SuppressWarnings("unchecked")
|
||||
List<Binding> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
|
||||
assertEquals(1, bindings.size());
|
||||
try {
|
||||
AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
|
||||
assertThat(getEndpointRouting(endpoint), containsString("part.0-' + headers['partition']"));
|
||||
}
|
||||
catch (UnsupportedOperationException ignored) {
|
||||
|
||||
}
|
||||
|
||||
properties.clear();
|
||||
properties.put("concurrency", "2");
|
||||
properties.put("partitionIndex", "0");
|
||||
properties.put("count","3");
|
||||
QueueChannel input0 = new QueueChannel();
|
||||
input0.setBeanName("test.input0S");
|
||||
binder.bindConsumer("part.0", input0, properties);
|
||||
properties.put("partitionIndex", "1");
|
||||
QueueChannel input1 = new QueueChannel();
|
||||
input1.setBeanName("test.input1S");
|
||||
binder.bindConsumer("part.0", input1, properties);
|
||||
properties.put("partitionIndex", "2");
|
||||
QueueChannel input2 = new QueueChannel();
|
||||
input2.setBeanName("test.input2S");
|
||||
binder.bindConsumer("part.0", input2, properties);
|
||||
|
||||
Message<Integer> message2 = MessageBuilder.withPayload(2)
|
||||
.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo")
|
||||
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)
|
||||
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43)
|
||||
.setHeader(BinderHeaders.BINDER_REPLY_CHANNEL, "bar")
|
||||
.build();
|
||||
output.send(message2);
|
||||
output.send(new GenericMessage<Integer>(1));
|
||||
output.send(new GenericMessage<Integer>(0));
|
||||
|
||||
Message<?> receive0 = input0.receive(1000);
|
||||
assertNotNull(receive0);
|
||||
Message<?> receive1 = input1.receive(1000);
|
||||
assertNotNull(receive1);
|
||||
Message<?> receive2 = input2.receive(1000);
|
||||
assertNotNull(receive2);
|
||||
|
||||
Matcher<Message<?>> fooMatcher = new CustomMatcher<Message<?>>("the message with 'foo' as its correlationId") {
|
||||
|
||||
@Override
|
||||
public boolean matches(Object item) {
|
||||
IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor((Message<?>) item);
|
||||
boolean result = "foo".equals(accessor.getCorrelationId()) &&
|
||||
42 == accessor.getSequenceNumber() &&
|
||||
43 == accessor.getSequenceSize() &&
|
||||
"bar".equals(accessor.getHeader(BinderHeaders.BINDER_REPLY_CHANNEL));
|
||||
return result;
|
||||
}
|
||||
};
|
||||
|
||||
if (usesExplicitRouting()) {
|
||||
assertEquals(0, receive0.getPayload());
|
||||
assertEquals(1, receive1.getPayload());
|
||||
assertEquals(2, receive2.getPayload());
|
||||
|
||||
assertThat(receive2, fooMatcher);
|
||||
|
||||
}
|
||||
else {
|
||||
|
||||
assertThat(Arrays.asList(
|
||||
(Integer) receive0.getPayload(),
|
||||
(Integer) receive1.getPayload(),
|
||||
(Integer) receive2.getPayload()),
|
||||
containsInAnyOrder(0, 1, 2));
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
Matcher<Iterable<? extends Message<?>>> containsOur3Messages = containsInAnyOrder(
|
||||
fooMatcher,
|
||||
hasProperty("payload", equalTo(0)),
|
||||
hasProperty("payload", equalTo(1))
|
||||
);
|
||||
assertThat(
|
||||
Arrays.asList(receive0, receive1, receive2),
|
||||
containsOur3Messages);
|
||||
|
||||
}
|
||||
|
||||
binder.unbindConsumers("part.0");
|
||||
binder.unbindProducers("part.0");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPartitionedModuleJava() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
Properties properties = new Properties();
|
||||
properties.put("partitionKeyExtractorClass", "org.springframework.cloud.stream.binder.PartitionTestSupport");
|
||||
properties.put("partitionSelectorClass", "org.springframework.cloud.stream.binder.PartitionTestSupport");
|
||||
properties.put(BinderProperties.NEXT_MODULE_COUNT, "3");
|
||||
properties.put(BinderProperties.NEXT_MODULE_CONCURRENCY, "2");
|
||||
|
||||
DirectChannel output = new DirectChannel();
|
||||
output.setBeanName("test.output");
|
||||
binder.bindProducer("partJ.0", output, properties);
|
||||
@SuppressWarnings("unchecked")
|
||||
List<Binding> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
|
||||
assertEquals(1, bindings.size());
|
||||
if (usesExplicitRouting()) {
|
||||
AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
|
||||
assertThat(getEndpointRouting(endpoint), containsString("partJ.0-' + headers['partition']"));
|
||||
}
|
||||
|
||||
properties.clear();
|
||||
properties.put("concurrency", "2");
|
||||
properties.put("count","3");
|
||||
properties.put("partitionIndex", "0");
|
||||
QueueChannel input0 = new QueueChannel();
|
||||
input0.setBeanName("test.input0J");
|
||||
binder.bindConsumer("partJ.0", input0, properties);
|
||||
properties.put("partitionIndex", "1");
|
||||
QueueChannel input1 = new QueueChannel();
|
||||
input1.setBeanName("test.input1J");
|
||||
binder.bindConsumer("partJ.0", input1, properties);
|
||||
properties.put("partitionIndex", "2");
|
||||
QueueChannel input2 = new QueueChannel();
|
||||
input2.setBeanName("test.input2J");
|
||||
binder.bindConsumer("partJ.0", input2, properties);
|
||||
|
||||
output.send(new GenericMessage<Integer>(2));
|
||||
output.send(new GenericMessage<Integer>(1));
|
||||
output.send(new GenericMessage<Integer>(0));
|
||||
|
||||
Message<?> receive0 = input0.receive(1000);
|
||||
assertNotNull(receive0);
|
||||
Message<?> receive1 = input1.receive(1000);
|
||||
assertNotNull(receive1);
|
||||
Message<?> receive2 = input2.receive(1000);
|
||||
assertNotNull(receive2);
|
||||
|
||||
if (usesExplicitRouting()) {
|
||||
assertEquals(0, receive0.getPayload());
|
||||
assertEquals(1, receive1.getPayload());
|
||||
assertEquals(2, receive2.getPayload());
|
||||
}
|
||||
else {
|
||||
|
||||
assertThat(Arrays.asList(
|
||||
(Integer) receive0.getPayload(),
|
||||
(Integer) receive1.getPayload(),
|
||||
(Integer) receive2.getPayload()),
|
||||
containsInAnyOrder(0, 1, 2));
|
||||
}
|
||||
|
||||
binder.unbindConsumers("partJ.0");
|
||||
binder.unbindProducers("partJ.0");
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementations should return whether the binder under test uses "explicit" routing (e.g. Rabbit)
|
||||
* whereby XD is responsible for assigning a partition and knows which exact consumer will receive the
|
||||
* message (i.e. honor "partitionIndex") or "implicit" routing (e.g. Kafka) whereby the only guarantee
|
||||
* is that messages will be spread, but we don't control exactly which consumer gets which message.
|
||||
*/
|
||||
protected abstract boolean usesExplicitRouting();
|
||||
|
||||
/**
|
||||
* For implementations that rely on explicit routing, return the routing expression.
|
||||
*/
|
||||
protected String getEndpointRouting(AbstractEndpoint endpoint) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
/**
|
||||
* For implementations that rely on explicit routing, return the routing expression.
|
||||
*/
|
||||
protected String getPubSubEndpointRouting(AbstractEndpoint endpoint) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
protected abstract String getClassUnderTestName();
|
||||
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
/*
|
||||
* Copyright 2014 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.binder;
|
||||
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public class PartitionTestSupport implements PartitionKeyExtractorStrategy, PartitionSelectorStrategy {
|
||||
|
||||
@Override
|
||||
public int selectPartition(Object key, int divisor) {
|
||||
return key.hashCode() % divisor;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object extractKey(Message<?> message) {
|
||||
return message.getPayload();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
package org.springframework.cloud.stream.binder;
|
||||
|
||||
/**
|
||||
* Represents an out-of-band connection to the underlying middleware,
|
||||
* so that tests can check that some messages actually do (or do not)
|
||||
* transit through it.
|
||||
*
|
||||
* @author Eric Bottard
|
||||
*/
|
||||
public interface Spy {
|
||||
|
||||
public Object receive(boolean expectNull) throws Exception;
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
/*
|
||||
* 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.binder;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Copy of class in org.springframework.amqp.utils.test to avoid dependency on spring-amqp
|
||||
*/
|
||||
public class TestUtils {
|
||||
|
||||
/**
|
||||
* Uses nested {@link DirectFieldAccessor}s to obtain a property using dotted notation to traverse fields; e.g.
|
||||
* "foo.bar.baz" will obtain a reference to the baz field of the bar field of foo. Adopted from Spring Integration.
|
||||
* @param root The object.
|
||||
* @param propertyPath The path.
|
||||
* @return The field.
|
||||
*/
|
||||
public static Object getPropertyValue(Object root, String propertyPath) {
|
||||
Object value = null;
|
||||
DirectFieldAccessor accessor = new DirectFieldAccessor(root);
|
||||
String[] tokens = propertyPath.split("\\.");
|
||||
for (int i = 0; i < tokens.length; i++) {
|
||||
value = accessor.getPropertyValue(tokens[i]);
|
||||
if (value != null) {
|
||||
accessor = new DirectFieldAccessor(value);
|
||||
}
|
||||
else if (i == tokens.length - 1) {
|
||||
return null;
|
||||
}
|
||||
else {
|
||||
throw new IllegalArgumentException("intermediate property '" + tokens[i] + "' is null");
|
||||
}
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> T getPropertyValue(Object root, String propertyPath, Class<T> type) {
|
||||
Object value = getPropertyValue(root, propertyPath);
|
||||
if (value != null) {
|
||||
Assert.isAssignable(type, value.getClass());
|
||||
}
|
||||
return (T) value;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
<configuration>
|
||||
<appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>%d{ISO8601} %5p %t %c{2}:%L - %m%n</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
<logger name="org.springframework.web.client.RestTemplate" level="ERROR"/>
|
||||
<logger name="org.apache.hadoop.util.NativeCodeLoader" level="ERROR"/>
|
||||
<root level="WARN">
|
||||
<appender-ref ref="stdout"/>
|
||||
</root>
|
||||
</configuration>
|
||||
@@ -0,0 +1,302 @@
|
||||
/*
|
||||
* Copyright 2013 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.binder;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collections;
|
||||
import java.util.Properties;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.converter.ContentTypeResolver;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.util.MimeType;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport.JavaClassMimeTypeConversion;
|
||||
import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
|
||||
import org.springframework.xd.tuple.DefaultTuple;
|
||||
import org.springframework.xd.tuple.Tuple;
|
||||
import org.springframework.xd.tuple.TupleBuilder;
|
||||
import org.springframework.xd.tuple.serializer.kryo.TupleKryoRegistrar;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertSame;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @author David Turanski
|
||||
*/
|
||||
public class MessageChannelBinderSupportTests {
|
||||
|
||||
private ContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver();
|
||||
|
||||
private final TestMessageChannelBinder binder = new TestMessageChannelBinder();
|
||||
|
||||
@SuppressWarnings({"unchecked", "rawtypes"})
|
||||
@Before
|
||||
public void setUp() {
|
||||
binder.setCodec(new PojoCodec(new TupleKryoRegistrar()));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBytesPassThru() {
|
||||
byte[] payload = "foo".getBytes();
|
||||
Message<byte[]> message = MessageBuilder.withPayload(payload).build();
|
||||
MessageValues converted = binder.serializePayloadIfNecessary(message
|
||||
);
|
||||
assertSame(payload, converted.getPayload());
|
||||
Message<?> convertedMessage = converted.toMessage();
|
||||
assertSame(payload, convertedMessage.getPayload());
|
||||
assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM,
|
||||
contentTypeResolver.resolve(convertedMessage.getHeaders()));
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(convertedMessage);
|
||||
payload = (byte[]) reconstructed.getPayload();
|
||||
assertSame(converted.getPayload(), payload);
|
||||
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBytesPassThruContentType() {
|
||||
byte[] payload = "foo".getBytes();
|
||||
Message<byte[]> message = MessageBuilder.withPayload(payload)
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE)
|
||||
.build();
|
||||
MessageValues messageValues = binder.serializePayloadIfNecessary(message
|
||||
);
|
||||
Message<?> converted = messageValues.toMessage();
|
||||
assertSame(payload, converted.getPayload());
|
||||
assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM,
|
||||
contentTypeResolver.resolve(converted.getHeaders()));
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
payload = (byte[]) reconstructed.getPayload();
|
||||
assertSame(converted.getPayload(), payload);
|
||||
assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE,
|
||||
reconstructed.get(MessageHeaders.CONTENT_TYPE));
|
||||
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testString() throws IOException {
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(
|
||||
new GenericMessage<String>("foo"));
|
||||
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
assertEquals(MimeTypeUtils.TEXT_PLAIN,
|
||||
contentTypeResolver.resolve(converted.getHeaders()));
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertEquals("foo", reconstructed.getPayload());
|
||||
assertNull(reconstructed.get(MessageHeaders.CONTENT_TYPE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testContentTypePreserved() throws IOException {
|
||||
Message<String> inbound = MessageBuilder.withPayload("{\"foo\":\"foo\"}")
|
||||
.copyHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON))
|
||||
.build();
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(
|
||||
inbound);
|
||||
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
|
||||
assertEquals(MimeTypeUtils.TEXT_PLAIN,
|
||||
contentTypeResolver.resolve(converted.getHeaders()));
|
||||
assertEquals(MimeTypeUtils.APPLICATION_JSON,
|
||||
converted.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertEquals("{\"foo\":\"foo\"}", reconstructed.getPayload());
|
||||
assertEquals(MimeTypeUtils.APPLICATION_JSON, reconstructed.get(MessageHeaders.CONTENT_TYPE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPojoSerialization() {
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(
|
||||
new GenericMessage<Foo>(new Foo("bar"))
|
||||
);
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
|
||||
assertEquals("application", mimeType.getType());
|
||||
assertEquals("x-java-object", mimeType.getSubtype());
|
||||
assertEquals(Foo.class.getName(), mimeType.getParameter("type"));
|
||||
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertEquals("bar", ((Foo) reconstructed.getPayload()).getBar());
|
||||
assertNull(reconstructed.get(MessageHeaders.CONTENT_TYPE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPojoWithXJavaObjectMimeTypeNoType() {
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(
|
||||
new GenericMessage<Foo>(new Foo("bar"))
|
||||
);
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
|
||||
assertEquals("application", mimeType.getType());
|
||||
assertEquals("x-java-object", mimeType.getSubtype());
|
||||
assertEquals(Foo.class.getName(), mimeType.getParameter("type"));
|
||||
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertEquals("bar", ((Foo) reconstructed.getPayload()).getBar());
|
||||
assertNull(reconstructed.get(MessageHeaders.CONTENT_TYPE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPojoWithXJavaObjectMimeTypeExplicitType() {
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(
|
||||
new GenericMessage<Foo>(new Foo("bar"))
|
||||
);
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
|
||||
assertEquals("application", mimeType.getType());
|
||||
assertEquals("x-java-object", mimeType.getSubtype());
|
||||
assertEquals(Foo.class.getName(), mimeType.getParameter("type"));
|
||||
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertEquals("bar", ((Foo) reconstructed.getPayload()).getBar());
|
||||
assertNull(reconstructed.get(MessageHeaders.CONTENT_TYPE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTupleSerialization() {
|
||||
Tuple payload = TupleBuilder.tuple().of("foo", "bar");
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<Tuple>(payload)
|
||||
);
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
|
||||
assertEquals("application", mimeType.getType());
|
||||
assertEquals("x-java-object", mimeType.getSubtype());
|
||||
assertEquals(DefaultTuple.class.getName(), mimeType.getParameter("type"));
|
||||
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertEquals("bar", ((Tuple) reconstructed.getPayload()).getString("foo"));
|
||||
assertNull(reconstructed.get(MessageHeaders.CONTENT_TYPE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsSimpleObject() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new Object());
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertEquals(Object.class, Class.forName(className));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsObjectArray() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new String[0]);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertEquals(String[].class, Class.forName(className));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsMultiDimensionalObjectArray() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new String[0][0][0]);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertEquals(String[][][].class, Class.forName(className));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsPrimitiveArray() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new int[0]);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertEquals(int[].class, Class.forName(className));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsMultiDimensionalPrimitiveArray() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new int[0][0][0]);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertEquals(int[][][].class, Class.forName(className));
|
||||
}
|
||||
|
||||
public static class Foo {
|
||||
|
||||
private String bar;
|
||||
|
||||
public Foo() {
|
||||
}
|
||||
|
||||
public Foo(String bar) {
|
||||
this.bar = bar;
|
||||
}
|
||||
|
||||
public String getBar() {
|
||||
return bar;
|
||||
}
|
||||
|
||||
public void setBar(String bar) {
|
||||
this.bar = bar;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Bar {
|
||||
|
||||
private String foo;
|
||||
|
||||
public Bar() {
|
||||
}
|
||||
|
||||
public Bar(String foo) {
|
||||
this.foo = foo;
|
||||
}
|
||||
|
||||
public String getFoo() {
|
||||
return foo;
|
||||
}
|
||||
|
||||
public void setFoo(String foo) {
|
||||
this.foo = foo;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public class TestMessageChannelBinder extends MessageChannelBinderSupport {
|
||||
|
||||
@Override
|
||||
public void bindConsumer(String name, MessageChannel channel, Properties properties) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindPubSubConsumer(String name, MessageChannel moduleInputChannel,
|
||||
Properties properties) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindPubSubProducer(String name, MessageChannel moduleOutputChannel,
|
||||
Properties properties) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindProducer(String name, MessageChannel channel, Properties properties) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindRequestor(String name, MessageChannel requests, MessageChannel replies,
|
||||
Properties properties) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindReplier(String name, MessageChannel requests, MessageChannel replies,
|
||||
Properties properties) {
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user