Change streams->stream

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/README.md

@@ -0,0 +1,2 @@
+Spring XD Test Support
+======================

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/pom.xml

@@ -0,0 +1,120 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>org.springframework.cloud</groupId>
+	<artifactId>spring-cloud-stream-binding-test</artifactId>
+	<version>1.0.0.BUILD-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>spring-cloud-stream-binding-test</name>
+	<description>Test support for binding implementations</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-stream-bindings-parent</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<spring-amqp.version>1.4.5.RELEASE</spring-amqp.version>
+	</properties>
+	<dependencies>
+		<!-- There's no comparison with gradle's syntax, but I think we'll be happy sticking with
+				maven for the superior dependency and project management -->
+
+		<!--
+				compile ("org.springframework.integration:spring-integration-test"){
+				exclude group: 'org.apache.avro', module: 'avro-compiler'
+
+				}
+				compile "org.springframework.integration:spring-integration-amqp"
+				compile "org.springframework.integration:spring-integration-redis"
+				compile "org.springframework.integration:spring-integration-mqtt"
+				compile ("org.springframework.integration:spring-integration-kafka:$springIntegrationKafkaVersion") {
+				exclude group: 'org.apache.avro', module: 'avro-compiler'
+				}
+		-->
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-test</artifactId>
+			<version>${spring-integration.version}</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.apache.avro</groupId>
+					<artifactId>avro-compiler</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<!--<dependency>-->
+			<!--<groupId>org.springframework.integration</groupId>-->
+			<!--<artifactId>spring-integration-redis</artifactId>-->
+			<!--<version>${spring-integration.version}</version>-->
+		<!--</dependency>-->
+		<!--<dependency>-->
+			<!--<groupId>org.springframework.integration</groupId>-->
+			<!--<artifactId>spring-integration-mqtt</artifactId>-->
+			<!--<version>${spring-integration.version}</version>-->
+		<!--</dependency>-->
+		<!--<dependency>-->
+			<!--<groupId>org.springframework.integration</groupId>-->
+			<!--<artifactId>spring-integration-kafka</artifactId>-->
+			<!--<version>1.1.2.RELEASE</version>-->
+			<!--<exclusions>-->
+				<!--<exclusion>-->
+					<!--<groupId>org.apache.avro</groupId>-->
+					<!--<artifactId>avro-compiler</artifactId>-->
+				<!--</exclusion>-->
+			<!--</exclusions>-->
+		<!--</dependency>-->
+		<!--<dependency>-->
+			<!--<groupId>org.apache.kafka</groupId>-->
+			<!--<artifactId>kafka_2.10</artifactId>-->
+		<!--</dependency>-->
+		<!--<dependency>-->
+			<!--<groupId>org.apache.kafka</groupId>-->
+			<!--<artifactId>kafka_2.10</artifactId>-->
+			<!--<classifier>test</classifier>-->
+		<!--</dependency>-->
+		<!--<dependency>-->
+			<!--<groupId>org.apache.kafka</groupId>-->
+			<!--<artifactId>kafka-clients</artifactId>-->
+		<!--</dependency>-->
+		<!--<dependency>-->
+			<!--<groupId>org.apache.curator</groupId>-->
+			<!--<artifactId>curator-test</artifactId>-->
+		<!--</dependency>-->
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-web</artifactId>
+			<version>${spring-framework.version}</version>
+		</dependency>
+		<!--<dependency>-->
+			<!--<groupId>org.springframework.amqp</groupId>-->
+			<!--<artifactId>spring-rabbit</artifactId>-->
+			<!--<version>${spring-amqp.version}</version>-->
+		<!--</dependency>-->
+		<dependency>
+			<groupId>org.springframework.xd</groupId>
+			<artifactId>spring-xd-tuple</artifactId>
+			<version>${spring-xd.version}</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.springframework.xd</groupId>
+					<artifactId>spring-xd-codec</artifactId>
+				</exclusion>
+			</exclusions>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binding-spi</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-codec</artifactId>
+		</dependency>
+	</dependencies>
+</project>

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/dirt/integration/bus/AbstractMessageBusTests.java

