XD-3381: Provide test support for modules

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2015 the original author or authors.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.test.binder;
+
+import java.util.concurrent.BlockingQueue;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * Maintains a map between (output) channels and messages received (in FIFO order). To be injected in tests that
+ * can then run assertions on the enqueued messages.
+ *
+ * @author Eric Bottard
+ */
+public interface MessageCollector {
+
+	/**
+	 * Obtain a queue that will receive messages sent to the given channel.
+	 */
+	public BlockingQueue<Message<?>> forChannel(MessageChannel channel);
+}

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

@@ -0,0 +1,153 @@
+/*
+ * Copyright 2015 the original author or authors.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.test.binder;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.test.matcher.MessageQueueMatcher;
+import org.springframework.integration.channel.QueueChannel;
+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.util.Assert;
+
+/**
+ * A minimal binder that<ul>
+ *     <li>does nothing about binding consumers, leaving the channel as-is, so that a test author can interact with it directly,</li>
+ *     <li>registers a queue channel on the producer side, so that it is easy to assert what is received.</li>
+ * </ul>
+ *
+ * @author Eric Bottard
+ * @see MessageQueueMatcher
+ */
+public class TestSupportBinder implements Binder<MessageChannel> {
+
+	private final MessageCollectorImpl messageCollector = new MessageCollectorImpl();
+
+
+	@Override
+	public void bindConsumer(String name, MessageChannel inboundBindTarget, Properties properties) {
+	}
+
+	@Override
+	public void bindPubSubConsumer(String name, MessageChannel inboundBindTarget, Properties properties) {
+
+	}
+
+	/**
+	 * Registers a single subscriber to the channel, that enqueues messages for later retrieval and assertion in tests.
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public void bindProducer(String name, MessageChannel outboundBindTarget, Properties properties) {
+		final BlockingQueue queue = messageCollector.register(outboundBindTarget);
+		((SubscribableChannel)outboundBindTarget).subscribe(new MessageHandler() {
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				queue.add(message);
+			}
+		});
+
+	}
+
+	@Override
+	public void unbindProducer(String name, MessageChannel channel) {
+		messageCollector.unregister(channel);
+	}
+
+	@Override
+	public void bindPubSubProducer(String name, MessageChannel outboundBindTarget, Properties properties) {
+
+	}
+
+	@Override
+	public void unbindConsumers(String name) {
+
+	}
+
+	@Override
+	public void unbindProducers(String name) {
+
+	}
+
+	@Override
+	public void unbindConsumer(String name, MessageChannel inboundBindTarget) {
+
+	}
+
+	@Override
+	public void bindRequestor(String name, MessageChannel requests, MessageChannel replies, Properties properties) {
+
+	}
+
+	@Override
+	public void bindReplier(String name, MessageChannel requests, MessageChannel replies, Properties properties) {
+
+	}
+
+	@Override
+	public MessageChannel bindDynamicProducer(String name, Properties properties) {
+		return null;
+	}
+
+	@Override
+	public MessageChannel bindDynamicPubSubProducer(String name, Properties properties) {
+		return null;
+	}
+
+	@Override
+	public boolean isCapable(Capability capability) {
+		return false;
+	}
+
+	public MessageCollector messageCollector() {
+		return messageCollector;
+	}
+
+	/**
+	 * Maintains mappings between channels and queues.
+	 *
+	 * @author Eric Bottard
+	 */
+	private static class MessageCollectorImpl implements MessageCollector{
+
+		private Map<MessageChannel, BlockingQueue<Message<?>>> results = new HashMap<>();
+
+		private BlockingQueue register(MessageChannel channel) {
+			LinkedBlockingDeque<Message<?>> result = new LinkedBlockingDeque<>();
+			Assert.isTrue(!results.containsKey(channel), "Channel [" + channel + "] was already bound");
+			results.put(channel, result);
+			return result;
+		}
+
+		private void unregister(MessageChannel channel) {
+			Assert.notNull(results.remove(channel), "Trying to unregister a mapping for an unknown channel [" + channel + "]");
+		}
+
+		public BlockingQueue<Message<?>> forChannel(MessageChannel channel) {
+			BlockingQueue<Message<?>> queue = results.get(channel);
+			Assert.notNull(queue, "Channel [" + channel + "] was not bound by " + TestSupportBinder.class);
+			return queue;
+		}
+	}
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2015 the original author or authors.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.test.binder;
+
+import org.springframework.boot.autoconfigure.AutoConfigureOrder;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.Ordered;
+
+/**
+ * Installs the {@link TestSupportBinder} and exposes {@link TestSupportBinder.MessageCollectorImpl} to be injected in tests.
+ *
+ * Note that this auto-configuration has higher priority than regular binders, so adding
+ * this on the classpath in test scope is sufficient to have support kick in.
+ *
+ * @author Eric Bottard
+ */
+@Configuration
+@AutoConfigureOrder(Ordered.HIGHEST_PRECEDENCE)
+public class TestSupportBinderAutoConfiguration {
+
+	@Bean
+	public MessageCollector messageCollector() {
+		return testSupportBinder().messageCollector();
+	}
+
+	@Bean
+	public TestSupportBinder testSupportBinder() {
+		return new TestSupportBinder();
+	}
+
+}

spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/matcher/MessageQueueMatcher.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright 2015 the original author or authors.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.test.matcher;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.SelfDescribing;
+
+import org.springframework.cloud.stream.test.binder.TestSupportBinder;
+import org.springframework.integration.util.Function;
+import org.springframework.messaging.Message;
+
+/**
+ * A Hamcrest Matcher meant to be used in conjunction with {@link TestSupportBinder}.
+ *
+ * <p>Expected usage is of the form (with appropriate static imports):
+ * <pre>
+ * public class TransformProcessorApplicationTests {
+ *
+ *    {@literal @}Autowired
+ *    {@literal @}ModuleChannels(TransformProcessor.class)
+ *    private Processor processor;
+ *
+ *    {@literal @}Autowired
+ *    private MessageCollectorImpl messageCollector;
+ *
+ *
+ *    {@literal @}Test
+ *    public void testUsingExpression() {
+ *        processor.input().send(new GenericMessage{@literal <}Object>("hello"));
+ *        assertThat(messageCollector.forChannel(processor.output()), receivesPayloadThat(is("hellofoo")).within(10));
+ *    }
+ *
+ * }</pre>
+ * </p>
+ *
+ * @author Eric Bottard
+ */
+public class MessageQueueMatcher<T> extends BaseMatcher<BlockingQueue<Message<?>>> {
+
+	private final Matcher<T> delegate;
+
+	private final long timeout;
+
+	private Extractor<Message<?>, T> extractor;
+
+	private Map<BlockingQueue<Message<?>>, T> actuallyReceived = new HashMap<>();
+
+	private final TimeUnit unit;
+
+	public MessageQueueMatcher(Matcher<T> delegate, long timeout, TimeUnit unit, Extractor<Message<?>, T> extractor) {
+		this.delegate = delegate;
+		this.timeout = timeout;
+		this.unit = unit;
+		this.extractor = extractor;
+	}
+
+
+	@Override
+	public boolean matches(Object item) {
+		@SuppressWarnings("unchecked")
+		BlockingQueue<Message<?>> queue = (BlockingQueue<Message<?>>) item;
+		Message<?> received = null;
+		try {
+			if (timeout > 0) {
+				received = queue.poll(timeout, unit);
+			} else if (timeout == 0) {
+				received = queue.poll();
+			} else {
+				received = queue.take();
+			}
+		}
+		catch (InterruptedException e) {
+			return false;
+		}
+		T unwrapped = extractor.apply(received);
+		actuallyReceived.put(queue, unwrapped);
+		return delegate.matches(unwrapped);
+	}
+
+	@Override
+	public void describeMismatch(Object item, Description description) {
+		@SuppressWarnings("unchecked")
+		BlockingQueue<Message<?>> queue = (BlockingQueue<Message<?>>) item;
+		T value = actuallyReceived.get(queue);
+		if (value != null) {
+			description.appendText("received: ").appendValue(value);
+		} else {
+			description.appendText("timed out after " + timeout + " " + unit.name().toLowerCase());
+		}
+	}
+
+	public MessageQueueMatcher<T> within(long timeout, TimeUnit unit) {
+		return new MessageQueueMatcher<>(this.delegate, timeout, unit, this.extractor);
+	}
+
+	@Override
+	public void describeTo(Description description) {
+		description.appendText("Channel to receive ").appendDescriptionOf(extractor).appendDescriptionOf(delegate);
+	}
+
+	@SuppressWarnings("unchecked")
+	public static <P> MessageQueueMatcher<P> receivesMessageThat(Matcher<Message<P>> messageMatcher) {
+		return new MessageQueueMatcher(messageMatcher, -1, null, new Extractor<Message<P>, Message<P>>("a message that ") {
+			@Override
+			public Message<P> apply(Message<P> m) {
+				return m;
+			}
+		});
+	}
+
+	@SuppressWarnings("unchecked")
+	public static <P> MessageQueueMatcher<P> receivesPayloadThat(Matcher<P> payloadMatcher) {
+		return new MessageQueueMatcher(payloadMatcher, -1, null, new Extractor<Message<P>, P>("a message whose payload ") {
+			@Override
+			public P apply(Message<P> m) {
+				return m.getPayload();
+			}
+		});
+	}
+
+	/**
+	 * A transformation to be applied to a received message before asserting, <i>e.g.</i> to only inspect the payload.
+	 */
+	public static abstract class Extractor<R, T> implements Function<R, T>, SelfDescribing {
+
+		private final String behaviorDescription;
+
+		protected Extractor(String behaviorDescription) {
+			this.behaviorDescription = behaviorDescription;
+		}
+
+		@Override
+		public void describeTo(Description description) {
+			description.appendText(behaviorDescription);
+		}
+	}
+
+
+
+}

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

@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration:\
+org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration