GH-1153 Fixed consumer to honor channel-bound contentType

- Added SpringIntegration backed test binder
- Added initial binder test suite

Resolves #1153

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java

@@ -225,7 +225,9 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea
 		@Override
 		public Message<?> preSend(Message<?> message, MessageChannel channel) {
 			Message<?> sentMessage = null;
-			if (this.klazz.isAssignableFrom(message.getPayload().getClass())) {
+			if (this.klazz.isAssignableFrom(message.getPayload().getClass()) ||
+					(this.klazz.isAssignableFrom(String.class) && message.getPayload() instanceof byte[]
+							&& !message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE) && !this.contentType.equals("text/plain"))) {
 				Object contentTypeFromMessage = message.getHeaders().get(MessageHeaders.CONTENT_TYPE);
 				if (contentTypeFromMessage == null) {
 					sentMessage = MessageConverterConfigurer.this.messageBuilderFactory

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/contenttype/BinderConversionTests.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.contenttype;
+
+import java.nio.charset.StandardCharsets;
+
+import org.junit.Test;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.integration.SourceDestination;
+import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration;
+import org.springframework.cloud.stream.binder.integration.TargetDestination;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.handler.annotation.SendTo;
+import org.springframework.messaging.support.GenericMessage;
+
+/**
+ * Sort of a TCK test suite to validate payload conversion is
+ * done properly by interacting with binder's input/output destinations
+ * instead of its bridged channels.
+ * This means that all payloads (sent/received) must be expressed in the
+ *  wire format (byte[])
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class BinderConversionTests {
+	@Test
+	public void test() {
+		ApplicationContext context = new SpringApplicationBuilder(ApplicationJsonDefaultType.class).web(false)
+				.run("--spring.cloud.stream.default.contentType=application/json", "--spring.jmx.enabled=false");
+		SourceDestination source = context.getBean(SourceDestination.class);
+		TargetDestination target = context.getBean(TargetDestination.class);
+		String jsonPayload = "{\"name\":\"oleg\"}";
+		source.send(new GenericMessage<byte[]>(jsonPayload.getBytes()));
+		System.out.println(new String((byte[])target.receive().getPayload(), StandardCharsets.UTF_8));
+	}
+
+	@SpringBootApplication
+	@EnableBinding(Processor.class)
+	@Import(SpringIntegrationBinderConfiguration.class)
+	public static class ApplicationJsonDefaultType {
+		@StreamListener(Processor.INPUT)
+		@SendTo(Processor.OUTPUT)
+		public Person echo(Person value) {
+			return value;
+		}
+
+		public static class Person {
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+		}
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/AbstractDestination.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.integration;
+
+import org.springframework.messaging.SubscribableChannel;
+
+/**
+ * @author Oleg Zhurakousky
+ *
+ */
+abstract class AbstractDestination {
+
+	private SubscribableChannel channel;
+
+	SubscribableChannel getChannel() {
+		return channel;
+	}
+
+	void setChannel(SubscribableChannel channel) {
+		this.channel = channel;
+		this.afterChannelIsSet();
+	}
+
+	void afterChannelIsSet() {
+		// noop
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.integration;
+
+import java.nio.charset.StandardCharsets;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.annotation.SendTo;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Sample spring cloud stream application that demonstrates the usage of {@link SpringIntegrationChannelBinder}.
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+@SpringBootApplication
+@EnableBinding(Processor.class)
+@Import(SpringIntegrationBinderConfiguration.class)
+public class SampleStreamApp {
+
+	public static void main(String[] args) {
+		ApplicationContext context = new SpringApplicationBuilder(SampleStreamApp.class).web(false)
+				.run("--server.port=0");
+		SourceDestination source = context.getBean(SourceDestination.class);
+		TargetDestination target = context.getBean(TargetDestination.class);
+		source.send(new GenericMessage<byte[]>("Hello".getBytes()));
+		Message<?> message = target.receive();
+		assertEquals("Hello", new String((byte[])message.getPayload(), StandardCharsets.UTF_8));
+		System.out.println();
+	}
+
+	@StreamListener(Processor.INPUT)
+	@SendTo(Processor.OUTPUT)
+	public String receive(String value) {
+		System.out.println("Handling payload: " + value);
+		return value;
+	}
+
+	@ServiceActivator(inputChannel="input.anonymous.errors")
+	public void error(String value) {
+		System.out.println("Handling ERROR payload: " + value);
+	}
+}
+
+

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SourceDestination.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.integration;
+
+import org.springframework.messaging.Message;
+
+/**
+ * Implementation of binder endpoint that represents the source destination
+ * (e.g., destination from which messages will be received by Processor.INPUT).
+ * <br>
+ * You can interact with it by calling {@link #send(Message)} operation.
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class SourceDestination extends AbstractDestination {
+
+	/**
+	 * Allows the {@link Message} to be sent to a Binder to be delegated
+	 * to binder's input destination (e.g., Processor.INPUT).
+	 *
+	 */
+	public void send(Message<?> message) {
+		this.getChannel().send(message);
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.integration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.BinderType;
+import org.springframework.cloud.stream.binder.BinderTypeRegistry;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.integration.config.EnableIntegration;
+
+/**
+ * {@link Binder} configuration backed by Spring Integration.
+ *
+ * Please see {@link SpringIntegrationChannelBinder} for more details.
+ *
+ * @author Oleg Zhurakousky
+ *
+ * @see SpringIntegrationChannelBinder
+ */
+@Configuration
+@ConditionalOnMissingBean(Binder.class)
+@EnableIntegration
+public class SpringIntegrationBinderConfiguration<T> {
+
+	public static final String NAME = "integration";
+
+	/**
+	 * Utility operation to return an array of configuration classes
+	 * defined in {@link EnableBinding} annotation.
+	 * Typically used for tests that do not rely on creating an SCSt boot
+	 * application annotated with {@link EnableBinding}, yet require
+	 * full {@link Binder} configuration.
+	 */
+	public static Class<?>[] getCompleteConfiguration() {
+		List<Class<?>> configClasses = new ArrayList<>();
+		configClasses.add(SpringIntegrationBinderConfiguration.class);
+		Import annotation = AnnotationUtils.getAnnotation(EnableBinding.class, Import.class);
+		Map<String, Object> annotationAttributes = AnnotationUtils.getAnnotationAttributes(annotation);
+		configClasses.addAll(Arrays.asList((Class<?>[])annotationAttributes.get("value")));
+		return configClasses.toArray(new Class<?>[] {});
+	}
+
+	@Bean
+	public BinderTypeRegistry binderTypeRegistry() {
+		BinderType binderType = new BinderType(NAME, new Class[] {SpringIntegrationBinderConfiguration.class});
+		BinderTypeRegistry btr = new DefaultBinderTypeRegistry(Collections.singletonMap(NAME, binderType));
+		return btr;
+	}
+
+	@Bean
+	public SourceDestination sourceDestination() {
+		return new SourceDestination();
+	}
+
+	@Bean
+	public TargetDestination targetDestination() {
+		return new TargetDestination();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Bean
+	public Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties> springIntegrationChannelBinder(SpringIntegrationProvisioner provisioner) {
+		return (Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties>) new SpringIntegrationChannelBinder(provisioner);
+	}
+
+	@Bean
+	public SpringIntegrationProvisioner springIntegrationProvisioner() {
+		return new SpringIntegrationProvisioner();
+	}
+
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java

@@ -0,0 +1,267 @@
+/*
+ * Copyright 2015-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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.integration;
+
+import java.util.function.Consumer;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner.SpringIntegrationConsumerDestination;
+import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner.SpringIntegrationProducerDestination;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.core.AttributeAccessor;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.handler.BridgeHandler;
+import org.springframework.integration.support.DefaultErrorMessageStrategy;
+import org.springframework.integration.support.ErrorMessageStrategy;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.RetryCallback;
+import org.springframework.retry.RetryContext;
+import org.springframework.retry.RetryListener;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Implementation of {@link Binder} backed by Spring Integration framework.
+ * It is useful for localized demos and testing.
+ * <p>
+ * This binder extends from the same base class ({@link AbstractMessageChannelBinder}) as
+ * other binders (i.e., Rabbit, Kafka etc). Interaction with this binder is done via source
+ * and target destination which emulate real binder's destinations (i.e., Kafka topic)
+ * <br>
+ * The destination classes are
+ * <ul>
+ * <li>{@link SourceDestination}</li>
+ * <li>{@link TargetDestination}</li>
+ * </ul>
+ * Simply autowire them in your your application and send/receive messages.
+ * </p>
+ * You must also add {@link SpringIntegrationBinderConfiguration} to your configuration.
+ * Below is the example using Spring Boot test.
+ * <pre class="code">
+ *
+ * &#064;RunWith(SpringJUnit4ClassRunner.class)
+ * &#064;SpringBootTest(classes = {SpringIntegrationBinderConfiguration.class, TestWithSIBinder.MyProcessor.class})
+ * public class TestWithSIBinder {
+ *     &#064;Autowired
+ *     private SourceDestination sourceDestination;
+ *
+ *     &#064;Autowired
+ *     private TargetDestination targetDestination;
+ *
+ *     &#064;Test
+ *     public void testWiring() {
+ *         sourceDestination.send(new GenericMessage<String>("Hello"));
+ *         assertEquals("Hello world", new String((byte[])targetDestination.receive().getPayload(), StandardCharsets.UTF_8));
+ *     }
+ *
+ *     &#064;SpringBootApplication
+ *     &#064;EnableBinding(Processor.class)
+ *     public static class MyProcessor {
+ *         &#064;StreamListener(Processor.INPUT)
+ *         &#064;SendTo(Processor.OUTPUT)
+ *         public String transform(String in) {
+ *             return in + " world";
+ *         }
+ *     }
+ * }
+ * </pre>
+ *
+ * @author Oleg Zhurakousky
+ */
+class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder<ConsumerProperties,
+	ProducerProperties, SpringIntegrationProvisioner> {
+
+	@Autowired
+	private BeanFactory beanFactory;
+
+	SpringIntegrationChannelBinder(SpringIntegrationProvisioner provisioningProvider) {
+		super(true, new String[] {}, provisioningProvider);
+	}
+
+	@Override
+	protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
+			ProducerProperties producerProperties, MessageChannel errorChannel) throws Exception {
+		BridgeHandler handler = new BridgeHandler();
+		handler.setBeanFactory(this.beanFactory);
+		handler.setOutputChannel(((SpringIntegrationProducerDestination)destination).getChannel());
+		return handler;
+	}
+
+	@Override
+	protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, ConsumerProperties properties)
+			throws Exception {
+		ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
+		SubscribableChannel siBinderInputChannel = ((SpringIntegrationConsumerDestination)destination).getChannel();
+
+		IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer();
+		IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(messageListenerContainer);
+
+		String groupName = StringUtils.hasText(group) ? group : "anonymous";
+		ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, groupName, properties);
+		if (properties.getMaxAttempts() > 1) {
+			adapter.setRetryTemplate(buildRetryTemplate(properties));
+			adapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
+		}
+		else {
+			adapter.setErrorMessageStrategy(errorMessageStrategy);
+			adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
+		}
+
+		siBinderInputChannel.subscribe(messageListenerContainer);
+
+		return adapter;
+	}
+
+	/**
+	 * Implementation of simple message listener container modeled after AMQP SimpleMessageListenerContainer
+	 */
+	private static class IntegrationMessageListeningContainer implements MessageHandler {
+
+		private Consumer<Message<?>> listener;
+
+		@Override
+		public void handleMessage(Message<?> message) throws MessagingException {
+			this.listener.accept(message);
+		}
+
+		public void setMessageListener(Consumer<Message<?>> listener) {
+			this.listener = listener;
+		}
+	}
+
+	/**
+	 * Implementation of inbound channel adapter modeled after AmqpInboundChannelAdapter
+	 */
+	private static class IntegrationBinderInboundChannelAdapter extends MessageProducerSupport {
+
+		private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
+
+		private final IntegrationMessageListeningContainer listenerContainer;
+
+		private RetryTemplate retryTemplate;
+
+		private RecoveryCallback<? extends Object> recoveryCallback;
+
+		IntegrationBinderInboundChannelAdapter(IntegrationMessageListeningContainer listenerContainer) {
+			this.listenerContainer = listenerContainer;
+		}
+
+		@SuppressWarnings("unused")
+		// Temporarily unused until DLQ strategy for this binder becomes a requirement
+		public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
+			this.recoveryCallback = recoveryCallback;
+		}
+
+		public void setRetryTemplate(RetryTemplate retryTemplate) {
+			this.retryTemplate = retryTemplate;
+		}
+
+		@Override
+		protected void onInit() {
+			if (this.retryTemplate != null) {
+				Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+						+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+						+ "send an error message when retries are exhausted");
+			}
+			Listener messageListener = new Listener();
+			if (this.retryTemplate != null) {
+				this.retryTemplate.registerListener(messageListener);
+			}
+			this.listenerContainer.setMessageListener(messageListener);
+		}
+
+		protected class Listener implements RetryListener, Consumer<Message<?>> {
+
+			@Override
+			@SuppressWarnings({ "unchecked", "rawtypes" })
+			public void accept(final Message<?> message) {
+				try {
+					if (IntegrationBinderInboundChannelAdapter.this.retryTemplate == null) {
+						try {
+							processMessage(message);
+						}
+						finally {
+							attributesHolder.remove();
+						}
+					}
+					else {
+						try {
+							IntegrationBinderInboundChannelAdapter.this.retryTemplate.execute(new RetryCallback() {
+
+								@Override
+								public Object doWithRetry(RetryContext context) throws Throwable {
+									processMessage(message);
+									return null;
+								}
+							},(RecoveryCallback<Object>) IntegrationBinderInboundChannelAdapter.this.recoveryCallback);
+						}
+						catch (Throwable e) {
+							// TODO Auto-generated catch block
+							e.printStackTrace();
+						}
+					}
+				}
+				catch (RuntimeException e) {
+					if (getErrorChannel() != null) {
+						getMessagingTemplate().send(getErrorChannel(), buildErrorMessage(null,
+								new IllegalStateException("Message conversion failed: " + message, e)));
+					}
+					else {
+						throw e;
+					}
+				}
+			}
+
+			private void processMessage(Message<?> message) {
+				sendMessage(message);
+			}
+
+			@Override
+			public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
+				if (IntegrationBinderInboundChannelAdapter.this.recoveryCallback != null) {
+					attributesHolder.set(context);
+				}
+				return true;
+			}
+
+			@Override
+			public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
+					Throwable throwable) {
+				attributesHolder.remove();
+			}
+
+			@Override
+			public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
+					Throwable throwable) {
+				// Empty
+			}
+		}
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.integration;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.cloud.stream.provisioning.ProvisioningException;
+import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.SubscribableChannel;
+
+/**
+ * {@link ProvisioningProvider} to support {@link SpringIntegrationChannelBinder}. It
+ * exists primarily to support {@link AbstractMessageChannel} semantics for creating
+ * {@link ConsumerDestination} and {@link ProducerDestination}, to interact with this
+ * {@link Binder}.
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+class SpringIntegrationProvisioner implements ProvisioningProvider<ConsumerProperties, ProducerProperties> {
+
+	private final Map<String, SubscribableChannel> provisionedDestinations = new HashMap<>();
+
+	@Autowired
+	private SourceDestination source;
+
+	@Autowired
+	private TargetDestination target;
+
+	/**
+	 * Will provision producer destination as an SI {@link PublishSubscribeChannel}.
+	 * <br>
+	 * This provides convenience of registering additional subscriber (handler in the test method)
+	 * along side of being able to call {@link TargetDestination#receive()} to get a
+	 * {@link Message} for additional assertions.
+	 */
+	@Override
+	public ProducerDestination provisionProducerDestination(String name, ProducerProperties properties) throws ProvisioningException {
+		SubscribableChannel destination = this.provisionDestination(name, true);
+		this.target.setChannel(destination);
+		return new SpringIntegrationProducerDestination(name, destination);
+	}
+
+	/**
+	 * Will provision consumer destination as SI {@link DirectChannel}
+	 */
+	@Override
+	public ConsumerDestination provisionConsumerDestination(String name, String group, ConsumerProperties properties) throws ProvisioningException {
+		SubscribableChannel destination = this.provisionDestination(name, false);
+		this.source.setChannel(destination);
+		return new SpringIntegrationConsumerDestination(name, destination);
+	}
+
+	private SubscribableChannel provisionDestination(String name, boolean pubSub) {
+		String destinationName = name + ".destination";
+		SubscribableChannel destination = this.provisionedDestinations.get(destinationName);
+		if (destination == null) {
+			destination = pubSub ? new PublishSubscribeChannel() : new DirectChannel();
+			((AbstractMessageChannel)destination).setBeanName(destinationName);
+			((AbstractMessageChannel)destination).setComponentName(destinationName);
+			this.provisionedDestinations.put(destinationName, destination);
+		}
+		return destination;
+	}
+
+	class SpringIntegrationConsumerDestination implements ConsumerDestination {
+		private final String name;
+		private final SubscribableChannel channel;
+
+		SpringIntegrationConsumerDestination(String name, SubscribableChannel channel) {
+			this.name = name;
+			this.channel = channel;
+		}
+
+		public SubscribableChannel getChannel() {
+			return this.channel;
+		}
+
+		@Override
+		public String getName() {
+			return this.name;
+		}
+	}
+
+	class SpringIntegrationProducerDestination implements ProducerDestination {
+		private final String name;
+		private final SubscribableChannel channel;
+
+		SpringIntegrationProducerDestination(String name, SubscribableChannel channel) {
+			this.name = name;
+			this.channel = channel;
+		}
+
+		@Override
+		public String getNameForPartition(int partition) {
+			return this.getName() + partition;
+		}
+
+		public SubscribableChannel getChannel() {
+			return this.channel;
+		}
+
+		@Override
+		public String getName() {
+			return this.name;
+		}
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/TargetDestination.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.integration;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedTransferQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+
+/**
+ * Implementation of binder endpoint that represents the target destination
+ * (e.g., destination which receives messages sent to Processor.OUTPUT)
+ * <br>
+ * You can interact with it by calling {@link #receive()} operation.
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class TargetDestination extends AbstractDestination {
+
+	private BlockingQueue<Message<?>> messages;
+
+	/**
+	 * Allows to access {@link Message}s received by this {@link TargetDestination}.
+	 * @param timeout how long to wait before giving up
+	 */
+	public Message<?> receive(long timeout) {
+		try {
+			return this.messages.poll(timeout, TimeUnit.MILLISECONDS);
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
+		return null;
+	}
+
+	/**
+	 * Allows to access {@link Message}s received by this {@link TargetDestination}.
+	 */
+	public Message<?> receive() {
+		return this.receive(0);
+	}
+
+	@Override
+	void afterChannelIsSet() {
+		this.messages = new LinkedTransferQueue<>();
+		this.getChannel().subscribe(new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				messages.offer(message);
+			}
+		});
+	}
+}