Ensures that the proper channel will be picked for the same destination

fixes gh-694

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

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.contract.verifier.messaging.stream;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
@@ -29,6 +30,7 @@ import org.springframework.cloud.stream.test.binder.MessageCollector;
 import org.springframework.context.ApplicationContext;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.util.StringUtils;
 
 /**
  * @author Marcin Grzejszczak
@@ -56,7 +58,7 @@ public class StreamStubMessages implements MessageVerifier<Message<?>> {
 	public void send(Message<?> message, String destination) {
 		try {
 			MessageChannel messageChannel = this.context
-					.getBean(resolvedDestination(destination), MessageChannel.class);
+					.getBean(resolvedDestination(destination, DefaultChannels.OUTPUT), MessageChannel.class);
 			messageChannel.send(message);
 		}
 		catch (Exception e) {
@@ -70,7 +72,7 @@ public class StreamStubMessages implements MessageVerifier<Message<?>> {
 	public Message<?> receive(String destination, long timeout, TimeUnit timeUnit) {
 		try {
 			MessageChannel messageChannel = this.context
-					.getBean(resolvedDestination(destination), MessageChannel.class);
+					.getBean(resolvedDestination(destination, DefaultChannels.INPUT), MessageChannel.class);
 			return this.messageCollector.forChannel(messageChannel).poll(timeout, timeUnit);
 		}
 		catch (Exception e) {
@@ -80,10 +82,11 @@ public class StreamStubMessages implements MessageVerifier<Message<?>> {
 		}
 	}
 
-	private String resolvedDestination(String destination) {
+	private String resolvedDestination(String destination, DefaultChannels defaultChannel) {
 		try {
 			BindingServiceProperties channelBindingServiceProperties = this.context
 					.getBean(BindingServiceProperties.class);
+			Map<String, String> channels = new HashMap<>();
 			for (Map.Entry<String, BindingProperties> entry : channelBindingServiceProperties
 					.getBindings().entrySet()) {
 				if (destination.equals(entry.getValue().getDestination())) {
@@ -91,9 +94,24 @@ public class StreamStubMessages implements MessageVerifier<Message<?>> {
 						log.debug("Found a channel named [{}] with destination [{}]",
 								entry.getKey(), destination);
 					}
-					return entry.getKey();
+					channels.put(entry.getKey().toLowerCase(), destination);
 				}
 			}
+			if (channels.size() == 1) {
+				return channels.keySet().iterator().next();
+			} else if (channels.size() > 0) {
+				if (log.isDebugEnabled()) {
+					log.debug("Found following channels [{}] for destination [{}]. "
+									+ "Will pick the one that matches the default channel name or the first one if none is matching",
+							channels, destination);
+				}
+				String defaultChannelName = channels.get(defaultChannel.name().toLowerCase());
+				String matchingChannelName = StringUtils.hasText(defaultChannelName) ? defaultChannel.name().toLowerCase() : channels.keySet().iterator().next();
+				if (log.isDebugEnabled()) {
+					log.debug("Picked channel name is [{}]", matchingChannelName);
+				}
+				return matchingChannelName;
+			}
 		} catch (Exception e) {
 			log.error("Exception took place while trying to resolve the destination. Will assume the name [" + destination + "]", e);
 		}
@@ -111,3 +129,8 @@ public class StreamStubMessages implements MessageVerifier<Message<?>> {
 	}
 
 }
+
+
+enum DefaultChannels {
+	INPUT, OUTPUT
+}

spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessagesSpec.groovy

@@ -0,0 +1,109 @@
+package org.springframework.cloud.contract.verifier.messaging.stream
+
+import spock.lang.Issue
+import spock.lang.Specification
+
+import org.springframework.cloud.stream.config.BindingProperties
+import org.springframework.cloud.stream.config.BindingServiceProperties
+import org.springframework.cloud.stream.test.binder.MessageCollector
+import org.springframework.context.ApplicationContext
+import org.springframework.messaging.MessageChannel
+
+/**
+ * @author Marcin Grzejszczak
+ */
+class StreamStubMessagesSpec extends Specification {
+
+	@Issue("694")
+	def "should resolve input channel if input and output have same destination and receive is called"() {
+		given:
+			ApplicationContext applicationContext = Mock(ApplicationContext)
+			BindingServiceProperties properties = new BindingServiceProperties(
+					bindings: [
+							input: new BindingProperties(destination: "verifications"),
+							output: new BindingProperties(destination: "verifications"),
+					]
+			)
+			MessageCollector collector = Stub(MessageCollector)
+		and:
+			applicationContext.getBean(BindingServiceProperties) >> properties
+			applicationContext.getBean(MessageCollector) >> collector
+		and:
+			StreamStubMessages messages = new StreamStubMessages(applicationContext)
+		when:
+			messages.receive("verifications")
+		then:
+			1 * applicationContext.getBean("input", MessageChannel) >> null
+	}
+
+	@Issue("694")
+	def "should resolve output channel if input and output have same destination and send is called"() {
+		given:
+			ApplicationContext applicationContext = Mock(ApplicationContext)
+			BindingServiceProperties properties = new BindingServiceProperties(
+					bindings: [
+							input: new BindingProperties(destination: "verifications"),
+							output: new BindingProperties(destination: "verifications"),
+					]
+			)
+			MessageCollector collector = Stub(MessageCollector)
+			MessageChannel channel = Stub(MessageChannel)
+		and:
+			applicationContext.getBean(BindingServiceProperties) >> properties
+			applicationContext.getBean(MessageCollector) >> collector
+		and:
+			StreamStubMessages messages = new StreamStubMessages(applicationContext)
+		when:
+			messages.send("foo", [:], "verifications")
+		then:
+			1 * applicationContext.getBean("output", MessageChannel) >> channel
+	}
+
+	def "should resolve channel via destination for send and receive"() {
+		given:
+			ApplicationContext applicationContext = Mock(ApplicationContext)
+			BindingServiceProperties properties = new BindingServiceProperties(
+					bindings: [
+							foo: new BindingProperties(destination: "verifications")
+					]
+			)
+			MessageCollector collector = Stub(MessageCollector)
+			MessageChannel channel = Stub(MessageChannel)
+		and:
+			applicationContext.getBean(BindingServiceProperties) >> properties
+			applicationContext.getBean(MessageCollector) >> collector
+		and:
+			StreamStubMessages messages = new StreamStubMessages(applicationContext)
+		when:
+			messageInteraction(messages)
+		then:
+			1 * applicationContext.getBean("foo", MessageChannel) >> channel
+		where:
+			messageInteraction << [ { StreamStubMessages stream -> stream.send("foo", [:], "verifications")},
+									{ StreamStubMessages stream -> stream.receive("verifications")}]
+	}
+
+	def "should resolve channel via channel name for send and receive"() {
+		given:
+			ApplicationContext applicationContext = Mock(ApplicationContext)
+			BindingServiceProperties properties = new BindingServiceProperties(
+					bindings: [
+							verifications: new BindingProperties(destination: "bar")
+					]
+			)
+			MessageCollector collector = Stub(MessageCollector)
+			MessageChannel channel = Stub(MessageChannel)
+		and:
+			applicationContext.getBean(BindingServiceProperties) >> properties
+			applicationContext.getBean(MessageCollector) >> collector
+		and:
+			StreamStubMessages messages = new StreamStubMessages(applicationContext)
+		when:
+			messageInteraction(messages)
+		then:
+			1 * applicationContext.getBean("verifications", MessageChannel) >> channel
+		where:
+			messageInteraction << [ { StreamStubMessages stream -> stream.send("foo", [:], "verifications")},
+									{ StreamStubMessages stream -> stream.receive("verifications")}]
+	}
+}

tests/samples-messaging-stream/src/test/groovy/com/example/StreamMessagingApplicationSpec.groovy

@@ -40,7 +40,7 @@ import javax.inject.Inject
 @DirtiesContext
 @SpringBootTest(properties = "debug=true")
 @AutoConfigureMessageVerifier
-public class StreamMessagingApplicationSpec extends Specification {
+class StreamMessagingApplicationSpec extends Specification {
 
 	// ALL CASES
 	@Inject MessageVerifier<Message<?>> contractVerifierMessaging