Add MockMessageHandler to the Testing Framework

Fix PayloadMatcherTests for generics

Address PR comments and other improvements

* Revert `rawtypes` mode for the `PayloadMatcher`
* Make `HeaderMatcher` as `rawtypes` as well
* Make `MockMessageHandler` expect `rawtypes` for `Matcher`s.
This way we can just support `Matcher`s like `notNullValue(Message.class)`
* Rename `expect()` to `assertNext()`
* Rename `andReply()` to `thenReply()`
* Track replies are supplied in the `MockMessageHandler`
* Distinguish simple `MH` from the `MP` types in the
`MockIntegrationContext#instead()` do not let to replace simple `MH`
with fully configured `MockMessageHandler` or any other `MP` implementation.
Fail replace if types mismatch; wrap `MockMessageHandler` to simple `MH`
if it doesn't have replies when we are going to replace simple `MH`
* Wrap `MockMessageHandler` to the `Mockito.spy()` in the
`MockIntegration#mockMessageHandler()` to allow to `verify()` interaction
in the test-case

Remove wrapping `MockMH` to raw `MH` when no reply supported.
If `MockMH` isn't supplied with replies ti's safe to use it as is - no harm to target endpoint
which supposed to be last one in the flow

Some polishing and JavaDocs

More JavaDocs

Add docs for the `MockMessageHandler` and fix some JavaDocs

Make the `MockMessageHandler` with an API like:
```
MockIntegration.mockMessageHandler()
             .handleNext(Consumer<Message<?>>)
             .handleNext(Consumer<Message<?>>)
             .handleNextAndReply(Function<Message<?>, Object>)
             .handleNext(Consumer<Message<?>>)
             .handleNextAndReply(Function<Message<?>, Object>)
             .handleNextAndReply(Function<Message<?>, Object>);
```

Doc Polishing

spring-integration-test-support/src/main/java/org/springframework/integration/test/matcher/HeaderMatcher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -16,11 +16,10 @@
 
 package org.springframework.integration.test.matcher;
 
-import static org.hamcrest.CoreMatchers.is;
-
 import java.util.Date;
 import java.util.Map;
 
+import org.hamcrest.CoreMatchers;
 import org.hamcrest.Description;
 import org.hamcrest.Factory;
 import org.hamcrest.Matcher;
@@ -68,6 +67,7 @@ import org.springframework.messaging.MessageHeaders;
  *
  * @author Alex Peters
  * @author Iwein Fuld
+ * @author Artem Bilan
  *
  */
 public class HeaderMatcher extends TypeSafeMatcher<Message<?>> {
@@ -75,27 +75,22 @@ public class HeaderMatcher extends TypeSafeMatcher<Message<?>> {
 	private final Matcher<?> matcher;
 
 	/**
-	 * @param matcher
+	 * @param matcher the target matcher to delegate
 	 */
-	HeaderMatcher(Matcher<?> matcher) {
+	private HeaderMatcher(Matcher<?> matcher) {
 		super();
 		this.matcher = matcher;
 	}
 
-	/**
-	 * {@inheritDoc}
-	 */
 	@Override
 	public boolean matchesSafely(Message<?> item) {
-		return matcher.matches(item.getHeaders());
+		return this.matcher.matches(item.getHeaders());
 	}
 
-	/**
-	 * {@inheritDoc}
-	 */
 	@Override
 	public void describeTo(Description description) {
-		description.appendText("a Message with Headers containing ").appendDescriptionOf(matcher);
+		description.appendText("a Message with Headers containing ")
+				.appendDescriptionOf(this.matcher);
 	}
 
 	@Factory
@@ -104,12 +99,12 @@ public class HeaderMatcher extends TypeSafeMatcher<Message<?>> {
 	}
 
 	@Factory
-	public static <T> Matcher<Message<?>> hasHeader(String key, Matcher<?> valueMatcher) {
+	public static <T> Matcher<Message<?>> hasHeader(String key, Matcher<T> valueMatcher) {
 		return new HeaderMatcher(MapContentMatchers.hasEntry(key, valueMatcher));
 	}
 
 	@Factory
-	public static <T> Matcher<Message<?>> hasHeaderKey(String key) {
+	public static Matcher<Message<?>> hasHeaderKey(String key) {
 		return new HeaderMatcher(MapContentMatchers.hasKey(key));
 	}
 
@@ -130,7 +125,7 @@ public class HeaderMatcher extends TypeSafeMatcher<Message<?>> {
 
 	@Factory
 	public static Matcher<Message<?>> hasSequenceNumber(Integer value) {
-		return hasSequenceNumber(is(value));
+		return hasSequenceNumber(CoreMatchers.is(value));
 	}
 
 	@Factory
@@ -140,7 +135,7 @@ public class HeaderMatcher extends TypeSafeMatcher<Message<?>> {
 
 	@Factory
 	public static Matcher<Message<?>> hasSequenceSize(Integer value) {
-		return hasSequenceSize(is(value));
+		return hasSequenceSize(CoreMatchers.is(value));
 	}
 
 	@Factory
@@ -150,7 +145,7 @@ public class HeaderMatcher extends TypeSafeMatcher<Message<?>> {
 
 	@Factory
 	public static Matcher<Message<?>> hasExpirationDate(Date value) {
-		return hasExpirationDate(is(value.getTime()));
+		return hasExpirationDate(CoreMatchers.is(value.getTime()));
 	}
 
 	@Factory
@@ -160,7 +155,7 @@ public class HeaderMatcher extends TypeSafeMatcher<Message<?>> {
 
 	@Factory
 	public static Matcher<Message<?>> hasTimestamp(Date value) {
-		return hasTimestamp(is(value.getTime()));
+		return hasTimestamp(CoreMatchers.is(value.getTime()));
 	}
 
 	@Factory

spring-integration-test-support/src/main/java/org/springframework/integration/test/matcher/MapContentMatchers.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -16,8 +16,6 @@
 
 package org.springframework.integration.test.matcher;
 
-import static org.hamcrest.CoreMatchers.anything;
-
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -25,8 +23,8 @@ import java.util.Map;
 import org.hamcrest.Description;
 import org.hamcrest.Factory;
 import org.hamcrest.Matcher;
+import org.hamcrest.Matchers;
 import org.hamcrest.core.AllOf;
-import org.hamcrest.core.IsEqual;
 
 /**
  * Matchers that examine the contents of a {@link Map}.
@@ -62,6 +60,7 @@ import org.hamcrest.core.IsEqual;
  * @author Alex Peters
  * @author Iwein Fuld
  * @author Gunnar Hillert
+ * @author Artem Bilan
  *
  */
 public class MapContentMatchers<T, V> extends
@@ -71,67 +70,48 @@ public class MapContentMatchers<T, V> extends
 
 	private final Matcher<V> valueMatcher;
 
-	/**
-	 * @param key
-	 * @param value
-	 */
-	MapContentMatchers(T key, V value) {
-		this(key, IsEqual.equalTo(value));
+	private MapContentMatchers(T key, V value) {
+		this(key, Matchers.equalTo(value));
 	}
 
-	/**
-	 * @param key
-	 * @param valueMatcher
-	 */
-	MapContentMatchers(T key, Matcher<V> valueMatcher) {
-		super();
+	private MapContentMatchers(T key, Matcher<V> valueMatcher) {
 		this.key = key;
 		this.valueMatcher = valueMatcher;
 	}
 
-	/**
-	 * {@inheritDoc}
-	 */
 	@Override
 	public boolean matchesSafely(Map<? super T, ? super V> item) {
 		return item.containsKey(key) && valueMatcher.matches(item.get(key));
 	}
 
-	/**
-	 * {@inheritDoc}
-	 */
 	@Override
 	public void describeTo(Description description) {
 		description.appendText("an entry with key ").appendValue(key)
 				.appendText(" and value matching ").appendDescriptionOf(
-						valueMatcher);
+				valueMatcher);
 
 	}
 
 	@Factory
-	public static <T, V> Matcher<Map<? super T, ? super V>> hasEntry(T key,
-			V value) {
-		return new MapContentMatchers<T, V>(key, value);
+	public static <T, V> Matcher<Map<? super T, ? super V>> hasEntry(T key, V value) {
+		return new MapContentMatchers<>(key, value);
 	}
 
 	@Factory
-	public static <T, V> Matcher<Map<? super T, ? super V>> hasEntry(T key,
-			Matcher<V> valueMatcher) {
-		return new MapContentMatchers<T, V>(key, valueMatcher);
+	public static <T, V> Matcher<Map<? super T, ? super V>> hasEntry(T key, Matcher<V> valueMatcher) {
+		return new MapContentMatchers<>(key, valueMatcher);
 	}
 
 	@Factory
 	@SuppressWarnings("unchecked")
 	public static <T, V> Matcher<Map<? super T, ? super V>> hasKey(T key) {
-		return new MapContentMatchers<T, V>(key, (Matcher<V>) anything("any Value"));
+		return new MapContentMatchers<>(key, (Matcher<V>) Matchers.anything());
 	}
 
 	@Factory
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public static <T, V> Matcher<Map<? super T, ? super V>> hasAllEntries(
-			Map<T, V> entries) {
-		List<Matcher<? extends Map<? super T, ? super V>>> matchers = new ArrayList<Matcher<? extends Map<? super T, ? super V>>>(
-				entries.size());
+	public static <T, V> Matcher<Map<? super T, ? super V>> hasAllEntries(Map<T, V> entries) {
+		List<Matcher<Map<? super T, ? super V>>> matchers = new ArrayList<>(entries.size());
 		for (Map.Entry<T, V> entry : entries.entrySet()) {
 			final V value = entry.getValue();
 			if (value instanceof Matcher<?>) {
@@ -144,4 +124,5 @@ public class MapContentMatchers<T, V> extends
 		//return AllOf.allOf(matchers); //Does not work with Hamcrest 1.3
 		return new AllOf(matchers);
 	}
+
 }

spring-integration-test-support/src/main/java/org/springframework/integration/test/matcher/MockitoMessageMatchers.java

@@ -16,14 +16,11 @@
 
 package org.springframework.integration.test.matcher;
 
-import static org.mockito.ArgumentMatchers.argThat;
-import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeader;
-import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
-
 import java.util.Map;
 
 import org.hamcrest.Matcher;
 import org.mockito.ArgumentMatcher;
+import org.mockito.ArgumentMatchers;
 import org.mockito.Mockito;
 import org.mockito.internal.hamcrest.HamcrestArgumentMatcher;
 
@@ -75,30 +72,28 @@ public class MockitoMessageMatchers {
 		super();
 	}
 
-	@SuppressWarnings("unchecked")
-	public static <T> Message<T> messageWithPayload(Matcher<T> payloadMatcher) {
-		return argThat(new HamcrestArgumentMatcher<>(hasPayload(payloadMatcher)));
+	public static <T> Message<?> messageWithPayload(Matcher<? super T> payloadMatcher) {
+		return ArgumentMatchers.argThat(new HamcrestArgumentMatcher<>(PayloadMatcher.hasPayload(payloadMatcher)));
 	}
 
-	@SuppressWarnings("unchecked")
-	public static <T> Message<T> messageWithPayload(T payload) {
-		return argThat(new HamcrestArgumentMatcher<>(hasPayload(payload)));
+	public static <T> Message<?> messageWithPayload(T payload) {
+		return ArgumentMatchers.argThat(new HamcrestArgumentMatcher<>(PayloadMatcher.hasPayload(payload)));
 	}
 
 	public static Message<?> messageWithHeaderEntry(String key, Object value) {
-		return argThat(new HamcrestArgumentMatcher<>(hasHeader(key, value)));
+		return ArgumentMatchers.argThat(new HamcrestArgumentMatcher<>(HeaderMatcher.hasHeader(key, value)));
 	}
 
 	public static Message<?> messageWithHeaderKey(String key) {
-		return argThat(new HamcrestArgumentMatcher<>(HeaderMatcher.hasHeaderKey(key)));
+		return ArgumentMatchers.argThat(new HamcrestArgumentMatcher<>(HeaderMatcher.hasHeaderKey(key)));
 	}
 
 	public static <T> Message<?> messageWithHeaderEntry(String key, Matcher<T> valueMatcher) {
-		return argThat(new HamcrestArgumentMatcher<>(HeaderMatcher.<T>hasHeader(key, valueMatcher)));
+		return ArgumentMatchers.argThat(new HamcrestArgumentMatcher<>(HeaderMatcher.<T>hasHeader(key, valueMatcher)));
 	}
 
 	public static Message<?> messageWithHeaderEntries(Map<String, ?> entries) {
-		return argThat(new HamcrestArgumentMatcher<>(HeaderMatcher.hasAllHeaders(entries)));
+		return ArgumentMatchers.argThat(new HamcrestArgumentMatcher<>(HeaderMatcher.hasAllHeaders(entries)));
 	}
 
 }

spring-integration-test-support/src/main/java/org/springframework/integration/test/matcher/PayloadAndHeaderMatcher.java

@@ -47,6 +47,7 @@ import org.springframework.messaging.MessageHeaders;
  * </pre>
  *
  * @author Dave Syer
+ * @author Artem Bilan
  *
  */
 public class PayloadAndHeaderMatcher extends BaseMatcher<Message<?>> {
@@ -65,28 +66,32 @@ public class PayloadAndHeaderMatcher extends BaseMatcher<Message<?>> {
 	private PayloadAndHeaderMatcher(Message<?> expected, String... ignoreKeys) {
 		this.ignoreKeys = ignoreKeys;
 		this.payload = expected.getPayload();
-		this.headers = getHeaders(expected);
+		this.headers = extractHeadersToAssert(expected);
 	}
 
-	private Map<String, Object> getHeaders(Message<?> operand) {
-		HashMap<String, Object> headers = new HashMap<String, Object>(operand.getHeaders());
+	private Map<String, Object> extractHeadersToAssert(Message<?> operand) {
+		HashMap<String, Object> headers = new HashMap<>(operand.getHeaders());
 		headers.remove(MessageHeaders.ID);
 		headers.remove(MessageHeaders.TIMESTAMP);
-		for (String key : ignoreKeys) {
-			headers.remove(key);
+		if (this.ignoreKeys != null) {
+			for (String key : this.ignoreKeys) {
+				headers.remove(key);
+			}
 		}
 		return headers;
 	}
 
 	public boolean matches(Object arg) {
 		Message<?> input = (Message<?>) arg;
-		Map<String, Object> inputHeaders = getHeaders(input);
-		return input.getPayload().equals(payload) && inputHeaders.equals(headers);
+		Map<String, Object> inputHeaders = extractHeadersToAssert(input);
+		return input.getPayload().equals(this.payload) && inputHeaders.equals(this.headers);
 	}
 
 	public void describeTo(Description description) {
-		description.appendText("a Message with Headers that match except ID and timestamp for payload: ").appendValue(
-				payload).appendText(" and headers: ").appendValue(headers);
+		description.appendText("a Message with Headers that match except ID and timestamp for payload: ")
+				.appendValue(this.payload)
+				.appendText(" and headers: ")
+				.appendValue(this.headers);
 	}
 
 }

spring-integration-test-support/src/main/java/org/springframework/integration/test/matcher/PayloadMatcher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -56,44 +56,38 @@ import org.springframework.messaging.Message;
  * @author Iwein Fuld
  *
  */
-@SuppressWarnings("rawtypes")
-public class PayloadMatcher extends TypeSafeMatcher<Message> {
+public class PayloadMatcher extends TypeSafeMatcher<Message<?>> {
 
-	private final Matcher matcher;
+	private final Matcher<?> matcher;
 
 	/**
 	 * Create a PayloadMatcher that matches the payload of messages against the given matcher
 	 */
-	PayloadMatcher(Matcher matcher) {
+	private PayloadMatcher(Matcher<?> matcher) {
 		super();
 		this.matcher = matcher;
 	}
 
-	/**
-	 * {@inheritDoc}
-	 */
 	@Override
-	public boolean matchesSafely(Message message) {
-		return matcher.matches(message.getPayload());
+	public boolean matchesSafely(Message<?> message) {
+		return this.matcher.matches(message.getPayload());
 	}
 
-	/**
-	 * {@inheritDoc}
-	 */
-	//@Override
 	@Override
 	public void describeTo(Description description) {
-		description.appendText("a Message with payload: ").appendDescriptionOf(matcher);
+		description.appendText("a Message with payload: ")
+				.appendDescriptionOf(this.matcher);
 
 	}
 
 	@Factory
-	public static <T> Matcher<Message> hasPayload(T payload) {
+	public static <T> Matcher<Message<?>> hasPayload(T payload) {
 		return new PayloadMatcher(IsEqual.equalTo(payload));
 	}
 
 	@Factory
-	public static <T> Matcher<Message> hasPayload(Matcher<? super T> payloadMatcher) {
+	public static <T> Matcher<Message<?>> hasPayload(Matcher<? super T> payloadMatcher) {
 		return new PayloadMatcher(payloadMatcher);
 	}
+
 }

spring-integration-test-support/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -16,11 +16,9 @@
 
 package org.springframework.integration.test.support;
 
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
 import java.util.List;
 
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -43,8 +41,10 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  * <li>A payload or message to send as a request message on the inputChannel</li>
  * <li>A handler to validate the response received on the outputChannel</li>
  * </ul>
+ *
  * @author David Turanski
  * @author Gary Russell
+ * @author Artem Bilan
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 public abstract class AbstractRequestResponseScenarioTests {
@@ -78,16 +78,16 @@ public abstract class AbstractRequestResponseScenarioTests {
 				((SubscribableChannel) outputChannel).subscribe(scenario.getResponseValidator());
 			}
 
-			assertTrue(name + ": message not sent on " + scenario.getInputChannelName(),
+			Assert.assertTrue(name + ": message not sent on " + scenario.getInputChannelName(),
 					inputChannel.send(scenario.getMessage()));
 
 			if (outputChannel instanceof PollableChannel) {
 				Message<?> response = ((PollableChannel) outputChannel).receive(10000);
-				assertNotNull(name + ": receive timeout on " + scenario.getOutputChannelName(), response);
+				Assert.assertNotNull(name + ": receive timeout on " + scenario.getOutputChannelName(), response);
 				scenario.getResponseValidator().handleMessage(response);
 			}
 
-			assertNotNull("message was not handled on " + outputChannel + " for scenario '" + name + "'.",
+			Assert.assertNotNull("message was not handled on " + outputChannel + " for scenario '" + name + "'.",
 					scenario.getResponseValidator().getLastMessage());
 
 			if (outputChannel instanceof SubscribableChannel) {

spring-integration-test-support/src/test/java/org/springframework/integration/test/matcher/PayloadMatcherTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -35,51 +35,43 @@ import org.springframework.messaging.support.MessageBuilder;
  * @author Alex Peters
  * @author Iwein Fuld
  * @author Gunnar Hillert
+ * @author Artem Bilan
  */
 public class PayloadMatcherTests {
 
-	static final BigDecimal ANY_PAYLOAD = new BigDecimal("1.123");
+	private static final BigDecimal ANY_PAYLOAD = new BigDecimal("1.123");
 
-	Message<BigDecimal> message = MessageBuilder.withPayload(ANY_PAYLOAD).build();
+	private final Message<BigDecimal> message = MessageBuilder.withPayload(ANY_PAYLOAD).build();
 
 	@Test
 	public void hasPayload_withEqualValue_matches() throws Exception {
-		assertThat(message, hasPayload(new BigDecimal("1.123")));
+		assertThat(this.message, hasPayload(new BigDecimal("1.123")));
 	}
 
 	@Test
 	public void hasPayload_withNotEqualValue_notMatching() throws Exception {
-		assertThat(message, not(hasPayload(new BigDecimal("456"))));
+		assertThat(this.message, not(hasPayload(new BigDecimal("456"))));
 	}
 
 	@Test
 	public void hasPayload_withMatcher_matches() throws Exception {
-		assertThat(message,
-				hasPayload(is(instanceOf(BigDecimal.class))));
-		assertThat(message, hasPayload(notNullValue()));
+		assertThat(this.message, hasPayload(is(instanceOf(BigDecimal.class))));
+		assertThat(this.message, hasPayload(notNullValue()));
 	}
 
 	@Test
-	public void hasPayload_withNotMatchingMatcher_notMatching()
-			throws Exception {
-		assertThat(message, not((hasPayload(is(instanceOf(String.class))))));
+	public void hasPayload_withNotMatchingMatcher_notMatching() throws Exception {
+		assertThat(this.message, not((hasPayload(is(instanceOf(String.class))))));
 	}
 
 	@Test
 	public void readableException() throws Exception {
 		try {
-			assertThat(message, hasPayload("woot"));
+			assertThat(this.message, hasPayload("woot"));
 		}
 		catch (AssertionError ae) {
 			assertTrue(ae.getMessage().contains("Expected: a Message with payload: "));
 		}
 	}
 
-	@SuppressWarnings("rawtypes")
-	@Test
-	public void shouldMatchNonParametrizedMessage() throws Exception {
-		Message message = this.message;
-		assertThat(message, hasPayload(new BigDecimal("1.123")));
-	}
-
 }

spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java

@@ -27,8 +27,14 @@ import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.context.Lifecycle;
+import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.endpoint.IntegrationConsumer;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.test.mock.MockMessageHandler;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
@@ -83,6 +89,9 @@ public class MockIntegrationContext implements BeanFactoryAware {
 					if (endpoint instanceof SourcePollingChannelAdapter) {
 						directFieldAccessor.setPropertyValue("source", e.getValue());
 					}
+					else if (endpoint instanceof IntegrationConsumer) {
+						directFieldAccessor.setPropertyValue("handler", e.getValue());
+					}
 				});
 	}
 
@@ -112,7 +121,47 @@ public class MockIntegrationContext implements BeanFactoryAware {
 		instead(pollingAdapterId, mockMessageSource, SourcePollingChannelAdapter.class, "source", autoStartup);
 	}
 
-	private void instead(String endpointId, Object mock, Class<?> endpointClass, String property,
+	public void instead(String consumerEndpointId, MessageHandler mockMessageHandler) {
+		instead(consumerEndpointId, mockMessageHandler, true);
+	}
+
+	public void instead(String consumerEndpointId, MessageHandler mockMessageHandler, boolean autoStartup) {
+		Object endpoint = this.beanFactory.getBean(consumerEndpointId, IntegrationConsumer.class);
+		if (autoStartup && endpoint instanceof Lifecycle) {
+			((Lifecycle) endpoint).stop();
+		}
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(endpoint);
+		Object targetMessageHandler = directFieldAccessor.getPropertyValue("handler");
+		this.beans.put(consumerEndpointId, targetMessageHandler);
+
+		if (mockMessageHandler instanceof MessageProducer) {
+			if (targetMessageHandler instanceof MessageProducer) {
+				MessageChannel outputChannel = TestUtils.getPropertyValue(targetMessageHandler, "outputChannel",
+						MessageChannel.class);
+				((MessageProducer) mockMessageHandler).setOutputChannel(outputChannel);
+			}
+			else {
+				if (mockMessageHandler instanceof MockMessageHandler) {
+					if (TestUtils.getPropertyValue(mockMessageHandler, "hasReplies", Boolean.class)) {
+						throw new IllegalStateException("The [" + mockMessageHandler + "] " +
+								"with replies can't replace simple MessageHandler [" + targetMessageHandler + "]");
+					}
+				}
+				else {
+					throw new IllegalStateException("The MessageProducer handler [" + mockMessageHandler + "] " +
+							"can't replace simple MessageHandler [" + targetMessageHandler + "]");
+				}
+			}
+		}
+
+		directFieldAccessor.setPropertyValue("handler", mockMessageHandler);
+
+		if (autoStartup && endpoint instanceof Lifecycle) {
+			((Lifecycle) endpoint).start();
+		}
+	}
+
+	private void instead(String endpointId, Object messagingComponent, Class<?> endpointClass, String property,
 			boolean autoStartup) {
 		Object endpoint = this.beanFactory.getBean(endpointId, endpointClass);
 		if (autoStartup && endpoint instanceof Lifecycle) {
@@ -120,7 +169,7 @@ public class MockIntegrationContext implements BeanFactoryAware {
 		}
 		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(endpoint);
 		this.beans.put(endpointId, directFieldAccessor.getPropertyValue(property));
-		directFieldAccessor.setPropertyValue("source", mock);
+		directFieldAccessor.setPropertyValue(property, messagingComponent);
 		if (autoStartup && endpoint instanceof Lifecycle) {
 			((Lifecycle) endpoint).start();
 		}

spring-integration-test/src/main/java/org/springframework/integration/test/mock/MockIntegration.java

@@ -16,11 +16,11 @@
 
 package org.springframework.integration.test.mock;
 
-import static org.mockito.BDDMockito.given;
-
 import java.util.ArrayList;
 import java.util.List;
 
+import org.mockito.ArgumentCaptor;
+import org.mockito.BDDMockito;
 import org.mockito.Mockito;
 
 import org.springframework.integration.core.MessageSource;
@@ -89,8 +89,8 @@ public final class MockIntegration {
 	public static MessageSource<?> mockMessageSource(Message<?> message) {
 		MessageSource messageSource = Mockito.mock(MessageSource.class);
 
-		given(messageSource.receive())
-				.<Message<?>>willReturn(message);
+		BDDMockito.given(messageSource.receive())
+				.willReturn(message);
 
 		return messageSource;
 	}
@@ -108,12 +108,29 @@ public final class MockIntegration {
 	public static MessageSource<?> mockMessageSource(Message<?> message, Message<?>... messages) {
 		MessageSource messageSource = Mockito.mock(MessageSource.class);
 
-		given(messageSource.receive())
+		BDDMockito.given(messageSource.receive())
 				.willReturn(message, messages);
 
 		return messageSource;
 	}
 
+	/**
+	 * Build a {@link MockMessageHandler} instance.
+	 * @return the {@link MockMessageHandler} instance ready for interaction
+	 */
+	public static MockMessageHandler mockMessageHandler() {
+		return mockMessageHandler(null);
+	}
+
+	/**
+	 * Build a {@link MockMessageHandler} instance based on the provided {@link ArgumentCaptor}.
+	 * @param messageArgumentCaptor the Mockito ArgumentCaptor to capture incoming messages
+	 * @return the MockMessageHandler instance ready for interaction
+	 */
+	public static MockMessageHandler mockMessageHandler(ArgumentCaptor<Message<?>> messageArgumentCaptor) {
+		return new MockMessageHandler(messageArgumentCaptor);
+	}
+
 	private MockIntegration() {
 	}
 

spring-integration-test/src/main/java/org/springframework/integration/test/mock/MockMessageHandler.java

@@ -0,0 +1,128 @@
+/*
+ * 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.integration.test.mock;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.mockito.ArgumentCaptor;
+import org.mockito.internal.matchers.CapturingMatcher;
+
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
+
+/**
+ * The {@link AbstractMessageProducingHandler} extension for the mocking purpose in tests.
+ * <p>
+ * The provided {@link Consumer}s and {@link Function}s are applied to the incoming
+ * messages one at a time until the last, which is applied for all subsequent messages.
+ * The similar behavior exists in the
+ * {@code Mockito.doReturn(Object toBeReturned, Object... toBeReturnedNext)}.
+ * <p>
+ * Typically is used as a chain of stub actions:
+ * <pre class="code">
+ * {@code
+ *      MockIntegration.mockMessageHandler()
+ *               .handleNext(...)
+ *               .handleNext(...)
+ *               .handleNextAndReply(...)
+ *               .handleNextAndReply(...)
+ *               .handleNext(...)
+ *               .handleNextAndReply(...);
+ * }
+ * </pre>
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public class MockMessageHandler extends AbstractMessageProducingHandler {
+
+	protected final List<Function<Message<?>, ?>> messageFunctions = new LinkedList<>();
+
+	private final CapturingMatcher<Message<?>> capturingMatcher;
+
+	protected Function<Message<?>, ?> lastFunction;
+
+	protected boolean hasReplies;
+
+	@SuppressWarnings("unchecked")
+	protected MockMessageHandler(ArgumentCaptor<Message<?>> messageArgumentCaptor) {
+		if (messageArgumentCaptor != null) {
+			this.capturingMatcher = (CapturingMatcher<Message<?>>) TestUtils.getPropertyValue(messageArgumentCaptor,
+					"capturingMatcher", CapturingMatcher.class);
+		}
+		else {
+			this.capturingMatcher = null;
+		}
+	}
+
+	/**
+	 * Add the {@link Consumer} to the stack to handle the next incoming message.
+	 * @param nextMessageConsumer the Consumer to handle the next incoming message.
+	 * @return this
+	 */
+	public MockMessageHandler handleNext(Consumer<Message<?>> nextMessageConsumer) {
+		this.lastFunction = m -> {
+			nextMessageConsumer.accept(m);
+			return null;
+		};
+		this.messageFunctions.add(this.lastFunction);
+		return this;
+	}
+
+	/**
+	 * Add the {@link Function} to the stack to handle the next incoming message
+	 * and produce reply for it.
+	 * @param nextMessageFunction the Function to handle the next incoming message.
+	 * @return this
+	 */
+	public MockMessageHandler handleNextAndReply(Function<Message<?>, ?> nextMessageFunction) {
+		this.lastFunction = nextMessageFunction;
+		this.messageFunctions.add(this.lastFunction);
+		this.hasReplies = true;
+		return this;
+	}
+
+	@Override
+	protected void handleMessageInternal(Message<?> message) throws Exception {
+		if (this.capturingMatcher != null) {
+			this.capturingMatcher.captureFrom(message);
+		}
+
+		Function<Message<?>, ?> function = this.lastFunction;
+
+		synchronized (this) {
+			Iterator<Function<Message<?>, ?>> iterator = this.messageFunctions.iterator();
+			if (iterator.hasNext()) {
+				function = iterator.next();
+				iterator.remove();
+			}
+		}
+
+		Object result = function.apply(message);
+
+		if (result != null) {
+			sendOutputs(result, message);
+		}
+	}
+
+}

spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageHandlerTests.java

@@ -0,0 +1,260 @@
+/*
+ * 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.integration.test.mock;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeader;
+import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
+import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
+import static org.springframework.integration.test.mock.MockIntegration.mockMessageHandler;
+
+import java.util.List;
+
+import org.junit.After;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.Poller;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.handler.ExpressionEvaluatingMessageHandler;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.context.MockIntegrationContext;
+import org.springframework.integration.test.context.SpringIntegrationTest;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+@RunWith(SpringRunner.class)
+@ContextConfiguration(classes = MockMessageHandlerTests.Config.class)
+@SpringIntegrationTest
+public class MockMessageHandlerTests {
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Autowired
+	private MockIntegrationContext mockIntegrationContext;
+
+	@Autowired
+	private MessageChannel mockHandlerChannel;
+
+	@Autowired
+	private MessageChannel pojoServiceChannel;
+
+	@Autowired
+	private MessageChannel rawChannel;
+
+	@Autowired
+	private QueueChannel results;
+
+	@Autowired
+	private ArgumentCaptor<Message<?>> messageArgumentCaptor;
+
+	@After
+	public void tearDown() {
+		this.mockIntegrationContext.resetBeans();
+		results.purge(null);
+	}
+
+	@Test
+	public void testMockMessageHandler() {
+		QueueChannel replies = new QueueChannel();
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader("bar", "BAR")
+				.setHeader("baz", "BAZ")
+				.setReplyChannel(replies)
+				.build();
+
+		this.mockHandlerChannel.send(message);
+		this.mockHandlerChannel.send(message);
+		this.mockHandlerChannel.send(message);
+
+		Message<String> message1 = MessageBuilder.fromMessage(message)
+				.removeHeaders("bar", "baz")
+				.build();
+		this.mockHandlerChannel.send(message1);
+
+		for (int i = 0; i < 4; i++) {
+			Message<?> receive = replies.receive(10000);
+			assertNotNull(receive);
+			assertEquals("foo", receive.getPayload());
+		}
+
+		List<Message<?>> messages = this.messageArgumentCaptor.getAllValues();
+
+		assertEquals(4, messages.size());
+
+		assertThat(messages.get(0), hasHeader("bar", "BAR"));
+		assertThat(messages.get(1),
+				sameExceptIgnorableHeaders(MessageBuilder.withPayload("foo")
+								.setHeader("baz", "BAZ")
+								.build(),
+						"bar", MessageHeaders.REPLY_CHANNEL));
+
+		assertThat(messages.get(2), hasPayload("foo"));
+		assertThat(messages.get(3), hasPayload("foo"));
+	}
+
+	@Test
+	public void testMockMessageHandlerPojoService() {
+		this.pojoServiceChannel.send(new GenericMessage<>("bar"));
+
+		Message<?> receive = this.results.receive(10000);
+
+		assertNotNull(receive);
+		assertEquals("barbar", receive.getPayload());
+
+		MessageHandler mockMessageHandler =
+				mockMessageHandler()
+						.handleNextAndReply(m -> m.getPayload().toString().toUpperCase());
+
+		this.mockIntegrationContext.instead("mockMessageHandlerTests.Config.myService.serviceActivator",
+				mockMessageHandler);
+
+		this.pojoServiceChannel.send(new GenericMessage<>("foo"));
+		receive = this.results.receive(10000);
+
+		assertNotNull(receive);
+		assertEquals("FOO", receive.getPayload());
+
+		try {
+			this.pojoServiceChannel.send(new GenericMessage<>("bar"));
+			fail("AssertionError expected");
+		}
+		catch (Error e) {
+			assertThat(e, instanceOf(AssertionError.class));
+		}
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testMockRawHandler() {
+		ArgumentCaptor<Message<?>> messageArgumentCaptor = ArgumentCaptor.forClass(Message.class);
+		MessageHandler mockMessageHandler =
+				spy(mockMessageHandler(messageArgumentCaptor))
+						.handleNext(m -> { });
+
+		String endpointId = "rawHandlerConsumer";
+		this.mockIntegrationContext.instead(endpointId, mockMessageHandler);
+
+		Object endpoint = this.context.getBean(endpointId);
+		assertSame(mockMessageHandler, TestUtils.getPropertyValue(endpoint, "handler", MessageHandler.class));
+
+		GenericMessage<String> message = new GenericMessage<>("foo");
+
+		this.rawChannel.send(message);
+
+		verify(mockMessageHandler)
+				.handleMessage(message);
+
+		assertSame(message, messageArgumentCaptor.getValue());
+
+		this.mockIntegrationContext.resetBeans(endpointId);
+
+		mockMessageHandler =
+				mockMessageHandler()
+						.handleNextAndReply(m -> m);
+
+		try {
+			this.mockIntegrationContext.instead(endpointId, mockMessageHandler);
+			fail("IllegalStateException expected");
+		}
+		catch (Exception e) {
+			assertThat(e, instanceOf(IllegalStateException.class));
+			assertThat(e.getMessage(), containsString("with replies can't replace simple MessageHandler"));
+		}
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class Config {
+
+		@Bean
+		public PollableChannel results() {
+			return new QueueChannel();
+		}
+
+		@Bean
+		@SuppressWarnings("unchecked")
+		public ArgumentCaptor<Message<?>> messageArgumentCaptor() {
+			return ArgumentCaptor.forClass(Message.class);
+		}
+
+		@Bean
+		public PollableChannel mockHandlerChannel() {
+			return new QueueChannel();
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "mockHandlerChannel",
+				poller = @Poller(fixedDelay = "100"))
+		public MessageHandler mockHandler() {
+			return mockMessageHandler(messageArgumentCaptor())
+					.handleNextAndReply(m -> m)
+					.handleNextAndReply(m -> m)
+					.handleNextAndReply(m -> "foo");
+		}
+
+		@ServiceActivator(inputChannel = "pojoServiceChannel", outputChannel = "results")
+		public String myService(String payload) {
+			return payload + payload;
+		}
+
+		@Bean
+		public SubscribableChannel rawChannel() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		public EventDrivenConsumer rawHandlerConsumer() {
+			return new EventDrivenConsumer(rawChannel(),
+					new ExpressionEvaluatingMessageHandler(new ValueExpression<>("test")));
+		}
+
+	}
+
+}

src/checkstyle/checkstyle.xml

@@ -96,6 +96,7 @@
 				org.springframework.integration.test.matcher.MockitoMessageMatchers.*,
 				org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.*,
 				org.springframework.integration.test.matcher.PayloadMatcher.*,
+				org.springframework.integration.test.mock.MockIntegration.*,
 				org.springframework.integration.test.util.TestUtils.*,
 				org.springframework.test.web.client.match.MockRestRequestMatchers.*,
 				org.springframework.test.web.client.response.MockRestResponseCreators.*,

src/reference/asciidoc/testing.adoc

@@ -287,6 +287,40 @@ assertNotNull(receive);
 assertEquals("FOO", receive.getPayload());
 ----
 
+Unlike the Mockito `MessageSource` mock object, the `MockMessageHandler` is just a regular `AbstractMessageProducingHandler` extension with a chain API to stub handling for incoming messages.
+The `MockMessageHandler` provides `handleNext(Consumer<Message<?>>)` to specify a one-way stub for the next request message; used to mock message handlers that don't produce replies.
+The `handleNextAndReply(Function<Message<?>, ?>)` is provided for performing the same stub logic for the next request message and producing a reply for it.
+They can be chained to simulate any arbitrary request-reply scenarios for all expected request messages variants.
+These consumers and functions are applied to the incoming messages, one at a time from the stack, until the last, which is then used for all remaining messages.
+The behavior is similar to the Mockito `Answer` or `doReturn()` API.
+
+In addition, a Mockito `ArgumentCaptor<Message<?>>` can be supplied to the `MockMessageHandler` in a constructor argument.
+Each request message for the `MockMessageHandler` is captured by that `ArgumentCaptor`.
+During the test, its `getValue()/getAllValues()` can be used to verify and assert those request messages.
+
+The `MockIntegrationContext` provides an `instead()` API for replacing the actual configured `MessageHandler` with a `MockMessageHandler`, in the particular endpoint in the application context under test.
+
+A typical usage might be:
+
+[source,java]
+----
+ArgumentCaptor<Message<?>> messageArgumentCaptor = ArgumentCaptor.forClass(Message.class);
+
+MessageHandler mockMessageHandler =
+        mockMessageHandler(messageArgumentCaptor)
+                .handleNextAndReply(m -> m.getPayload().toString().toUpperCase());
+
+this.mockIntegrationContext.instead("myService.serviceActivator", mockMessageHandler);
+GenericMessage<String> message = new GenericMessage<>("foo");
+this.myChannel.send(message);
+Message<?> received = this.results.receive(10000);
+assertNotNull(received);
+assertEquals("FOO", received.getPayload());
+assertSame(message, messageArgumentCaptor.getValue());
+----
+
+See `MockIntegration` and `MockMessageHandler` JavaDocs for more information.
+
 [[testing-other-resources]]
 === Other Resources