Merge branch '2.0.x'

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/integration/StubRunnerIntegrationConfiguration.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.contract.stubrunner.messaging.integration;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.function.Consumer;
@@ -37,6 +38,9 @@ import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.transformer.MessageTransformingHandler;
 import org.springframework.messaging.Message;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.util.StringUtils;
 
 /**
  * Spring Integration configuration that iterates over the downloaded Groovy DSLs and
@@ -55,52 +59,60 @@ public class StubRunnerIntegrationConfiguration {
 			BatchStubRunner batchStubRunner) {
 		Map<StubConfiguration, Collection<Contract>> contracts = batchStubRunner
 				.getContracts();
+		IntegrationFlowBuilder dummyBuilder = IntegrationFlows.from(DummyMessageHandler.CHANNEL_NAME)
+				.handle(new DummyMessageHandler(), "handle");
+		beanFactory.initializeBean(dummyBuilder.get(), DummyMessageHandler.CHANNEL_NAME + ".flow");
 		for (Entry<StubConfiguration, Collection<Contract>> entry : contracts
 				.entrySet()) {
-			String name = entry.getKey().getGroupId() + "_"
-					+ entry.getKey().getArtifactId();
-			for (Contract dsl : entry.getValue()) {
+			StubConfiguration key = entry.getKey();
+			Collection<Contract> value = entry.getValue();
+			String name = key.getGroupId() + "_" + key.getArtifactId();
+			MultiValueMap<String, Contract> map = new LinkedMultiValueMap<>();
+			for (Contract dsl : value) {
+				if (dsl == null) {
+					continue;
+				}
 				if (dsl.getInput() != null && dsl.getInput().getMessageFrom() != null
-						&& dsl.getInput().getMessageFrom().getClientValue() != null) {
-					final String flowName = name + "_" + dsl.getLabel() + "_"
-							+ dsl.hashCode();
-					IntegrationFlowBuilder builder = IntegrationFlows
-							.from(dsl.getInput().getMessageFrom().getClientValue())
-							.filter(new StubRunnerIntegrationMessageSelector(dsl),
-									new Consumer<FilterEndpointSpec>() {
-										@Override
-										public void accept(FilterEndpointSpec e) {
-											e.id(flowName + ".filter");
-										}
-									})
-							.transform(new StubRunnerIntegrationTransformer(dsl),
-									new Consumer<GenericEndpointSpec<MessageTransformingHandler>>() {
-										@Override
-										public void accept(
-												GenericEndpointSpec<MessageTransformingHandler> e) {
-											e.id(flowName + ".transformer");
-										}
-									});
-					if (dsl.getOutputMessage() != null) {
-						builder = builder.channel(
-								dsl.getOutputMessage().getSentTo().getClientValue());
-					}
-					else {
-						builder = builder.handle(new DummyMessageHandler(), "handle");
-					}
-					beanFactory.initializeBean(builder.get(), flowName);
-					beanFactory.getBean(flowName + ".filter", Lifecycle.class).start();
-					beanFactory.getBean(flowName + ".transformer", Lifecycle.class)
-							.start();
+						&& StringUtils.hasText(
+						dsl.getInput().getMessageFrom().getClientValue())) {
+					String from = dsl.getInput().getMessageFrom().getClientValue();
+					map.add(from, dsl);
 				}
 			}
+			for (Entry<String, List<Contract>> entries : map.entrySet()) {
+				final String flowName = name + "_" + entries.getKey() + "_"
+						+ entries.getValue().hashCode();
+				IntegrationFlowBuilder builder = IntegrationFlows.from(entries.getKey())
+						.filter(new StubRunnerIntegrationMessageSelector(entries.getValue()),
+								new Consumer<FilterEndpointSpec>() {
+									@Override
+									public void accept(FilterEndpointSpec e) {
+										e.id(flowName + ".filter");
+									}
+								})
+						.transform(new StubRunnerIntegrationTransformer(entries.getValue()),
+								new Consumer<GenericEndpointSpec<MessageTransformingHandler>>() {
+									@Override
+									public void accept(
+											GenericEndpointSpec<MessageTransformingHandler> e) {
+										e.id(flowName + ".transformer");
+									}
+								})
+						.route(new StubRunnerIntegrationRouter(entries.getValue(), beanFactory));
+				beanFactory.initializeBean(builder.get(), flowName);
+				beanFactory.getBean(flowName + ".filter", Lifecycle.class).start();
+				beanFactory.getBean(flowName + ".transformer", Lifecycle.class)
+						.start();
+			}
+
 		}
 		return new FlowRegistrar();
 	}
 
-	private static class DummyMessageHandler {
+	static class DummyMessageHandler {
+
+		static String CHANNEL_NAME = "stub_runner_dummy_channel";
 
-		@SuppressWarnings("unused")
 		public void handle(Message<?> message) {
 		}
 

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/integration/StubRunnerIntegrationMessageSelector.java

@@ -16,25 +16,31 @@
 
 package org.springframework.cloud.contract.stubrunner.messaging.integration;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
+import java.util.WeakHashMap;
 import java.util.regex.Pattern;
 
-import org.springframework.cloud.contract.spec.Contract;
-import org.springframework.cloud.contract.spec.internal.BodyMatcher;
-import org.springframework.cloud.contract.spec.internal.BodyMatchers;
-import org.springframework.cloud.contract.spec.internal.Header;
-import org.springframework.cloud.contract.verifier.util.MapConverter;
-import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierObjectMapper;
-import org.springframework.cloud.contract.verifier.util.JsonPaths;
-import org.springframework.cloud.contract.verifier.util.JsonToJsonPathsConverter;
-import org.springframework.cloud.contract.verifier.util.MethodBufferingJsonVerifiable;
-import org.springframework.integration.core.MessageSelector;
-import org.springframework.messaging.Message;
-
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.jayway.jsonpath.DocumentContext;
 import com.jayway.jsonpath.JsonPath;
 import com.toomuchcoding.jsonassert.JsonAssertion;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.contract.spec.Contract;
+import org.springframework.cloud.contract.spec.internal.BodyMatcher;
+import org.springframework.cloud.contract.spec.internal.BodyMatchers;
+import org.springframework.cloud.contract.spec.internal.Header;
+import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierObjectMapper;
+import org.springframework.cloud.contract.verifier.util.JsonPaths;
+import org.springframework.cloud.contract.verifier.util.JsonToJsonPathsConverter;
+import org.springframework.cloud.contract.verifier.util.MapConverter;
+import org.springframework.cloud.contract.verifier.util.MethodBufferingJsonVerifiable;
+import org.springframework.integration.core.MessageSelector;
+import org.springframework.messaging.Message;
 
 /**
  * Passes through a message that matches the one defined in the DSL
@@ -44,23 +50,65 @@ import com.toomuchcoding.jsonassert.JsonAssertion;
  */
 class StubRunnerIntegrationMessageSelector implements MessageSelector {
 
-	private final Contract groovyDsl;
+	private static final Map<Message, Contract> CACHE =
+			Collections.synchronizedMap(new WeakHashMap<Message, Contract>());
+
+	private static final Log log = LogFactory.getLog(StubRunnerIntegrationMessageSelector.class);
+
+	private final List<Contract> groovyDsls;
 
 	private final ContractVerifierObjectMapper objectMapper = new ContractVerifierObjectMapper();
 
 	StubRunnerIntegrationMessageSelector(Contract groovyDsl) {
-		this.groovyDsl = groovyDsl;
+		this(Collections.singletonList(groovyDsl));
+	}
+
+	StubRunnerIntegrationMessageSelector(List<Contract> groovyDsls) {
+		this.groovyDsls = groovyDsls;
 	}
 
 	@Override
 	public boolean accept(Message<?> message) {
-		if (!headersMatch(message)) {
-			return false;
+		return matchingContract(message) != null;
+	}
+
+	Contract matchingContract(Message<?> message) {
+		if (CACHE.containsKey(message)) {
+			return CACHE.get(message);
+		}
+		Contract contract = getContract(message);
+		if (contract != null) {
+			CACHE.put(message, contract);
+		}
+		return contract;
+	}
+
+	void updateCache(Message<?> message, Contract contract) {
+		CACHE.put(message, contract);
+	}
+
+	private Contract getContract(Message<?> message) {
+		for (Contract groovyDsl : this.groovyDsls) {
+			Contract contract = matchContract(message, groovyDsl);
+			if (contract != null) {
+				return contract;
+			}
+		}
+		return null;
+	}
+
+	private Contract matchContract(Message<?> message, Contract groovyDsl) {
+		List<String> unmatchedHeaders = headersMatch(message, groovyDsl);
+		if (!unmatchedHeaders.isEmpty()) {
+			if (log.isDebugEnabled()) {
+				log.debug("Contract [" + groovyDsl
+						+ "] hasn't matched the following headers " + unmatchedHeaders);
+			}
+			return null;
 		}
 		Object inputMessage = message.getPayload();
-		BodyMatchers matchers = this.groovyDsl.getInput().getBodyMatchers();
-		Object dslBody = MapConverter
-				.getStubSideValues(this.groovyDsl.getInput().getMessageBody());
+		BodyMatchers matchers = groovyDsl.getInput().getBodyMatchers();
+		Object dslBody = MapConverter.getStubSideValues(groovyDsl.getInput().getMessageBody());
 		Object matchingInputMessage = JsonToJsonPathsConverter
 				.removeMatchingJsonPaths(dslBody, matchers);
 		JsonPaths jsonPaths = JsonToJsonPathsConverter
@@ -74,42 +122,72 @@ class StubRunnerIntegrationMessageSelector implements MessageSelector {
 		catch (JsonProcessingException e) {
 			throw new IllegalStateException("Cannot serialize to JSON", e);
 		}
+		List<String> unmatchedJsonPath = new ArrayList<>();
 		boolean matches = true;
 		for (MethodBufferingJsonVerifiable path : jsonPaths) {
-			matches &= matchesJsonPath(parsedJson, path.jsonPath());
+			matches &= matchesJsonPath(unmatchedJsonPath, parsedJson, path.jsonPath());
 		}
 		if (matchers != null && matchers.hasMatchers()) {
 			for (BodyMatcher matcher : matchers.jsonPathMatchers()) {
 				String jsonPath = JsonToJsonPathsConverter
 						.convertJsonPathAndRegexToAJsonPath(matcher, dslBody);
-				matches &= matchesJsonPath(parsedJson, jsonPath);
+				matches &= matchesJsonPath(unmatchedJsonPath, parsedJson, jsonPath);
 			}
 		}
-		return matches;
+		if (!unmatchedJsonPath.isEmpty()) {
+			if (log.isDebugEnabled()) {
+				log.debug("Contract [" + groovyDsl + "] didn't much the body due to "
+						+ unmatchedJsonPath);
+			}
+		}
+		if (matches) {
+			return groovyDsl;
+		}
+		return null;
 	}
 
-	private boolean matchesJsonPath(DocumentContext parsedJson, String jsonPath) {
+	private boolean matchesJsonPath(List<String> unmatchedJsonPath,
+			DocumentContext parsedJson, String jsonPath) {
 		try {
 			JsonAssertion.assertThat(parsedJson).matchesJsonPath(jsonPath);
 			return true;
 		}
 		catch (Exception e) {
+			unmatchedJsonPath.add(e.getLocalizedMessage());
 			return false;
 		}
 	}
 
-	private boolean headersMatch(Message<?> message) {
+	private List<String> headersMatch(Message<?> message, Contract groovyDsl) {
+		List<String> unmatchedHeaders = new ArrayList<>();
 		Map<String, Object> headers = message.getHeaders();
-		boolean matches = true;
-		for (Header it : this.groovyDsl.getInput().getMessageHeaders().getEntries()) {
+		for (Header it : groovyDsl.getInput().getMessageHeaders().getEntries()) {
 			String name = it.getName();
 			Object value = it.getClientValue();
 			Object valueInHeader = headers.get(name);
-			matches &= value instanceof Pattern
-					? ((Pattern) value).matcher(valueInHeader.toString()).matches()
-					: valueInHeader != null && valueInHeader.equals(value);
+			boolean matches;
+			if (value instanceof Pattern) {
+				Pattern pattern = (Pattern) value;
+				matches = pattern.matcher(valueInHeader.toString()).matches();
+			}
+			else {
+				matches = valueInHeader != null
+						&& valueInHeader.toString().equals(value.toString());
+			}
+			if (!matches) {
+				unmatchedHeaders.add("Header with name [" + name + "] was supposed to "
+						+ unmatchedText(value) + " but the value is ["
+						+ (valueInHeader != null ? valueInHeader.toString() : "null")
+						+ "]");
+			}
 		}
-		return matches;
+		return unmatchedHeaders;
+	}
+
+	private String unmatchedText(Object expectedValue) {
+		return expectedValue instanceof Pattern
+				? "match pattern [" + ((Pattern) expectedValue).pattern() + "]"
+				: "be equal to [" + expectedValue + "]";
 	}
 
 }

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/integration/StubRunnerIntegrationRouter.java

@@ -0,0 +1,39 @@
+package org.springframework.cloud.contract.stubrunner.messaging.integration;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.cloud.contract.spec.Contract;
+import org.springframework.integration.router.AbstractMessageRouter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Marcin Grzejszczak
+ */
+class StubRunnerIntegrationRouter extends AbstractMessageRouter {
+
+	private final StubRunnerIntegrationMessageSelector selector;
+	private final BeanFactory beanFactory;
+
+	StubRunnerIntegrationRouter(List<Contract> groovyDsls, BeanFactory beanFactory) {
+		this.selector = new StubRunnerIntegrationMessageSelector(groovyDsls);
+		this.beanFactory = beanFactory;
+	}
+
+	@Override
+	protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
+		Contract dsl = this.selector.matchingContract(message);
+		if (dsl != null && dsl.getOutputMessage() != null
+				&& dsl.getOutputMessage().getSentTo() != null) {
+			String channelName = dsl.getOutputMessage().getSentTo().getClientValue();
+			return Collections
+					.singleton((MessageChannel) this.beanFactory.getBean(channelName));
+		}
+		return Collections
+				.singleton((MessageChannel)
+						this.beanFactory.getBean(StubRunnerIntegrationConfiguration.DummyMessageHandler.CHANNEL_NAME));
+	}
+}

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/integration/StubRunnerIntegrationTransformer.java

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.contract.stubrunner.messaging.integration;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 import org.springframework.cloud.contract.spec.Contract;
@@ -33,22 +35,35 @@ import org.springframework.messaging.support.MessageBuilder;
 class StubRunnerIntegrationTransformer
 		implements GenericTransformer<Message<?>, Message<?>> {
 
-	private final Contract groovyDsl;
+	private final StubRunnerIntegrationMessageSelector selector;
 
 	StubRunnerIntegrationTransformer(Contract groovyDsl) {
-		this.groovyDsl = groovyDsl;
+		this(Collections.singletonList(groovyDsl));
+	}
+
+	StubRunnerIntegrationTransformer(List<Contract> groovyDsls) {
+		this.selector = new StubRunnerIntegrationMessageSelector(groovyDsls);
 	}
 
 	@Override
 	public Message<?> transform(Message<?> source) {
-		if (this.groovyDsl.getOutputMessage() == null) {
+		Contract groovyDsl = matchingContract(source);
+		if (groovyDsl == null || groovyDsl.getOutputMessage() == null) {
 			return source;
 		}
 		String payload = BodyExtractor
-				.extractStubValueFrom(this.groovyDsl.getOutputMessage().getBody());
-		Map<String, Object> headers = this.groovyDsl.getOutputMessage().getHeaders()
+				.extractStubValueFrom(groovyDsl.getOutputMessage().getBody());
+		Map<String, Object> headers = groovyDsl.getOutputMessage().getHeaders()
 				.asStubSideMap();
-		return MessageBuilder.createMessage(payload, new MessageHeaders(headers));
+		MessageHeaders messageHeaders = new MessageHeaders(headers);
+		Message<String> message = MessageBuilder.createMessage(payload,
+				messageHeaders);
+		this.selector.updateCache(message, groovyDsl);
+		return message;
+	}
+
+	Contract matchingContract(Message<?> source) {
+		return this.selector.matchingContract(source);
 	}
 
 }

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/stream/StubRunnerMessageRouter.java

@@ -0,0 +1,40 @@
+package org.springframework.cloud.contract.stubrunner.messaging.stream;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.cloud.contract.spec.Contract;
+import org.springframework.integration.router.AbstractMessageRouter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Marcin Grzejszczak
+ */
+class StubRunnerMessageRouter extends AbstractMessageRouter {
+
+	private final StubRunnerStreamMessageSelector selector;
+	private final BeanFactory beanFactory;
+
+	StubRunnerMessageRouter(List<Contract> groovyDsls, BeanFactory beanFactory) {
+		this.selector = new StubRunnerStreamMessageSelector(groovyDsls);
+		this.beanFactory = beanFactory;
+	}
+
+	@Override
+	protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
+		Contract dsl = this.selector.matchingContract(message);
+		if (dsl != null && dsl.getOutputMessage() != null
+				&& dsl.getOutputMessage().getSentTo() != null) {
+			String channelName = StubRunnerStreamConfiguration.resolvedDestination(this.beanFactory,
+					dsl.getOutputMessage().getSentTo().getClientValue());
+			return Collections
+					.singleton((MessageChannel) this.beanFactory.getBean(channelName));
+		}
+		return Collections
+				.singleton((MessageChannel)
+						this.beanFactory.getBean(StubRunnerStreamConfiguration.DummyMessageHandler.CHANNEL_NAME));
+	}
+}

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/stream/StubRunnerStreamConfiguration.java

@@ -17,12 +17,15 @@
 package org.springframework.cloud.contract.stubrunner.messaging.stream;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.function.Consumer;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
@@ -46,6 +49,8 @@ import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.transformer.MessageTransformingHandler;
 import org.springframework.messaging.Message;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
 import org.springframework.util.StringUtils;
 
 /**
@@ -69,11 +74,15 @@ public class StubRunnerStreamConfiguration {
 			BatchStubRunner batchStubRunner) {
 		Map<StubConfiguration, Collection<Contract>> contracts = batchStubRunner
 				.getContracts();
+		IntegrationFlowBuilder dummyBuilder = IntegrationFlows.from(DummyMessageHandler.CHANNEL_NAME)
+				.handle(new DummyMessageHandler(), "handle");
+		beanFactory.initializeBean(dummyBuilder.get(), DummyMessageHandler.CHANNEL_NAME + ".flow");
 		for (Entry<StubConfiguration, Collection<Contract>> entry : contracts
 				.entrySet()) {
 			StubConfiguration key = entry.getKey();
 			Collection<Contract> value = entry.getValue();
 			String name = key.getGroupId() + "_" + key.getArtifactId();
+			MultiValueMap<String, Contract> map = new LinkedMultiValueMap<>();
 			for (Contract dsl : value) {
 				if (dsl == null) {
 					continue;
@@ -81,38 +90,9 @@ public class StubRunnerStreamConfiguration {
 				if (dsl.getInput() != null && dsl.getInput().getMessageFrom() != null
 						&& StringUtils.hasText(
 								dsl.getInput().getMessageFrom().getClientValue())) {
-					final String flowName = name + "_" + dsl.getLabel() + "_"
-							+ dsl.hashCode();
 					String from = resolvedDestination(beanFactory,
 							dsl.getInput().getMessageFrom().getClientValue());
-					IntegrationFlowBuilder builder = IntegrationFlows.from(from)
-							.filter(new StubRunnerStreamMessageSelector(dsl),
-									new Consumer<FilterEndpointSpec>() {
-										@Override
-										public void accept(FilterEndpointSpec e) {
-											e.id(flowName + ".filter");
-										}
-									})
-							.transform(new StubRunnerStreamTransformer(dsl),
-									new Consumer<GenericEndpointSpec<MessageTransformingHandler>>() {
-										@Override
-										public void accept(
-												GenericEndpointSpec<MessageTransformingHandler> e) {
-											e.id(flowName + ".transformer");
-										}
-									});
-					if (dsl.getOutputMessage() != null
-							&& dsl.getOutputMessage().getSentTo() != null) {
-						builder = builder.channel(resolvedDestination(beanFactory,
-								dsl.getOutputMessage().getSentTo().getClientValue()));
-					}
-					else {
-						builder = builder.handle(new DummyMessageHandler(), "handle");
-					}
-					beanFactory.initializeBean(builder.get(), flowName);
-					beanFactory.getBean(flowName + ".filter", Lifecycle.class).start();
-					beanFactory.getBean(flowName + ".transformer", Lifecycle.class)
-							.start();
+					map.add(from, dsl);
 				}
 				else if (dsl.getOutputMessage() != null
 						&& dsl.getOutputMessage().getSentTo() != null
@@ -124,11 +104,37 @@ public class StubRunnerStreamConfiguration {
 							dsl.getOutputMessage().getSentTo().getClientValue());
 				}
 			}
+			for (Entry<String, List<Contract>> entries : map.entrySet()) {
+				final String flowName = name + "_" + entries.getKey() + "_"
+						+ entries.getValue().hashCode();
+				IntegrationFlowBuilder builder = IntegrationFlows.from(entries.getKey())
+						.filter(new StubRunnerStreamMessageSelector(entries.getValue()),
+								new Consumer<FilterEndpointSpec>() {
+									@Override
+									public void accept(FilterEndpointSpec e) {
+										e.id(flowName + ".filter");
+									}
+								})
+						.transform(new StubRunnerStreamTransformer(entries.getValue()),
+								new Consumer<GenericEndpointSpec<MessageTransformingHandler>>() {
+									@Override
+									public void accept(
+											GenericEndpointSpec<MessageTransformingHandler> e) {
+										e.id(flowName + ".transformer");
+									}
+								})
+						.route(new StubRunnerMessageRouter(entries.getValue(), beanFactory));
+				beanFactory.initializeBean(builder.get(), flowName);
+				beanFactory.getBean(flowName + ".filter", Lifecycle.class).start();
+				beanFactory.getBean(flowName + ".transformer", Lifecycle.class)
+						.start();
+			}
+
 		}
 		return new FlowRegistrar();
 	}
 
-	private String resolvedDestination(AutowireCapableBeanFactory context,
+	static String resolvedDestination(BeanFactory context,
 			String destination) {
 		Map<String, BindingProperties> bindings = bindingProperties(context);
 		for (Map.Entry<String, BindingProperties> entry : bindings.entrySet()) {
@@ -147,12 +153,13 @@ public class StubRunnerStreamConfiguration {
 		return destination;
 	}
 
-	private Map<String, BindingProperties> bindingProperties(
-			AutowireCapableBeanFactory context) {
+	private static Map<String, BindingProperties> bindingProperties(BeanFactory context) {
 		return context.getBean(BindingServiceProperties.class).getBindings();
 	}
 
-	private static class DummyMessageHandler {
+	static class DummyMessageHandler {
+
+		static String CHANNEL_NAME = "stub_runner_dummy_channel";
 
 		public void handle(Message<?> message) {
 		}

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/stream/StubRunnerStreamMessageSelector.java

@@ -17,12 +17,19 @@
 package org.springframework.cloud.contract.stubrunner.messaging.stream;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.WeakHashMap;
 import java.util.regex.Pattern;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.jayway.jsonpath.DocumentContext;
+import com.jayway.jsonpath.JsonPath;
+import com.toomuchcoding.jsonassert.JsonAssertion;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.cloud.contract.spec.Contract;
 import org.springframework.cloud.contract.spec.internal.BodyMatcher;
 import org.springframework.cloud.contract.spec.internal.BodyMatchers;
@@ -35,11 +42,6 @@ import org.springframework.cloud.contract.verifier.util.MethodBufferingJsonVerif
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.messaging.Message;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.jayway.jsonpath.DocumentContext;
-import com.jayway.jsonpath.JsonPath;
-import com.toomuchcoding.jsonassert.JsonAssertion;
-
 /**
  * Passes through a message that matches the one defined in the DSL
  *
@@ -48,31 +50,67 @@ import com.toomuchcoding.jsonassert.JsonAssertion;
  */
 class StubRunnerStreamMessageSelector implements MessageSelector {
 
+	private static final Map<Message, Contract> CACHE =
+			Collections.synchronizedMap(new WeakHashMap<Message, Contract>());
+
 	private static final Log log = LogFactory
 			.getLog(StubRunnerStreamMessageSelector.class);
 
-	private final Contract groovyDsl;
+	private final List<Contract> groovyDsls;
 
 	private final ContractVerifierObjectMapper objectMapper = new ContractVerifierObjectMapper();
 
 	StubRunnerStreamMessageSelector(Contract groovyDsl) {
-		this.groovyDsl = groovyDsl;
+		this(Collections.singletonList(groovyDsl));
+	}
+
+	StubRunnerStreamMessageSelector(List<Contract> groovyDsls) {
+		this.groovyDsls = groovyDsls;
 	}
 
 	@Override
 	public boolean accept(Message<?> message) {
-		List<String> unmatchedHeaders = headersMatch(message);
+		return matchingContract(message) != null;
+	}
+
+	Contract matchingContract(Message<?> message) {
+		if (CACHE.containsKey(message)) {
+			return CACHE.get(message);
+		}
+		Contract contract = getContract(message);
+		if (contract != null) {
+			CACHE.put(message, contract);
+		}
+		return contract;
+	}
+
+	void updateCache(Message<?> message, Contract contract) {
+		CACHE.put(message, contract);
+	}
+
+	private Contract getContract(Message<?> message) {
+		for (Contract groovyDsl : this.groovyDsls) {
+			Contract contract = matchContract(message, groovyDsl);
+			if (contract != null) {
+				return contract;
+			}
+		}
+		return null;
+	}
+
+	private Contract matchContract(Message<?> message, Contract groovyDsl) {
+		List<String> unmatchedHeaders = headersMatch(message, groovyDsl);
 		if (!unmatchedHeaders.isEmpty()) {
 			if (log.isDebugEnabled()) {
-				log.debug("Contract [" + this.groovyDsl
+				log.debug("Contract [" + groovyDsl
 						+ "] hasn't matched the following headers " + unmatchedHeaders);
 			}
-			return false;
+			return null;
 		}
 		Object inputMessage = message.getPayload();
-		BodyMatchers matchers = this.groovyDsl.getInput().getBodyMatchers();
+		BodyMatchers matchers = groovyDsl.getInput().getBodyMatchers();
 		Object dslBody = MapConverter
-				.getStubSideValues(this.groovyDsl.getInput().getMessageBody());
+				.getStubSideValues(groovyDsl.getInput().getMessageBody());
 		Object matchingInputMessage = JsonToJsonPathsConverter
 				.removeMatchingJsonPaths(dslBody, matchers);
 		JsonPaths jsonPaths = JsonToJsonPathsConverter
@@ -100,11 +138,14 @@ class StubRunnerStreamMessageSelector implements MessageSelector {
 		}
 		if (!unmatchedJsonPath.isEmpty()) {
 			if (log.isDebugEnabled()) {
-				log.debug("Contract [" + this.groovyDsl + "] didn't much the body due to "
+				log.debug("Contract [" + groovyDsl + "] didn't much the body due to "
 						+ unmatchedJsonPath);
 			}
 		}
-		return matches;
+		if (matches) {
+			return groovyDsl;
+		}
+		return null;
 	}
 
 	private boolean matchesJsonPath(List<String> unmatchedJsonPath,
@@ -119,10 +160,10 @@ class StubRunnerStreamMessageSelector implements MessageSelector {
 		}
 	}
 
-	private List<String> headersMatch(Message<?> message) {
+	private List<String> headersMatch(Message<?> message, Contract groovyDsl) {
 		List<String> unmatchedHeaders = new ArrayList<>();
 		Map<String, Object> headers = message.getHeaders();
-		for (Header it : this.groovyDsl.getInput().getMessageHeaders().getEntries()) {
+		for (Header it : groovyDsl.getInput().getMessageHeaders().getEntries()) {
 			String name = it.getName();
 			Object value = it.getClientValue();
 			Object valueInHeader = headers.get(name);
@@ -150,5 +191,4 @@ class StubRunnerStreamMessageSelector implements MessageSelector {
 				? "match pattern [" + ((Pattern) expectedValue).pattern() + "]"
 				: "be equal to [" + expectedValue + "]";
 	}
-
 }

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/stream/StubRunnerStreamTransformer.java

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.contract.stubrunner.messaging.stream;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 import org.springframework.cloud.contract.spec.Contract;
@@ -32,23 +34,35 @@ import org.springframework.messaging.support.MessageBuilder;
  */
 class StubRunnerStreamTransformer implements GenericTransformer<Message<?>, Message<?>> {
 
-	private final Contract groovyDsl;
+	private final StubRunnerStreamMessageSelector selector;
 
 	StubRunnerStreamTransformer(Contract groovyDsl) {
-		this.groovyDsl = groovyDsl;
+		this(Collections.singletonList(groovyDsl));
+	}
+
+	StubRunnerStreamTransformer(List<Contract> groovyDsls) {
+		this.selector = new StubRunnerStreamMessageSelector(groovyDsls);
 	}
 
 	@Override
 	public Message<?> transform(Message<?> source) {
-		if (this.groovyDsl.getOutputMessage() == null) {
+		Contract groovyDsl = matchingContract(source);
+		if (groovyDsl == null || groovyDsl.getOutputMessage() == null) {
 			return source;
 		}
 		String payload = BodyExtractor
-				.extractStubValueFrom(this.groovyDsl.getOutputMessage().getBody());
-		Map<String, Object> headers = this.groovyDsl.getOutputMessage().getHeaders()
+				.extractStubValueFrom(groovyDsl.getOutputMessage().getBody());
+		Map<String, Object> headers = groovyDsl.getOutputMessage().getHeaders()
 				.asStubSideMap();
-		return MessageBuilder.createMessage(payload.getBytes(),
-				new MessageHeaders(headers));
+		MessageHeaders messageHeaders = new MessageHeaders(headers);
+		Message<byte[]> message = MessageBuilder.createMessage(payload.getBytes(),
+				messageHeaders);
+		this.selector.updateCache(message, groovyDsl);
+		return message;
+	}
+
+	Contract matchingContract(Message<?> source) {
+		return this.selector.matchingContract(source);
 	}
 
 }

spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifier.java

@@ -52,3 +52,4 @@ public interface MessageVerifier<M> {
 	M receive(String destination);
 
 }
+

tests/spring-cloud-contract-stub-runner-boot-eureka/src/test/groovy/org/springframework/cloud/contract/stubrunner/spring/cloud/eureka/StubRunnerSpringCloudEurekaAutoConfigurationSpec.groovy

@@ -42,7 +42,7 @@ import spock.util.concurrent.PollingConditions
 /**
  * @author Marcin Grzejszczak
  */
-//TODO: Speed up this test somehow
+//TODO: Speed up this test somehow (move it out of Spring Cloud Contract core to samples)
 @ContextConfiguration(classes = Config, loader = SpringBootContextLoader)
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
 		properties = ["stubrunner.cloud.eureka.enabled=true",

tests/spring-cloud-contract-stub-runner-integration/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/integration/StubRunnerIntegrationTransformerSpec.groovy

@@ -24,7 +24,12 @@ class StubRunnerIntegrationTransformerSpec extends Specification {
 
 	def 'should not transform the message if there is no output message'() {
 		given:
-			StubRunnerIntegrationTransformer transformer = new StubRunnerIntegrationTransformer(noOutputMessageContract)
+			StubRunnerIntegrationTransformer transformer = new StubRunnerIntegrationTransformer(noOutputMessageContract) {
+				@Override
+				Contract matchingContract(Message<?> source) {
+					return noOutputMessageContract
+				}
+			}
 		when:
 			def result = transformer.transform(message)
 		then:
@@ -55,7 +60,12 @@ class StubRunnerIntegrationTransformerSpec extends Specification {
 
 	def 'should convert dsl into message'() {
 		given:
-			StubRunnerIntegrationTransformer transformer = new StubRunnerIntegrationTransformer(dsl)
+			StubRunnerIntegrationTransformer transformer = new StubRunnerIntegrationTransformer(dsl) {
+				@Override
+				Contract matchingContract(Message<?> source) {
+					return dsl
+				}
+			}
 		when:
 			def result = transformer.transform(message)
 		then:
@@ -86,7 +96,12 @@ class StubRunnerIntegrationTransformerSpec extends Specification {
 
 	def 'should convert dsl into message with regex in GString'() {
 		given:
-			StubRunnerIntegrationTransformer transformer = new StubRunnerIntegrationTransformer(dslWithRegexInGString)
+			StubRunnerIntegrationTransformer transformer = new StubRunnerIntegrationTransformer(dslWithRegexInGString) {
+				@Override
+				Contract matchingContract(Message<?> source) {
+					return dslWithRegexInGString
+				}
+			}
 		when:
 			def result = transformer.transform(message)
 		then:

tests/spring-cloud-contract-stub-runner-stream/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/stream/StubRunnerStreamTransformerSpec.groovy

@@ -55,7 +55,12 @@ class StubRunnerStreamTransformerSpec extends Specification {
 
 	def 'should convert dsl into message'() {
 		given:
-			StubRunnerStreamTransformer streamTransformer = new StubRunnerStreamTransformer(dsl)
+			StubRunnerStreamTransformer streamTransformer = new StubRunnerStreamTransformer(dsl) {
+				@Override
+				Contract matchingContract(Message<?> source) {
+					return dsl
+				}
+			}
 		when:
 			def result = streamTransformer.transform(message)
 		then:
@@ -86,7 +91,12 @@ class StubRunnerStreamTransformerSpec extends Specification {
 
 	def 'should convert dsl into message with regex in GString'() {
 		given:
-			StubRunnerStreamTransformer streamTransformer = new StubRunnerStreamTransformer(dslWithRegexInGString)
+			StubRunnerStreamTransformer streamTransformer = new StubRunnerStreamTransformer(dslWithRegexInGString) {
+				@Override
+				Contract matchingContract(Message<?> source) {
+					return dslWithRegexInGString
+				}
+			}
 		when:
 			def result = streamTransformer.transform(message)
 		then:
@@ -122,7 +132,12 @@ class StubRunnerStreamTransformerSpec extends Specification {
 					)
 				}
 			}
-			StubRunnerStreamTransformer streamTransformer = new StubRunnerStreamTransformer(contract)
+			StubRunnerStreamTransformer streamTransformer = new StubRunnerStreamTransformer(contract) {
+				@Override
+				Contract matchingContract(Message<?> source) {
+					return contract
+				}
+			}
 		when:
 			def result = streamTransformer.transform(message)
 		then: