Some improvement for testing sample

* Add JavaDocs
* Demonstrate more test utilities usage and testing approaches

README.adoc

@@ -13,7 +13,7 @@ To build the samples do:
 
 * `dynamic-source` publishes messages to dynamically created destinations.
 
-* `kinesis-produce-consume` An example application using spring-cloud-stream-binder-aws-kinesis. Presents a web endpoint to send Orders, these are placed on a Kinesis stream and then consumed by the application from that stream. 
+* `kinesis-produce-consume` An example application using spring-cloud-stream-binder-aws-kinesis. Presents a web endpoint to send Orders, these are placed on a Kinesis stream and then consumed by the application from that stream.
 
 * `multi-io` shows how to use configure multiple input/output channels inside a single application.
 
@@ -40,3 +40,5 @@ We generally recommend testing with the http://docs.spring.io/spring-cloud-strea
 
 * `kstream` is a collection of applications that demonstrate the capabilities of the Spring Cloud Stream support for Kafka Streams
 
+* `testing` is a bunch of applications and tests for them to demonstrate the capabilities of testing for the the Spring Cloud Stream applications.
+

testing/src/main/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessor.java

@@ -24,6 +24,9 @@ import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.messaging.handler.annotation.SendTo;
 
 /**
+ * The Spring Cloud Stream Processor application,
+ * which convert incoming String to its upper case representation.
+ *
  * @author Artem Bilan
  *
  */

testing/src/main/java/org/springframework/cloud/stream/testing/sink/JdbcSink.java

@@ -28,6 +28,9 @@ import org.springframework.integration.jdbc.JdbcMessageHandler;
 import org.springframework.messaging.MessageHandler;
 
 /**
+ * The Spring Cloud Stream Sink application,
+ * which insert payloads of incoming messages to the FOOBAR table in the RDBMS.
+ *
  * @author Artem Bilan
  *
  */
@@ -37,7 +40,7 @@ public class JdbcSink {
 
 	@Bean
 	@ServiceActivator(inputChannel = Sink.INPUT)
-	public MessageHandler logHandler(DataSource dataSource) {
+	public MessageHandler jdbcHandler(DataSource dataSource) {
 		return new JdbcMessageHandler(dataSource, "INSERT INTO foobar (value) VALUES (:payload)");
 	}
 

testing/src/main/java/org/springframework/cloud/stream/testing/source/FooBarSource.java

@@ -29,6 +29,9 @@ import org.springframework.integration.core.MessageSource;
 import org.springframework.messaging.support.GenericMessage;
 
 /**
+ * The Spring Cloud Stream Source application,
+ * which generates each 100 milliseconds "foo" or "bar" string in round-robin manner.
+ *
  * @author Artem Bilan
  *
  */

testing/src/test/java/org/springframework/cloud/stream/testing/processor/NaiveToUpperCaseTests.java

@@ -21,6 +21,9 @@ import static org.junit.Assert.assertEquals;
 import org.junit.Test;
 
 /**
+ * The simple test-case to demonstrate how it is easy to apply unit testing assertion
+ * for te Spring Cloud Stream applications.
+ *
  * @author Artem Bilan
  *
  */

testing/src/test/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessorTests.java

@@ -16,17 +16,19 @@
 
 package org.springframework.cloud.stream.testing.processor;
 
-
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.not;
 import static org.junit.Assert.assertThat;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesMessageThat;
 import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesPayloadThat;
+import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
 
 import java.util.concurrent.BlockingQueue;
 
+import org.hamcrest.Matcher;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -35,12 +37,17 @@ import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.cloud.stream.test.binder.MessageCollector;
+import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
+ * The Spring Boot-base test-case to demonstrate how can we test Spring Cloud Stream applications
+ * with available testing tools.
+ *
  * @author Artem Bilan
  *
  */
@@ -59,11 +66,14 @@ public class ToUpperCaseProcessorTests {
 	private ToUpperCaseProcessor toUpperCaseProcessor;
 
 	@Test
+	@SuppressWarnings("unchecked")
 	public void testMessages() {
-		this.channels.input().send(new GenericMessage<>("foo"));
-		this.channels.input().send(new GenericMessage<>("bar"));
-		this.channels.input().send(new GenericMessage<>("foo meets bar"));
-		this.channels.input().send(new GenericMessage<>("nothing but the best test"));
+		SubscribableChannel input = this.channels.input();
+
+		input.send(new GenericMessage<>("foo"));
+		input.send(new GenericMessage<>("bar"));
+		input.send(new GenericMessage<>("foo meets bar"));
+		input.send(new GenericMessage<>("nothing but the best test"));
 
 		BlockingQueue<Message<?>> messages = this.collector.forChannel(channels.output());
 
@@ -72,7 +82,24 @@ public class ToUpperCaseProcessorTests {
 		assertThat(messages, receivesPayloadThat(is("FOO MEETS BAR")));
 		assertThat(messages, receivesPayloadThat(not("nothing but the best test")));
 
-		verify(this.toUpperCaseProcessor, times(4)).transform(anyString());
+		Message<String> testMessage =
+				MessageBuilder.withPayload("headers")
+						.setHeader("foo", "bar")
+						.build();
+
+		input.send(testMessage);
+
+		Message<String> expected =
+				MessageBuilder.withPayload("HEADERS")
+						.copyHeaders(testMessage.getHeaders())
+						.build();
+
+		Matcher<Message<Object>> sameExceptIgnorableHeaders =
+				(Matcher<Message<Object>>) (Matcher<?>) sameExceptIgnorableHeaders(expected);
+
+		assertThat(messages, receivesMessageThat(sameExceptIgnorableHeaders));
+
+		verify(this.toUpperCaseProcessor, times(5)).transform(anyString());
 	}
 
 }

testing/src/test/java/org/springframework/cloud/stream/testing/processor/integration/ToUpperCaseProcessorIntTests.java

@@ -16,7 +16,6 @@
 
 package org.springframework.cloud.stream.testing.processor.integration;
 
-
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Iterator;
@@ -40,6 +39,9 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
+ * A Spring Boot integration test for the Spring Cloud Stream Processor application
+ * based on the Embedded Kafka.
+ *
  * @author Artem Bilan
  *
  */

testing/src/test/java/org/springframework/cloud/stream/testing/sink/JdbcSinkTests.java

@@ -18,22 +18,35 @@ package org.springframework.cloud.stream.testing.sink;
 
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.ChannelInterceptorAdapter;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
+ * The Spring Boot-base test-case to demonstrate how can we test Spring Cloud Stream applications
+ * with available testing tools.
+ *
  * @author Artem Bilan
  *
  */
@@ -48,15 +61,51 @@ public class JdbcSinkTests {
 	@Autowired
 	private JdbcTemplate jdbcTemplate;
 
+	@SpyBean(name = "jdbcHandler")
+	private MessageHandler jdbcMessageHandler;
+
 	@Test
+	@SuppressWarnings("unchecked")
 	public void testMessages() {
-		this.channels.input().send(new GenericMessage<>("foo"));
-		this.channels.input().send(new GenericMessage<>("bar"));
+		AbstractMessageChannel input = (AbstractMessageChannel) this.channels.input();
+
+		final AtomicReference<Message<?>> messageAtomicReference = new AtomicReference<>();
+
+		ChannelInterceptorAdapter assertionInterceptor = new ChannelInterceptorAdapter() {
+
+			@Override
+			public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
+				messageAtomicReference.set(message);
+				super.afterSendCompletion(message, channel, sent, ex);
+			}
+
+		};
+
+		input.addInterceptor(assertionInterceptor);
+
+		input.send(new GenericMessage<>("foo"));
+
+		input.removeInterceptor(assertionInterceptor);
+
+		input.send(new GenericMessage<>("bar"));
 
 		List<Map<String, Object>> data = this.jdbcTemplate.queryForList("SELECT * FROM foobar");
+
 		assertThat(data.size()).isEqualTo(2);
 		assertThat(data.get(0).get("value")).isEqualTo("foo");
 		assertThat(data.get(1).get("value")).isEqualTo("bar");
+
+		Message<?> message1 = messageAtomicReference.get();
+		assertThat(message1).isNotNull();
+		assertThat(message1).hasFieldOrPropertyWithValue("payload", "foo");
+
+		ArgumentCaptor<Message<?>> messageArgumentCaptor =
+				(ArgumentCaptor<Message<?>>) (ArgumentCaptor<?>) ArgumentCaptor.forClass(Message.class);
+
+		verify(this.jdbcMessageHandler, times(2)).handleMessage(messageArgumentCaptor.capture());
+
+		Message<?> message = messageArgumentCaptor.getValue();
+		assertThat(message).hasFieldOrPropertyWithValue("payload", "bar");
 	}
 
 }

testing/src/test/java/org/springframework/cloud/stream/testing/source/FooBarSourceTests.java

@@ -18,9 +18,7 @@ package org.springframework.cloud.stream.testing.source;
 
 
 import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.CoreMatchers.not;
 import static org.junit.Assert.assertThat;
-import static org.mockito.Mockito.verify;
 import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesPayloadThat;
 
 import java.util.concurrent.BlockingQueue;
@@ -37,6 +35,9 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
+ * The Spring Boot-base test-case to demonstrate how can we test Spring Cloud Stream applications
+ * with available testing tools.
+ *
  * @author Artem Bilan
  *
  */