Rename spring-cloud-stream-binder-test

- Rename to spring-cloud-stream-test-support
- Make downstream binder dependency changes

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java

@@ -0,0 +1,477 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      https://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.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
+import org.springframework.cloud.stream.config.BindingProperties;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.context.Lifecycle;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+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.PollableChannel;
+import org.springframework.messaging.converter.SmartMessageConverter;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeTypeUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Gary Russell
+ * @author Ilayaperumal Gopinathan
+ * @author David Turanski
+ * @author Mark Fisher
+ * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
+ * @author Jacob Severson
+ * @author Artem Bilan
+ */
+// @checkstyle:off
+@SuppressWarnings("unchecked")
+public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties> {
+
+	// @checkstyle:on
+
+	protected final Log logger = LogFactory.getLog(this.getClass());
+
+	protected B testBinder;
+
+	protected SmartMessageConverter messageConverter;
+
+	protected GenericApplicationContext applicationContext;
+
+	/**
+	 * Subclasses may override this default value to have tests wait longer for a message
+	 * receive, for example if running in an environment that is known to be slow (e.g.
+	 * travis).
+	 */
+	protected double timeoutMultiplier = 1.0D;
+
+	@BeforeEach
+	public void before() {
+		applicationContext = new GenericApplicationContext();
+		applicationContext.refresh();
+		this.messageConverter = new CompositeMessageConverterFactory()
+				.getMessageConverterForAllRegistered();
+	}
+
+	/**
+	 * Attempt to receive a message on the given channel, waiting up to 1s (times the
+	 * {@link #timeoutMultiplier}).
+	 */
+	protected Message<?> receive(PollableChannel channel) {
+		return receive(channel, 1);
+	}
+
+	/**
+	 * Attempt to receive a message on the given channel, waiting up to 1s *
+	 * additionalMultiplier * {@link #timeoutMultiplier}).
+	 *
+	 * Allows accomodating tests which are slower than normal (e.g. retry).
+	 */
+	protected Message<?> receive(PollableChannel channel, int additionalMultiplier) {
+		long startTime = System.currentTimeMillis();
+		Message<?> receive = channel
+				.receive((int) (1000 * this.timeoutMultiplier * additionalMultiplier));
+		long elapsed = System.currentTimeMillis() - startTime;
+		this.logger.debug("receive() took " + elapsed / 1000 + " seconds");
+		return receive;
+	}
+
+	@Test
+	@SuppressWarnings("rawtypes")
+	public void testClean(TestInfo testInfo) throws Exception {
+		Binder binder = getBinder();
+		Binding<MessageChannel> foo0ProducerBinding = binder.bindProducer(
+				String.format("foo%s0", getDestinationNameDelimiter()),
+				this.createBindableChannel("output", new BindingProperties()),
+				createProducerProperties(testInfo));
+		Binding<MessageChannel> foo0ConsumerBinding = binder.bindConsumer(
+				String.format("foo%s0", getDestinationNameDelimiter()), "testClean",
+				this.createBindableChannel("input", new BindingProperties()),
+				createConsumerProperties());
+		Binding<MessageChannel> foo1ProducerBinding = binder.bindProducer(
+				String.format("foo%s1", getDestinationNameDelimiter()),
+				this.createBindableChannel("output", new BindingProperties()),
+				createProducerProperties(testInfo));
+		Binding<MessageChannel> foo1ConsumerBinding = binder.bindConsumer(
+				String.format("foo%s1", getDestinationNameDelimiter()), "testClean",
+				this.createBindableChannel("input", new BindingProperties()),
+				createConsumerProperties());
+		Binding<MessageChannel> foo2ProducerBinding = binder.bindProducer(
+				String.format("foo%s2", getDestinationNameDelimiter()),
+				this.createBindableChannel("output", new BindingProperties()),
+				createProducerProperties(testInfo));
+		foo0ProducerBinding.unbind();
+		assertThat(TestUtils
+				.getPropertyValue(foo0ProducerBinding, "lifecycle", Lifecycle.class)
+				.isRunning()).isFalse();
+		foo0ConsumerBinding.unbind();
+		foo1ProducerBinding.unbind();
+		assertThat(TestUtils
+				.getPropertyValue(foo0ConsumerBinding, "lifecycle", Lifecycle.class)
+				.isRunning()).isFalse();
+		assertThat(TestUtils
+				.getPropertyValue(foo1ProducerBinding, "lifecycle", Lifecycle.class)
+				.isRunning()).isFalse();
+		foo1ConsumerBinding.unbind();
+		foo2ProducerBinding.unbind();
+		assertThat(TestUtils
+				.getPropertyValue(foo1ConsumerBinding, "lifecycle", Lifecycle.class)
+				.isRunning()).isFalse();
+		assertThat(TestUtils
+				.getPropertyValue(foo2ProducerBinding, "lifecycle", Lifecycle.class)
+				.isRunning()).isFalse();
+	}
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	public void testSendAndReceive(TestInfo testInfo) throws Exception {
+		Binder binder = getBinder();
+		BindingProperties outputBindingProperties = createProducerBindingProperties(
+				createProducerProperties(testInfo));
+		DirectChannel moduleOutputChannel = createBindableChannel("output",
+				outputBindingProperties);
+
+		BindingProperties inputBindingProperties = createConsumerBindingProperties(
+				createConsumerProperties());
+		DirectChannel moduleInputChannel = createBindableChannel("input",
+				inputBindingProperties);
+
+		Binding<MessageChannel> producerBinding = binder.bindProducer(
+				String.format("foo%s0", getDestinationNameDelimiter()),
+				moduleOutputChannel, outputBindingProperties.getProducer());
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer(
+				String.format("foo%s0", getDestinationNameDelimiter()),
+				"testSendAndReceive", moduleInputChannel,
+				inputBindingProperties.getConsumer());
+		Message<?> message = MessageBuilder.withPayload("foo")
+				.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build();
+		// Let the consumer actually bind to the producer before sending a msg
+		binderBindUnbindLatency();
+
+		CountDownLatch latch = new CountDownLatch(1);
+		AtomicReference<Message<byte[]>> inboundMessageRef = new AtomicReference<Message<byte[]>>();
+		moduleInputChannel.subscribe(message1 -> {
+			try {
+				inboundMessageRef.set((Message<byte[]>) message1);
+			}
+			finally {
+				latch.countDown();
+			}
+		});
+
+		moduleOutputChannel.send(message);
+		Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message");
+
+		assertThat(inboundMessageRef.get().getPayload()).isEqualTo("foo".getBytes());
+		assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE)
+				.toString()).isEqualTo("text/plain");
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
+	@Test
+	@SuppressWarnings("rawtypes")
+	public void testSendAndReceiveMultipleTopics(TestInfo testInfo) throws Exception {
+		Binder binder = getBinder();
+
+		BindingProperties producerBindingProperties = createProducerBindingProperties(
+				createProducerProperties(testInfo));
+
+		DirectChannel moduleOutputChannel1 = createBindableChannel("output1",
+				producerBindingProperties);
+
+		DirectChannel moduleOutputChannel2 = createBindableChannel("output2",
+				producerBindingProperties);
+
+		QueueChannel moduleInputChannel = new QueueChannel();
+
+		Binding<MessageChannel> producerBinding1 = binder.bindProducer(
+				String.format("foo%sxy", getDestinationNameDelimiter()),
+				moduleOutputChannel1, producerBindingProperties.getProducer());
+		Binding<MessageChannel> producerBinding2 = binder.bindProducer(
+				String.format("foo%syz",
+
+						getDestinationNameDelimiter()),
+				moduleOutputChannel2, producerBindingProperties.getProducer());
+
+		Binding<MessageChannel> consumerBinding1 = binder.bindConsumer(
+				String.format("foo%sxy", getDestinationNameDelimiter()),
+				"testSendAndReceiveMultipleTopics", moduleInputChannel,
+				createConsumerProperties());
+		Binding<MessageChannel> consumerBinding2 = binder.bindConsumer(
+				String.format("foo%syz", getDestinationNameDelimiter()),
+				"testSendAndReceiveMultipleTopics", moduleInputChannel,
+				createConsumerProperties());
+
+		String testPayload1 = "foo" + UUID.randomUUID().toString();
+		Message<?> message1 = MessageBuilder.withPayload(testPayload1.getBytes())
+				.setHeader(MessageHeaders.CONTENT_TYPE,
+						MimeTypeUtils.APPLICATION_OCTET_STREAM)
+				.build();
+		String testPayload2 = "foo" + UUID.randomUUID().toString();
+		Message<?> message2 = MessageBuilder.withPayload(testPayload2.getBytes())
+				.setHeader(MessageHeaders.CONTENT_TYPE,
+						MimeTypeUtils.APPLICATION_OCTET_STREAM)
+				.build();
+
+		// Let the consumer actually bind to the producer before sending a msg
+		binderBindUnbindLatency();
+		moduleOutputChannel1.send(message1);
+		moduleOutputChannel2.send(message2);
+
+		Message<?>[] messages = new Message[2];
+		messages[0] = receive(moduleInputChannel);
+		messages[1] = receive(moduleInputChannel);
+
+		assertThat(messages[0]).isNotNull();
+		assertThat(messages[1]).isNotNull();
+		assertThat(messages).extracting("payload").containsExactlyInAnyOrder(
+				testPayload1.getBytes(), testPayload2.getBytes());
+
+		producerBinding1.unbind();
+		producerBinding2.unbind();
+
+		consumerBinding1.unbind();
+		consumerBinding2.unbind();
+	}
+
+	@Test
+	@SuppressWarnings("rawtypes")
+	public void testSendAndReceiveNoOriginalContentType(TestInfo testInfo) throws Exception {
+		Binder binder = getBinder();
+
+		BindingProperties producerBindingProperties = createProducerBindingProperties(
+				createProducerProperties(testInfo));
+		DirectChannel moduleOutputChannel = createBindableChannel("output",
+				producerBindingProperties);
+		BindingProperties inputBindingProperties = createConsumerBindingProperties(
+				createConsumerProperties());
+		DirectChannel moduleInputChannel = createBindableChannel("input",
+				inputBindingProperties);
+		Binding<MessageChannel> producerBinding = binder.bindProducer(
+				String.format("bar%s0", getDestinationNameDelimiter()),
+				moduleOutputChannel, producerBindingProperties.getProducer());
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer(
+				String.format("bar%s0", getDestinationNameDelimiter()),
+				"testSendAndReceiveNoOriginalContentType", moduleInputChannel,
+				createConsumerProperties());
+		binderBindUnbindLatency();
+
+		Message<?> message = MessageBuilder.withPayload("foo")
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build();
+		moduleOutputChannel.send(message);
+		CountDownLatch latch = new CountDownLatch(1);
+		AtomicReference<Message<byte[]>> inboundMessageRef = new AtomicReference<Message<byte[]>>();
+		moduleInputChannel.subscribe(message1 -> {
+			try {
+				inboundMessageRef.set((Message<byte[]>) message1);
+			}
+			finally {
+				latch.countDown();
+			}
+		});
+
+		moduleOutputChannel.send(message);
+		Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message");
+		assertThat(inboundMessageRef.get()).isNotNull();
+		assertThat(inboundMessageRef.get().getPayload()).isEqualTo("foo".getBytes());
+		assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE)
+				.toString()).isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
+	protected abstract B getBinder() throws Exception;
+
+	protected abstract CP createConsumerProperties();
+
+	protected abstract PP createProducerProperties(TestInfo testInfo);
+
+	protected final BindingProperties createConsumerBindingProperties(
+			CP consumerProperties) {
+		BindingProperties bindingProperties = new BindingProperties();
+		bindingProperties.setConsumer(consumerProperties);
+		return bindingProperties;
+	}
+
+	protected BindingProperties createProducerBindingProperties(PP producerProperties) {
+		BindingProperties bindingProperties = new BindingProperties();
+		bindingProperties.setProducer(producerProperties);
+		return bindingProperties;
+	}
+
+	protected DirectChannel createBindableChannel(String channelName,
+			BindingProperties bindingProperties) throws Exception {
+		// The 'channelName.contains("input")' is strictly for convenience to avoid
+		// modifications in multiple tests
+		return this.createBindableChannel(channelName, bindingProperties,
+				channelName.contains("input"));
+	}
+
+	protected DirectChannel createBindableChannel(String channelName,
+			BindingProperties bindingProperties, boolean inputChannel) throws Exception {
+		MessageConverterConfigurer messageConverterConfigurer = createConverterConfigurer(
+				channelName, bindingProperties);
+		DirectChannel channel = new DirectChannel();
+		channel.setBeanName(channelName);
+		if (inputChannel) {
+			messageConverterConfigurer.configureInputChannel(channel, channelName);
+		}
+		else {
+			messageConverterConfigurer.configureOutputChannel(channel, channelName);
+		}
+		return channel;
+	}
+
+	protected DefaultPollableMessageSource createBindableMessageSource(String bindingName,
+			BindingProperties bindingProperties) throws Exception {
+		DefaultPollableMessageSource source = new DefaultPollableMessageSource(
+				new CompositeMessageConverterFactory()
+						.getMessageConverterForAllRegistered());
+		createConverterConfigurer(bindingName, bindingProperties)
+				.configurePolledMessageSource(source, bindingName);
+		return source;
+	}
+
+	private MessageConverterConfigurer createConverterConfigurer(String channelName,
+			BindingProperties bindingProperties) throws Exception {
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		bindingServiceProperties.getBindings().put(channelName, bindingProperties);
+		bindingServiceProperties.setApplicationContext(applicationContext);
+		bindingServiceProperties.setConversionService(new DefaultConversionService());
+		bindingServiceProperties.afterPropertiesSet();
+		MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer(
+				bindingServiceProperties,
+				new CompositeMessageConverterFactory(null, null, null).getMessageConverterForAllRegistered());
+		messageConverterConfigurer.setBeanFactory(applicationContext.getBeanFactory());
+		return messageConverterConfigurer;
+	}
+
+	@AfterEach
+	public void cleanup() {
+		if (this.testBinder != null) {
+			this.testBinder.cleanup();
+		}
+	}
+
+	/**
+	 * If appropriate, let the binder middleware settle down a bit while binding/unbinding
+	 * actually happens.
+	 */
+	protected void binderBindUnbindLatency() throws InterruptedException {
+		// default none
+	}
+
+	/**
+	 * 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(String name);
+
+	/**
+	 * Set the delimiter that will be used in the message source/target name. Some brokers
+	 * may have naming constraints (such as SQS), so this provides a way to override the
+	 * character being used as a delimiter. The default is a period.
+	 */
+	protected String getDestinationNameDelimiter() {
+		return ".";
+	}
+
+	@SuppressWarnings("unused") // it is used via reflection
+	private Station echoStation(Station station) {
+		return station;
+	}
+
+	@SuppressWarnings("unused") // it is used via reflection
+	private String echoStationString(String station) {
+		return station;
+	}
+
+	public static class Station {
+
+		List<Readings> readings = new ArrayList<>();
+
+		public List<Readings> getReadings() {
+			return this.readings;
+		}
+
+		public void setReadings(List<Readings> readings) {
+			this.readings = readings;
+		}
+
+		@SuppressWarnings("serial")
+		public static class Readings implements Serializable {
+
+			public String stationid;
+
+			public String customerid;
+
+			public String timestamp;
+
+			public String getStationid() {
+				return this.stationid;
+			}
+
+			public void setStationid(String stationid) {
+				this.stationid = stationid;
+			}
+
+			public String getCustomerid() {
+				return this.customerid;
+			}
+
+			public void setCustomerid(String customerid) {
+				this.customerid = customerid;
+			}
+
+			public String getTimestamp() {
+				return this.timestamp;
+			}
+
+			public void setTimestamp(String timestamp) {
+				this.timestamp = timestamp;
+			}
+
+		}
+
+	}
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractPollableConsumerTestBinder.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ *      https://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.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * @param <C> binder type
+ * @param <CP> consumer properties type
+ * @param <PP> producer properties type
+ * @author Gary Russell
+ * @since 2.0
+ */
+// @checkstyle:off
+public abstract class AbstractPollableConsumerTestBinder<C extends AbstractBinder<MessageChannel, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties>
+		extends AbstractTestBinder<C, CP, PP>
+		implements PollableConsumerBinder<MessageHandler, CP> {
+
+	// @checkstyle:on
+
+	private PollableConsumerBinder<MessageHandler, CP> binder;
+
+	@SuppressWarnings("unchecked")
+	public void setPollableConsumerBinder(
+			PollableConsumerBinder<MessageHandler, CP> binder) {
+		super.setBinder((C) binder);
+		this.binder = binder;
+	}
+
+	@Override
+	public Binding<PollableSource<MessageHandler>> bindPollableConsumer(String name,
+			String group, PollableSource<MessageHandler> inboundBindTarget,
+			CP consumerProperties) {
+		return this.binder.bindPollableConsumer(name, group, inboundBindTarget,
+				consumerProperties);
+	}
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2014-2017 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
+ *
+ *      https://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.Set;
+
+import org.springframework.integration.channel.AbstractSubscribableChannel;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Abstract class that adds test support for {@link Binder}.
+ *
+ * @param <C> binder type
+ * @param <CP> consumer properties type
+ * @param <PP> producer properties type
+ * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
+ * @author Mark Fisher
+ * @author Oleg Zhurakousky
+ */
+// @checkstyle:off
+public abstract class AbstractTestBinder<C extends AbstractBinder<MessageChannel, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties>
+		implements Binder<MessageChannel, CP, PP> {
+
+	// @checkstyle:on
+
+	protected Set<String> queues = new HashSet<String>();
+
+	private C binder;
+
+	@Override
+	public Binding<MessageChannel> bindConsumer(String name, String group,
+			MessageChannel moduleInputChannel, CP properties) {
+		this.checkChannelIsConfigured(moduleInputChannel, properties);
+		this.queues.add(name);
+		return this.binder.bindConsumer(name, group, moduleInputChannel, properties);
+	}
+
+	@Override
+	public Binding<MessageChannel> bindProducer(String name,
+			MessageChannel moduleOutputChannel, PP properties) {
+		this.queues.add(name);
+		return this.binder.bindProducer(name, moduleOutputChannel, properties);
+	}
+
+	public C getCoreBinder() {
+		return this.binder;
+	}
+
+	public abstract void cleanup();
+
+	public C getBinder() {
+		return this.binder;
+	}
+
+	public void setBinder(C binder) {
+		try {
+			binder.afterPropertiesSet();
+		}
+		catch (Exception e) {
+			throw new RuntimeException("Failed to initialize binder", e);
+		}
+		this.binder = binder;
+	}
+
+	/*
+	 * This will ensure that any MessageChannel that was passed to one of the bind*()
+	 * methods was properly configured (i.e., interceptors, converters etc). see
+	 * org.springframework.cloud.stream.binding.MessageConverterConfigurer
+	 */
+	private void checkChannelIsConfigured(MessageChannel messageChannel, CP properties) {
+		if (messageChannel instanceof AbstractSubscribableChannel
+				&& !properties.isUseNativeDecoding()) {
+			Assert.isTrue(
+					!CollectionUtils
+							.isEmpty(((AbstractSubscribableChannel) messageChannel)
+								.getInterceptors()),
+					"'messageChannel' appears to be misconfigured. "
+							+ "Consider creating channel via AbstractBinderTest.createBindableChannel(..)");
+		}
+	}
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/BinderTestEnvironmentPostProcessor.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2016-2020 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
+ *
+ *      https://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.HashMap;
+import java.util.Map;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.env.EnvironmentPostProcessor;
+import org.springframework.core.env.ConfigurableEnvironment;
+import org.springframework.core.env.MapPropertySource;
+
+/**
+ * An {@link EnvironmentPostProcessor} that sets some common configuration properties (log
+ * config etc.,) for binder tests.
+ *
+ * @author Ilayaperumal Gopinathan
+ */
+public class BinderTestEnvironmentPostProcessor implements EnvironmentPostProcessor {
+
+	@Override
+	public void postProcessEnvironment(ConfigurableEnvironment environment,
+			SpringApplication application) {
+		Map<String, Object> propertiesToAdd = new HashMap<>();
+		propertiesToAdd.put("logging.pattern.console",
+				"%d{ISO8601} %5p %t %c{2}:%L - %m%n");
+		environment.getPropertySources().addLast(
+				new MapPropertySource("binderTestPropertiesConfig", propertiesToAdd));
+	}
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/BinderTestUtils.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2014-2019 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
+ *
+ *      https://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.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.integration.support.MessageBuilderFactory;
+import org.springframework.integration.support.MutableMessageBuilderFactory;
+import org.springframework.integration.support.utils.IntegrationUtils;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Gary Russell
+ */
+public abstract class BinderTestUtils {
+
+	/**
+	 * Mocked application context.
+	 */
+	public static final AbstractApplicationContext MOCK_AC = mock(
+			AbstractApplicationContext.class);
+
+	/**
+	 * Mocked application bean factory.
+	 */
+	public static final ConfigurableListableBeanFactory MOCK_BF = mock(
+			ConfigurableListableBeanFactory.class);
+
+	private static final MessageBuilderFactory mbf = new MutableMessageBuilderFactory();
+
+	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);
+	}
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java

@@ -0,0 +1,341 @@
+/*
+ * Copyright 2014-2018 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
+ *
+ *      https://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.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+import org.assertj.core.api.Condition;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.cloud.stream.config.BindingProperties;
+import org.springframework.context.Lifecycle;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.MimeTypeUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for binders that support partitioning.
+ *
+ * @author Gary Russell
+ * @author Mark Fisher
+ * @author Marius Bogoevici
+ * @author Vinicius Carvalho
+ */
+// @checkstyle:off
+public abstract class PartitionCapableBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties>
+		extends AbstractBinderTests<B, CP, PP> {
+
+	// @checkstyle:on
+
+	protected static final SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testAnonymousGroup(TestInfo testInfo) throws Exception {
+		B binder = getBinder();
+		BindingProperties producerBindingProperties = createProducerBindingProperties(
+				createProducerProperties(testInfo));
+		DirectChannel output = createBindableChannel("output", producerBindingProperties);
+		Binding<MessageChannel> producerBinding = binder.bindProducer(
+				String.format("defaultGroup%s0", getDestinationNameDelimiter()), output,
+				(PP) producerBindingProperties.getProducer());
+
+		QueueChannel input1 = new QueueChannel();
+		Binding<MessageChannel> binding1 = binder.bindConsumer(
+				String.format("defaultGroup%s0", getDestinationNameDelimiter()), null,
+				input1, createConsumerProperties());
+
+		QueueChannel input2 = new QueueChannel();
+		Binding<MessageChannel> binding2 = binder.bindConsumer(
+				String.format("defaultGroup%s0", getDestinationNameDelimiter()), null,
+				input2, createConsumerProperties());
+
+		String testPayload1 = "foo-" + UUID.randomUUID().toString();
+		output.send(MessageBuilder.withPayload(testPayload1)
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				.build());
+
+		Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
+		assertThat(receivedMessage1).isNotNull();
+		assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1);
+
+		Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input2);
+		assertThat(receivedMessage2).isNotNull();
+		assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1);
+
+		binding2.unbind();
+
+		String testPayload2 = "foo-" + UUID.randomUUID().toString();
+		output.send(MessageBuilder.withPayload(testPayload2)
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				.build());
+
+		binding2 = binder.bindConsumer(
+				String.format("defaultGroup%s0", getDestinationNameDelimiter()), null,
+				input2, createConsumerProperties());
+		String testPayload3 = "foo-" + UUID.randomUUID().toString();
+		output.send(MessageBuilder.withPayload(testPayload3)
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				.build());
+
+		receivedMessage1 = (Message<byte[]>) receive(input1);
+		assertThat(receivedMessage1).isNotNull();
+		assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload2);
+		receivedMessage1 = (Message<byte[]>) receive(input1);
+		assertThat(receivedMessage1).isNotNull();
+		assertThat(new String(receivedMessage1.getPayload())).isNotNull();
+
+		receivedMessage2 = (Message<byte[]>) receive(input2);
+		assertThat(receivedMessage2).isNotNull();
+		assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload3);
+
+		producerBinding.unbind();
+		binding1.unbind();
+		binding2.unbind();
+	}
+
+	@Test
+	public void testOneRequiredGroup(TestInfo testInfo) throws Exception {
+		B binder = getBinder();
+		PP producerProperties = createProducerProperties(testInfo);
+		DirectChannel output = createBindableChannel("output",
+				createProducerBindingProperties(producerProperties));
+
+		String testDestination = "testDestination"
+				+ UUID.randomUUID().toString().replace("-", "");
+
+		producerProperties.setRequiredGroups("test1");
+		Binding<MessageChannel> producerBinding = binder.bindProducer(testDestination,
+				output, producerProperties);
+
+		String testPayload = "foo-" + UUID.randomUUID().toString();
+		output.send(MessageBuilder.withPayload(testPayload)
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				.build());
+
+		QueueChannel inbound1 = new QueueChannel();
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer(testDestination,
+				"test1", inbound1, createConsumerProperties());
+
+		Message<?> receivedMessage1 = receive(inbound1);
+		assertThat(receivedMessage1).isNotNull();
+		assertThat(new String((byte[]) receivedMessage1.getPayload()))
+				.isEqualTo(testPayload);
+
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
+	@Test
+	public void testTwoRequiredGroups(TestInfo testInfo) throws Exception {
+		B binder = getBinder();
+		PP producerProperties = createProducerProperties(testInfo);
+
+		DirectChannel output = createBindableChannel("output",
+				createProducerBindingProperties(producerProperties));
+
+		String testDestination = "testDestination"
+				+ UUID.randomUUID().toString().replace("-", "");
+
+		producerProperties.setRequiredGroups("test1", "test2");
+		Binding<MessageChannel> producerBinding = binder.bindProducer(testDestination,
+				output, producerProperties);
+
+		String testPayload = "foo-" + UUID.randomUUID().toString();
+		output.send(MessageBuilder.withPayload(testPayload)
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				.build());
+
+		QueueChannel inbound1 = new QueueChannel();
+		Binding<MessageChannel> consumerBinding1 = binder.bindConsumer(testDestination,
+				"test1", inbound1, createConsumerProperties());
+		QueueChannel inbound2 = new QueueChannel();
+		Binding<MessageChannel> consumerBinding2 = binder.bindConsumer(testDestination,
+				"test2", inbound2, createConsumerProperties());
+
+		Message<?> receivedMessage1 = receive(inbound1);
+		assertThat(receivedMessage1).isNotNull();
+		assertThat(new String((byte[]) receivedMessage1.getPayload()))
+				.isEqualTo(testPayload);
+		Message<?> receivedMessage2 = receive(inbound2);
+		assertThat(receivedMessage2).isNotNull();
+		assertThat(new String((byte[]) receivedMessage2.getPayload()))
+				.isEqualTo(testPayload);
+
+		consumerBinding1.unbind();
+		consumerBinding2.unbind();
+		producerBinding.unbind();
+	}
+
+	@Test
+	public void testPartitionedModuleSpEL(TestInfo testInfo) throws Exception {
+		B binder = getBinder();
+
+		CP consumerProperties = createConsumerProperties();
+		consumerProperties.setConcurrency(2);
+		consumerProperties.setInstanceIndex(0);
+		consumerProperties.setInstanceCount(3);
+		consumerProperties.setPartitioned(true);
+		QueueChannel input0 = new QueueChannel();
+		input0.setBeanName("test.input0S");
+		Binding<MessageChannel> input0Binding = binder.bindConsumer(
+				String.format("part%s0", getDestinationNameDelimiter()),
+				"testPartitionedModuleSpEL", input0, consumerProperties);
+		consumerProperties.setInstanceIndex(1);
+		QueueChannel input1 = new QueueChannel();
+		input1.setBeanName("test.input1S");
+		Binding<MessageChannel> input1Binding = binder.bindConsumer(
+				String.format("part%s0", getDestinationNameDelimiter()),
+				"testPartitionedModuleSpEL", input1, consumerProperties);
+		consumerProperties.setInstanceIndex(2);
+		QueueChannel input2 = new QueueChannel();
+		input2.setBeanName("test.input2S");
+		Binding<MessageChannel> input2Binding = binder.bindConsumer(
+				String.format("part%s0", getDestinationNameDelimiter()),
+				"testPartitionedModuleSpEL", input2, consumerProperties);
+
+		PP producerProperties = createProducerProperties(testInfo);
+		producerProperties.setPartitionKeyExpression(
+				spelExpressionParser.parseExpression("payload"));
+		producerProperties.setPartitionSelectorExpression(
+				spelExpressionParser.parseExpression("hashCode()"));
+		producerProperties.setPartitionCount(3);
+
+		DirectChannel output = createBindableChannel("output",
+				createProducerBindingProperties(producerProperties));
+		output.setBeanName("test.output");
+		Binding<MessageChannel> outputBinding = binder.bindProducer(
+				String.format("part%s0", getDestinationNameDelimiter()), output,
+				producerProperties);
+		try {
+			Object endpoint = extractEndpoint(outputBinding);
+			checkRkExpressionForPartitionedModuleSpEL(endpoint);
+		}
+		catch (UnsupportedOperationException ignored) {
+		}
+
+		Message<String> message2 = MessageBuilder.withPayload("2")
+				.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo")
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)
+				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build();
+		output.send(message2);
+		output.send(MessageBuilder.withPayload("1")
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				.build());
+		output.send(MessageBuilder.withPayload("0")
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				.build());
+
+		Message<?> receive0 = receive(input0);
+		assertThat(receive0).isNotNull();
+		Message<?> receive1 = receive(input1);
+		assertThat(receive1).isNotNull();
+		Message<?> receive2 = receive(input2);
+		assertThat(receive2).isNotNull();
+
+		Condition<Message<?>> correlationHeadersForPayload2 = new Condition<Message<?>>() {
+			@Override
+			public boolean matches(Message<?> value) {
+				IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor(
+						value);
+				return "foo".equals(accessor.getCorrelationId())
+						&& 42 == accessor.getSequenceNumber()
+						&& 43 == accessor.getSequenceSize();
+			}
+		};
+
+		if (usesExplicitRouting()) {
+			assertThat(receive0.getPayload()).isEqualTo("0".getBytes());
+			assertThat(receive1.getPayload()).isEqualTo("1".getBytes());
+			assertThat(receive2.getPayload()).isEqualTo("2".getBytes());
+			assertThat(receive2).has(correlationHeadersForPayload2);
+		}
+		else {
+			List<Message<?>> receivedMessages = Arrays.asList(receive0, receive1,
+					receive2);
+			assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(
+					"0".getBytes(), "1".getBytes(), "2".getBytes());
+			Condition<Message<?>> payloadIs2 = new Condition<Message<?>>() {
+
+				@Override
+				public boolean matches(Message<?> value) {
+					return value.getPayload().equals("2".getBytes());
+				}
+			};
+			assertThat(receivedMessages).filteredOn(payloadIs2).areExactly(1,
+					correlationHeadersForPayload2);
+
+		}
+		input0Binding.unbind();
+		input1Binding.unbind();
+		input2Binding.unbind();
+		outputBinding.unbind();
+	}
+
+	protected void checkRkExpressionForPartitionedModuleSpEL(Object endpoint) {
+		assertThat(getEndpointRouting(endpoint))
+				.contains(getExpectedRoutingBaseDestination(
+						String.format("part%s0", getDestinationNameDelimiter()), "test")
+						+ "-' + headers['partition']");
+	}
+
+	/**
+	 * Implementations should return whether the binder under test uses "explicit" routing
+	 * (e.g. Rabbit) whereby Spring Cloud Stream 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(Object endpoint) {
+		throw new UnsupportedOperationException();
+	}
+
+	/**
+	 * For implementations that rely on explicit routing, return the expected base
+	 * destination (the part that precedes '-partition' within the expression).
+	 */
+	protected String getExpectedRoutingBaseDestination(String name, String group) {
+		throw new UnsupportedOperationException();
+	}
+
+	protected abstract String getClassUnderTestName();
+
+	protected Lifecycle extractEndpoint(Binding<MessageChannel> binding) {
+		DirectFieldAccessor accessor = new DirectFieldAccessor(binding);
+		return (Lifecycle) accessor.getPropertyValue("lifecycle");
+	}
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/PartitionTestSupport.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2014-2019 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
+ *
+ *      https://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();
+	}
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/SerializableFoo.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2017-2019 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
+ *
+ *      https://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.Serializable;
+
+/**
+ * @author Oleg Zhurakousky
+ *
+ */
+public class SerializableFoo implements Serializable {
+
+	/**
+	 *
+	 */
+	private static final long serialVersionUID = 1L;
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/Spy.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2015-2019 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
+ *
+ *      https://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;
+
+/**
+ * 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 {
+
+	Object receive(boolean expectNull) throws Exception;
+
+}

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/TestUtils.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2015-2019 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
+ *
+ *      https://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.
+ *
+ * @author David Turanski
+ */
+public abstract 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;
+	}
+
+}

core/spring-cloud-stream-test-support/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,2 @@
+org.springframework.boot.env.EnvironmentPostProcessor=\
+  org.springframework.cloud.stream.binder.BinderTestEnvironmentPostProcessor