From e65aa5d6d5db70fc56f5f38f4c192ca6cf802084 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Wed, 1 Aug 2018 12:54:23 +0200 Subject: [PATCH] Ensures that the proper channel will be picked for the same destination fixes gh-694 --- .../messaging/stream/StreamStubMessages.java | 31 ++++- .../stream/StreamStubMessagesSpec.groovy | 109 ++++++++++++++++++ .../StreamMessagingApplicationSpec.groovy | 2 +- 3 files changed, 137 insertions(+), 5 deletions(-) create mode 100644 spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessagesSpec.groovy diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessages.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessages.java index 00cebd3016..4122cac3f9 100644 --- a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessages.java +++ b/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> { 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> { 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> { } } - private String resolvedDestination(String destination) { + private String resolvedDestination(String destination, DefaultChannels defaultChannel) { try { BindingServiceProperties channelBindingServiceProperties = this.context .getBean(BindingServiceProperties.class); + Map channels = new HashMap<>(); for (Map.Entry entry : channelBindingServiceProperties .getBindings().entrySet()) { if (destination.equals(entry.getValue().getDestination())) { @@ -91,9 +94,24 @@ public class StreamStubMessages implements MessageVerifier> { 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> { } } + + +enum DefaultChannels { + INPUT, OUTPUT +} \ No newline at end of file diff --git a/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessagesSpec.groovy b/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessagesSpec.groovy new file mode 100644 index 0000000000..8cfcd09210 --- /dev/null +++ b/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")}] + } +} diff --git a/tests/samples-messaging-stream/src/test/groovy/com/example/StreamMessagingApplicationSpec.groovy b/tests/samples-messaging-stream/src/test/groovy/com/example/StreamMessagingApplicationSpec.groovy index 71f804412d..abcf7f3a9b 100644 --- a/tests/samples-messaging-stream/src/test/groovy/com/example/StreamMessagingApplicationSpec.groovy +++ b/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> contractVerifierMessaging