@@ -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.xd.dirt.integration.bus;
+
+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.xd.dirt.integration.bus.MessageBus.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 AbstractMessageBusTests {
+
+	protected static final Collection<MediaType> ALL = Collections.singletonList(MediaType.ALL);
+
+	protected AbstractTestMessageBus<?> testMessageBus;
+
+	@Test
+	public void testClean() throws Exception {
+		MessageBus messageBus = getMessageBus();
+		messageBus.bindProducer("foo.0", new DirectChannel(), null);
+		messageBus.bindConsumer("foo.0", new DirectChannel(), null);
+		messageBus.bindProducer("foo.1", new DirectChannel(), null);
+		messageBus.bindConsumer("foo.1", new DirectChannel(), null);
+		messageBus.bindProducer("foo.2", new DirectChannel(), null);
+		Collection<?> bindings = getBindings(messageBus);
+		assertEquals(5, bindings.size());
+		messageBus.unbindProducers("foo.0");
+		assertEquals(4, bindings.size());
+		messageBus.unbindConsumers("foo.0");
+		messageBus.unbindProducers("foo.1");
+		assertEquals(2, bindings.size());
+		messageBus.unbindConsumers("foo.1");
+		messageBus.unbindProducers("foo.2");
+		assertTrue(bindings.isEmpty());
+	}
+
+	@Test
+	public void testSendAndReceive() throws Exception {
+		MessageBus messageBus = getMessageBus();
+		DirectChannel moduleOutputChannel = new DirectChannel();
+		QueueChannel moduleInputChannel = new QueueChannel();
+		messageBus.bindProducer("foo.0", moduleOutputChannel, null);
+		messageBus.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
+		busBindUnbindLatency();
+		moduleOutputChannel.send(message);
+		Message<?> inbound = moduleInputChannel.receive(5000);
+		assertNotNull(inbound);
+		assertEquals("foo", inbound.getPayload());
+		assertNull(inbound.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
+		assertEquals("foo/bar", inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
+		messageBus.unbindProducers("foo.0");
+		messageBus.unbindConsumers("foo.0");
+	}
+
+	@Test
+	public void testSendAndReceiveNoOriginalContentType() throws Exception {
+		MessageBus messageBus = getMessageBus();
+		DirectChannel moduleOutputChannel = new DirectChannel();
+		QueueChannel moduleInputChannel = new QueueChannel();
+		messageBus.bindProducer("bar.0", moduleOutputChannel, null);
+		messageBus.bindConsumer("bar.0", moduleInputChannel, null);
+		busBindUnbindLatency();
+
+		Message<?> message = MessageBuilder.withPayload("foo").build();
+		moduleOutputChannel.send(message);
+		Message<?> inbound = moduleInputChannel.receive(5000);
+		assertNotNull(inbound);
+		assertEquals("foo", inbound.getPayload());
+		assertNull(inbound.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
+		assertNull(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
+		messageBus.unbindProducers("bar.0");
+		messageBus.unbindConsumers("bar.0");
+	}
+
+	@Test
+	public void testSendAndReceivePubSub() throws Exception {
+		MessageBus messageBus = getMessageBus();
+		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();
+		messageBus.bindProducer("baz.0", moduleOutputChannel, null);
+		messageBus.bindConsumer("baz.0", moduleInputChannel, null);
+		moduleOutputChannel.addInterceptor(new WireTap(tapChannel));
+		messageBus.bindPubSubProducer("tap:baz.http", tapChannel, null);
+		// A new module is using the tap as an input channel
+		String fooTapName = messageBus.isCapable(Capability.DURABLE_PUBSUB) ? "foo.tap:baz.http" : "tap:baz.http";
+		messageBus.bindPubSubConsumer(fooTapName, module2InputChannel, null);
+		// Another new module is using tap as an input channel
+		String barTapName = messageBus.isCapable(Capability.DURABLE_PUBSUB) ? "bar.tap:baz.http" : "tap:baz.http";
+		messageBus.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(XdHeaders.XD_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(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
+			assertEquals("foo/bar", tapped1.getHeaders().get(MessageHeaders.CONTENT_TYPE));
+			assertEquals("foo", tapped2.getPayload());
+			assertNull(tapped2.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
+			assertEquals("foo/bar", tapped2.getHeaders().get(MessageHeaders.CONTENT_TYPE));
+		}
+		// delete one tap stream is deleted
+		messageBus.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
+		messageBus.unbindConsumer(fooTapName, module2InputChannel);
+		// Clean up as StreamPlugin would
+		messageBus.unbindConsumer("baz.0", moduleInputChannel);
+		messageBus.unbindProducer("baz.0", moduleOutputChannel);
+		messageBus.unbindProducers("tap:baz.http");
+		assertTrue(getBindings(messageBus).isEmpty());
+	}
+
+	@Test
+	public void createInboundPubSubBeforeOutboundPubSub() throws Exception {
+		MessageBus messageBus = getMessageBus();
+		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 = messageBus.isCapable(Capability.DURABLE_PUBSUB) ? "foo.tap:baz.http" : "tap:baz.http";
+		messageBus.bindPubSubConsumer(fooTapName, module2InputChannel, null);
+
+		// Then create the stream
+		messageBus.bindProducer("baz.0", moduleOutputChannel, null);
+		messageBus.bindConsumer("baz.0", moduleInputChannel, null);
+		moduleOutputChannel.addInterceptor(new WireTap(tapChannel));
+		messageBus.bindPubSubProducer("tap:baz.http", tapChannel, null);
+
+		// Another new module is using tap as an input channel
+		String barTapName = messageBus.isCapable(Capability.DURABLE_PUBSUB) ? "bar.tap:baz.http" : "tap:baz.http";
+		messageBus.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(XdHeaders.XD_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(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
+			assertEquals("foo/bar", tapped1.getHeaders().get(MessageHeaders.CONTENT_TYPE));
+			assertEquals("foo", tapped2.getPayload());
+			assertNull(tapped2.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
+			assertEquals("foo/bar", tapped2.getHeaders().get(MessageHeaders.CONTENT_TYPE));
+		}
+		// delete one tap stream is deleted
+		messageBus.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
+		messageBus.unbindConsumer(fooTapName, module2InputChannel);
+		// Clean up as StreamPlugin would
+		messageBus.unbindConsumer("baz.0", moduleInputChannel);
+		messageBus.unbindProducer("baz.0", moduleOutputChannel);
+		messageBus.unbindProducers("tap:baz.http");
+		assertTrue(getBindings(messageBus).isEmpty());
+	}
+
+	@Test
+	public void testBadDynamic() throws Exception {
+		Properties properties = new Properties();
+		properties.setProperty(BusProperties.PARTITION_KEY_EXPRESSION, "'foo'");
+		MessageBus messageBus = getMessageBus();
+		try {
+			messageBus.bindDynamicProducer("queue:foo", properties);
+			fail("Exception expected");
+		}
+		catch (MessageBusException mbe) {
+			Assert.assertEquals("Failed to bind dynamic channel 'queue:foo' with properties " +
+							"{partitionKeyExpression='foo'}",
+					mbe.getMessage());
+			if (messageBus instanceof AbstractTestMessageBus) {
+				messageBus = ((AbstractTestMessageBus) messageBus).getCoreMessageBus();
+			}
+			assertFalse(((MessageBusSupport) messageBus).getApplicationContext().containsBean("queue:foo"));
+		}
+	}
+
+	protected Collection<?> getBindings(MessageBus testMessageBus) {
+		if (testMessageBus instanceof AbstractTestMessageBus) {
+			return getBindingsFromMsgBus(((AbstractTestMessageBus) testMessageBus).getCoreMessageBus());
+		}
+		return Collections.EMPTY_LIST;
+	}
+
+	protected Collection<?> getBindingsFromMsgBus(MessageBus messageBus) {
+		DirectFieldAccessor accessor = new DirectFieldAccessor(messageBus);
+		return (List<?>) accessor.getPropertyValue("bindings");
+	}
+
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	protected MultiTypeCodec<Object> getCodec() {
+		return new PojoCodec();
+	}
+
+	protected abstract MessageBus getMessageBus() throws Exception;
+
+	@After
+	public void cleanup() {
+		if (testMessageBus != null) {
+			testMessageBus.cleanup();
+		}
+	}
+
+	/**
+	 * If appropriate, let the bus middleware settle down a bit while binding/unbinding actually happens.
+	 */
+	protected void busBindUnbindLatency() throws InterruptedException {
+		// default none
+	}
+
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/dirt/integration/bus/AbstractTestMessageBus.java

@@ -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.xd.dirt.integration.bus;
+
+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 MessageBus}.
+ *
+ * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
+ */
+public abstract class AbstractTestMessageBus<C extends MessageBusSupport> implements MessageBus {
+
+	protected Set<String> queues = new HashSet<String>();
+
+	protected Set<String> topics = new HashSet<String>();
+
+	private C messageBus;
+
+	public void setMessageBus(C messageBus) {
+		messageBus.setIntegrationEvaluationContext(new StandardEvaluationContext());
+		try {
+			messageBus.afterPropertiesSet();
+		}
+		catch (Exception e) {
+			throw new RuntimeException("Failed to initialize message bus", e);
+		}
+		this.messageBus = messageBus;
+	}
+
+	@Override
+	public void bindConsumer(String name, MessageChannel moduleInputChannel, Properties properties) {
+		messageBus.bindConsumer(name, moduleInputChannel, properties);
+		queues.add(name);
+	}
+
+	@Override
+	public void bindPubSubConsumer(String name, MessageChannel inputChannel, Properties properties) {
+		messageBus.bindPubSubConsumer(name, inputChannel, properties);
+		addTopic(name);
+	}
+
+	@Override
+	public void bindProducer(String name, MessageChannel moduleOutputChannel, Properties properties) {
+		messageBus.bindProducer(name, moduleOutputChannel, properties);
+		queues.add(name);
+	}
+
+	@Override
+	public void bindPubSubProducer(String name, MessageChannel outputChannel, Properties properties) {
+		messageBus.bindPubSubProducer(name, outputChannel, properties);
+		addTopic(name);
+	}
+
+	@Override
+	public void bindRequestor(String name, MessageChannel requests, MessageChannel replies,
+			Properties properties) {
+		messageBus.bindRequestor(name, requests, replies, properties);
+		queues.add(name + ".requests");
+	}
+
+	@Override
+	public void bindReplier(String name, MessageChannel requests, MessageChannel replies,
+			Properties properties) {
+		messageBus.bindReplier(name, requests, replies, properties);
+		queues.add(name + ".requests");
+	}
+
+	private void addTopic(String topicName) {
+		topics.add("topic." + topicName);
+	}
+
+	public C getCoreMessageBus() {
+		return messageBus;
+	}
+
+	public abstract void cleanup();
+
+	@Override
+	public void unbindConsumers(String name) {
+		messageBus.unbindConsumers(name);
+	}
+
+	@Override
+	public void unbindProducers(String name) {
+		messageBus.unbindProducers(name);
+	}
+
+	@Override
+	public void unbindConsumer(String name, MessageChannel channel) {
+		messageBus.unbindConsumer(name, channel);
+	}
+
+	@Override
+	public void unbindProducer(String name, MessageChannel channel) {
+		messageBus.unbindProducer(name, channel);
+	}
+
+	@Override
+	public MessageChannel bindDynamicProducer(String name, Properties properties) {
+		this.queues.add(name);
+		return this.messageBus.bindDynamicProducer(name, properties);
+	}
+
+	@Override
+	public MessageChannel bindDynamicPubSubProducer(String name, Properties properties) {
+		this.topics.add(name);
+		return this.messageBus.bindDynamicPubSubProducer(name, properties);
+	}
+
+	@Override
+	public boolean isCapable(Capability capability) {
+		return this.messageBus.isCapable(capability);
+	}
+
+	public MessageBus getMessageBus() {
+		return this.messageBus;
+	}
+
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/dirt/integration/bus/BrokerBusTests.java

@@ -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.xd.dirt.integration.bus;
+
+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 buses that use an external broker.
+ *
+ * @author Gary Russell
+ */
+public abstract class BrokerBusTests extends
+AbstractMessageBusTests {
+
+	@Test
+	public void testDirectBinding() throws Exception {
+		MessageBus bus = getMessageBus();
+		Properties properties = new Properties();
+		properties.setProperty(BusProperties.DIRECT_BINDING_ALLOWED, "true");
+
+		DirectChannel moduleInputChannel = new DirectChannel();
+		moduleInputChannel.setBeanName("direct.input");
+		DirectChannel moduleOutputChannel = new DirectChannel();
+		moduleOutputChannel.setBeanName("direct.output");
+		bus.bindConsumer("direct.0", moduleInputChannel, null);
+		bus.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 bus
+		bus.unbindConsumers("direct.0");
+		busBindUnbindLatency();
+
+		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 bus and bind directly again
+		caller.set(null);
+		bus.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));
+
+		bus.unbindProducers("direct.0");
+		bus.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);
+
+
+
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/dirt/integration/bus/BusTestUtils.java

@@ -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.xd.dirt.integration.bus;
+
+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 BusTestUtils {
+
+	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);
+	}
+
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/dirt/integration/bus/PartitionCapableBusTests.java

@@ -0,0 +1,271 @@
+/*
+ * 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.xd.dirt.integration.bus;
+
+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;
+import org.springframework.xd.test.TestUtils;
+
+
+/**
+ * Tests for buses that support partitioning.
+ *
+ * @author Gary Russell
+ */
+abstract public class PartitionCapableBusTests extends BrokerBusTests {
+
+	@Test
+	public void testBadProperties() throws Exception {
+		MessageBus bus = getMessageBus();
+		Properties properties = new Properties();
+		properties.put("foo", "bar");
+		properties.put("baz", "qux");
+
+		DirectChannel output = new DirectChannel();
+		try {
+			bus.bindProducer("badprops.0", output, properties);
+		}
+		catch (IllegalArgumentException e) {
+			assertThat(e.getMessage(), allOf(Matchers.containsString(bus.getClass().getSimpleName().replace("Test", "")
+					+ " does not support producer "),
+					containsString("foo"),
+					containsString("baz"),
+					containsString(" for badprops.0.")));
+		}
+
+		properties.remove("baz");
+		try {
+			bus.bindConsumer("badprops.0", output, properties);
+		}
+		catch (IllegalArgumentException e) {
+			assertThat(e.getMessage(), equalTo(bus.getClass().getSimpleName().replace("Test", "")
+					+ " does not support consumer property: foo for badprops.0."));
+		}
+	}
+
+	@Test
+	public void testPartitionedModuleSpEL() throws Exception {
+		MessageBus bus = getMessageBus();
+		Properties properties = new Properties();
+		properties.put("partitionKeyExpression", "payload");
+		properties.put("partitionSelectorExpression", "hashCode()");
+		properties.put(BusProperties.NEXT_MODULE_COUNT, "3");
+		properties.put(BusProperties.NEXT_MODULE_CONCURRENCY, "2");
+
+		DirectChannel output = new DirectChannel();
+		output.setBeanName("test.output");
+		bus.bindProducer("part.0", output, properties);
+		@SuppressWarnings("unchecked")
+		List<Binding> bindings = TestUtils.getPropertyValue(bus, "messageBus.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");
+		bus.bindConsumer("part.0", input0, properties);
+		properties.put("partitionIndex", "1");
+		QueueChannel input1 = new QueueChannel();
+		input1.setBeanName("test.input1S");
+		bus.bindConsumer("part.0", input1, properties);
+		properties.put("partitionIndex", "2");
+		QueueChannel input2 = new QueueChannel();
+		input2.setBeanName("test.input2S");
+		bus.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("xdReplyChannel", "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("xdReplyChannel"));
+				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);
+
+		}
+
+		bus.unbindConsumers("part.0");
+		bus.unbindProducers("part.0");
+	}
+
+	@Test
+	public void testPartitionedModuleJava() throws Exception {
+		MessageBus bus = getMessageBus();
+		Properties properties = new Properties();
+		properties.put("partitionKeyExtractorClass", "org.springframework.xd.dirt.integration.bus.PartitionTestSupport");
+		properties.put("partitionSelectorClass", "org.springframework.xd.dirt.integration.bus.PartitionTestSupport");
+		properties.put(BusProperties.NEXT_MODULE_COUNT, "3");
+		properties.put(BusProperties.NEXT_MODULE_CONCURRENCY, "2");
+
+		DirectChannel output = new DirectChannel();
+		output.setBeanName("test.output");
+		bus.bindProducer("partJ.0", output, properties);
+		@SuppressWarnings("unchecked")
+		List<Binding> bindings = TestUtils.getPropertyValue(bus, "messageBus.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");
+		bus.bindConsumer("partJ.0", input0, properties);
+		properties.put("partitionIndex", "1");
+		QueueChannel input1 = new QueueChannel();
+		input1.setBeanName("test.input1J");
+		bus.bindConsumer("partJ.0", input1, properties);
+		properties.put("partitionIndex", "2");
+		QueueChannel input2 = new QueueChannel();
+		input2.setBeanName("test.input2J");
+		bus.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));
+		}
+
+		bus.unbindConsumers("partJ.0");
+		bus.unbindProducers("partJ.0");
+	}
+
+	/**
+	 * Implementations should return whether the bus 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();
+	}
+
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/dirt/integration/bus/PartitionTestSupport.java

@@ -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.xd.dirt.integration.bus;
+
+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();
+	}
+
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/dirt/integration/bus/Spy.java

@@ -0,0 +1,13 @@
+package org.springframework.xd.dirt.integration.bus;
+
+/**
+ * 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;
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/test/AbstractExternalResourceTestSupport.java

@@ -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.xd.test;
+
+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;
+
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/main/java/org/springframework/xd/test/TestUtils.java

@@ -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.xd.test;
+
+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;
+    }
+}

spring-cloud-stream-bindings/spring-cloud-stream-binding-test/src/test/java/org/springframework/xd/dirt/integration/bus/MessageBusSupportTests.java

@@ -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.xd.dirt.integration.bus;
+
+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.xd.dirt.integration.bus.MessageBusSupport.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 MessageBusSupportTests {
+
+	private ContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver();
+
+	private final TestMessageBus messageBus = new TestMessageBus();
+
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	@Before
+	public void setUp() {
+		messageBus.setCodec(new PojoCodec(new TupleKryoRegistrar()));
+	}
+
+	@Test
+	public void testBytesPassThru() {
+		byte[] payload = "foo".getBytes();
+		Message<byte[]> message = MessageBuilder.withPayload(payload).build();
+		MessageValues converted = messageBus.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 = messageBus.deserializePayloadIfNecessary(convertedMessage);
+		payload = (byte[]) reconstructed.getPayload();
+		assertSame(converted.getPayload(), payload);
+		assertNull(reconstructed.get(XdHeaders.XD_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 = messageBus.serializePayloadIfNecessary(message
+		);
+		Message<?> converted = messageValues.toMessage();
+		assertSame(payload, converted.getPayload());
+		assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM,
+				contentTypeResolver.resolve(converted.getHeaders()));
+		MessageValues reconstructed = messageBus.deserializePayloadIfNecessary(converted);
+		payload = (byte[]) reconstructed.getPayload();
+		assertSame(converted.getPayload(), payload);
+		assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE,
+				reconstructed.get(MessageHeaders.CONTENT_TYPE));
+		assertNull(reconstructed.get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
+	}
+
+	@Test
+	public void testString() throws IOException {
+		MessageValues convertedValues = messageBus.serializePayloadIfNecessary(
+				new GenericMessage<String>("foo"));
+
+		Message<?> converted = convertedValues.toMessage();
+		assertEquals(MimeTypeUtils.TEXT_PLAIN,
+				contentTypeResolver.resolve(converted.getHeaders()));
+		MessageValues reconstructed = messageBus.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 = messageBus.serializePayloadIfNecessary(
+				inbound);
+
+		Message<?> converted = convertedValues.toMessage();
+
+		assertEquals(MimeTypeUtils.TEXT_PLAIN,
+				contentTypeResolver.resolve(converted.getHeaders()));
+		assertEquals(MimeTypeUtils.APPLICATION_JSON,
+				converted.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
+		MessageValues reconstructed = messageBus.deserializePayloadIfNecessary(converted);
+		assertEquals("{\"foo\":\"foo\"}", reconstructed.getPayload());
+		assertEquals(MimeTypeUtils.APPLICATION_JSON, reconstructed.get(MessageHeaders.CONTENT_TYPE));
+	}
+
+	@Test
+	public void testPojoSerialization() {
+		MessageValues convertedValues = messageBus.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 = messageBus.deserializePayloadIfNecessary(converted);
+		assertEquals("bar", ((Foo) reconstructed.getPayload()).getBar());
+		assertNull(reconstructed.get(MessageHeaders.CONTENT_TYPE));
+	}
+
+	@Test
+	public void testPojoWithXJavaObjectMimeTypeNoType() {
+		MessageValues convertedValues = messageBus.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 = messageBus.deserializePayloadIfNecessary(converted);
+		assertEquals("bar", ((Foo) reconstructed.getPayload()).getBar());
+		assertNull(reconstructed.get(MessageHeaders.CONTENT_TYPE));
+	}
+
+	@Test
+	public void testPojoWithXJavaObjectMimeTypeExplicitType() {
+		MessageValues convertedValues = messageBus.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 = messageBus.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 = messageBus.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 = messageBus.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 TestMessageBus extends MessageBusSupport {
+
+		@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) {
+		}
+	}
+
+}