diff --git a/pom.xml b/pom.xml index 35f0d18d8a..6ce3af1c8a 100644 --- a/pom.xml +++ b/pom.xml @@ -31,6 +31,8 @@ 2.2.1.BUILD-SNAPSHOT 2.2.1.BUILD-SNAPSHOT Horsham.RELEASE + + 3.0.1.BUILD-SNAPSHOT 2.2.1.BUILD-SNAPSHOT 2.2.1.BUILD-SNAPSHOT 2.2.1.BUILD-SNAPSHOT diff --git a/spring-cloud-contract-verifier/pom.xml b/spring-cloud-contract-verifier/pom.xml index ad8cc70b69..333eb54a34 100644 --- a/spring-cloud-contract-verifier/pom.xml +++ b/spring-cloud-contract-verifier/pom.xml @@ -83,6 +83,15 @@ spring-cloud-stream-test-support true + + org.springframework.cloud + spring-cloud-stream + test-jar + test-binder + true + + ${spring-cloud-stream-test-binder.version} + org.springframework.amqp spring-rabbit diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/ContractVerifierStreamAutoConfiguration.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/ContractVerifierStreamAutoConfiguration.java index 4d7378bc58..d901feee25 100644 --- a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/ContractVerifierStreamAutoConfiguration.java +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/ContractVerifierStreamAutoConfiguration.java @@ -26,6 +26,7 @@ import org.springframework.cloud.contract.verifier.messaging.internal.ContractVe import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessaging; import org.springframework.cloud.contract.verifier.messaging.noop.NoOpContractVerifierAutoConfiguration; import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.test.binder.MessageCollector; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; @@ -50,6 +51,22 @@ public class ContractVerifierStreamAutoConfiguration { return new ContractVerifierHelper(exchange); } + @Configuration(proxyBeanMethods = false) + @ConditionalOnClass(InputDestination.class) + @ConditionalOnMissingClass("org.springframework.cloud.stream.test.binder.MessageCollector") + static class InputDestinationConfiguration { + + @Bean + @ConditionalOnMissingBean + MessageVerifier> contractVerifierMessageExchangeWithDestinations( + ApplicationContext context) { + return new StreamStubMessages( + new StreamInputDestinationMessageSender(context), + new StreamOutputDestinationMessageReceiver(context)); + } + + } + @Configuration(proxyBeanMethods = false) @ConditionalOnClass(MessageCollector.class) static class MessageCollectorConfiguration { @@ -67,7 +84,9 @@ public class ContractVerifierStreamAutoConfiguration { } @Configuration(proxyBeanMethods = false) - @ConditionalOnMissingClass("org.springframework.cloud.stream.test.binder.MessageCollector") + @ConditionalOnMissingClass({ + "org.springframework.cloud.stream.test.binder.MessageCollector", + "org.springframework.cloud.stream.binder.test.InputDestination" }) static class NoMessageCollectorClassConfiguration { @Bean diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamInputDestinationMessageSender.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamInputDestinationMessageSender.java new file mode 100644 index 0000000000..6eead0ae1e --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamInputDestinationMessageSender.java @@ -0,0 +1,86 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.cloud.contract.verifier.messaging.stream; + +import java.util.Arrays; +import java.util.Map; +import java.util.stream.Collectors; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.contract.verifier.messaging.MessageVerifierSender; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.function.StreamFunctionProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.messaging.Message; +import org.springframework.util.StringUtils; + +/** + * @author Marcin Grzejszczak + */ +class StreamInputDestinationMessageSender implements MessageVerifierSender> { + + private static final Log log = LogFactory + .getLog(StreamInputDestinationMessageSender.class); + + private final ApplicationContext context; + + private final ContractVerifierStreamMessageBuilder builder = new ContractVerifierStreamMessageBuilder(); + + StreamInputDestinationMessageSender(ApplicationContext context) { + this.context = context; + } + + @Override + public void send(T payload, Map headers, String destination) { + send(this.builder.create(payload, headers), destination); + } + + @Override + public void send(Message message, String destination) { + try { + InputDestination inputDestination = this.context + .getBean(InputDestination.class); + StreamFunctionProperties streamFunctionProperties = this.context + .getBean(StreamFunctionProperties.class); + int indexOfDestination = StringUtils + .isEmpty(streamFunctionProperties.getDefinition()) ? 0 + : indexOfDestination(streamFunctionProperties, destination); + inputDestination.send(message, indexOfDestination); + } + catch (Exception e) { + log.error("Exception occurred while trying to send a message [" + message + + "] " + "to a destination with name [" + destination + "]", e); + throw e; + } + } + + private int indexOfDestination(StreamFunctionProperties streamFunctionProperties, + String destination) { + String[] split = streamFunctionProperties.getDefinition().split(";"); + int indexOfDestination = Arrays.stream(split).map(String::toLowerCase) + .collect(Collectors.toList()).indexOf(destination.toLowerCase()); + if (indexOfDestination == -1) { + throw new IllegalStateException("Destination with name [" + destination + + "] not found in the function definitions [" + + streamFunctionProperties.getDefinition() + "]"); + } + return indexOfDestination; + } + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamOutputDestinationMessageReceiver.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamOutputDestinationMessageReceiver.java new file mode 100644 index 0000000000..9e5a36c51a --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamOutputDestinationMessageReceiver.java @@ -0,0 +1,83 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.cloud.contract.verifier.messaging.stream; + +import java.util.Arrays; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.contract.verifier.messaging.MessageVerifierReceiver; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.function.StreamFunctionProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.messaging.Message; +import org.springframework.util.StringUtils; + +class StreamOutputDestinationMessageReceiver + implements MessageVerifierReceiver> { + + private static final Log log = LogFactory + .getLog(StreamOutputDestinationMessageReceiver.class); + + private final ApplicationContext context; + + StreamOutputDestinationMessageReceiver(ApplicationContext context) { + this.context = context; + } + + @Override + public Message receive(String destination, long timeout, TimeUnit timeUnit) { + try { + OutputDestination outputDestination = this.context + .getBean(OutputDestination.class); + StreamFunctionProperties streamFunctionProperties = this.context + .getBean(StreamFunctionProperties.class); + int indexOfDestination = StringUtils + .isEmpty(streamFunctionProperties.getDefinition()) ? 0 + : indexOfDestination(streamFunctionProperties, destination); + return outputDestination.receive(timeUnit.toMillis(timeout), + indexOfDestination); + } + catch (Exception e) { + log.error("Exception occurred while trying to read a message from " + + " a channel with name [" + destination + "]", e); + throw new IllegalStateException(e); + } + } + + private int indexOfDestination(StreamFunctionProperties streamFunctionProperties, + String destination) { + String[] split = streamFunctionProperties.getDefinition().split(";"); + int indexOfDestination = Arrays.stream(split).map(String::toLowerCase) + .collect(Collectors.toList()).indexOf(destination.toLowerCase()); + if (indexOfDestination == -1) { + throw new IllegalStateException("Destination with name [" + destination + + "] not found in the function definitions [" + + streamFunctionProperties.getDefinition() + "]"); + } + return indexOfDestination; + } + + @Override + public Message receive(String destination) { + return receive(destination, 5, TimeUnit.SECONDS); + } + +} diff --git a/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessagesWithDestinationsSpec.groovy b/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessagesWithDestinationsSpec.groovy new file mode 100644 index 0000000000..ddc8de0621 --- /dev/null +++ b/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessagesWithDestinationsSpec.groovy @@ -0,0 +1,100 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.cloud.contract.verifier.messaging.stream + +import spock.lang.Issue +import spock.lang.Specification + +import org.springframework.cloud.stream.binder.test.InputDestination +import org.springframework.cloud.stream.binder.test.OutputDestination +import org.springframework.cloud.stream.function.StreamFunctionProperties +import org.springframework.context.ApplicationContext +/** + * @author Marcin Grzejszczak + */ +class StreamStubMessagesWithDestinationsSpec 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) + StreamFunctionProperties functionProperties = new StreamFunctionProperties(definition: 'verifications') + OutputDestination output = Mock(OutputDestination) + and: + applicationContext.getBean(OutputDestination) >> output + applicationContext.getBean(StreamFunctionProperties) >> functionProperties + and: + StreamOutputDestinationMessageReceiver messages = new StreamOutputDestinationMessageReceiver(applicationContext) + when: + messages.receive("verifications") + then: + 1 * output.receive(5000, 0) >> null + } + + @Issue("694") + def "should resolve output channel if input and output have same destination and send is called"() { + given: + ApplicationContext applicationContext = Mock(ApplicationContext) + StreamFunctionProperties functionProperties = new StreamFunctionProperties(definition: 'verifications') + InputDestination input = Mock(InputDestination) + and: + applicationContext.getBean(InputDestination) >> input + applicationContext.getBean(StreamFunctionProperties) >> functionProperties + and: + StreamInputDestinationMessageSender messages = new StreamInputDestinationMessageSender(applicationContext) + when: + messages.send("foo", [:], "verifications") + then: + 1 * input.send(_, 0) + } + + def "should resolve channel via destination for send"() { + given: + ApplicationContext applicationContext = Mock(ApplicationContext) + StreamFunctionProperties functionProperties = new StreamFunctionProperties(definition: 'verifications') + InputDestination input = Mock(InputDestination) + OutputDestination output = Mock(OutputDestination) + and: + applicationContext.getBean(InputDestination) >> input + applicationContext.getBean(OutputDestination) >> output + applicationContext.getBean(StreamFunctionProperties) >> functionProperties + and: + StreamStubMessages messages = new StreamStubMessages(new StreamInputDestinationMessageSender(applicationContext), new StreamOutputDestinationMessageReceiver(applicationContext)) + when: + messages.send("foo", [:], "verifications") + then: + 1 * input.send(_, 0) + } + + def "should resolve channel via destination for receive"() { + given: + ApplicationContext applicationContext = Mock(ApplicationContext) + StreamFunctionProperties functionProperties = new StreamFunctionProperties(definition: 'verifications') + InputDestination input = Mock(InputDestination) + OutputDestination output = Mock(OutputDestination) + and: + applicationContext.getBean(InputDestination) >> input + applicationContext.getBean(OutputDestination) >> output + applicationContext.getBean(StreamFunctionProperties) >> functionProperties + and: + StreamStubMessages messages = new StreamStubMessages(new StreamInputDestinationMessageSender(applicationContext), new StreamOutputDestinationMessageReceiver(applicationContext)) + when: + messages.receive("verifications") + then: + 1 * output.receive(5000, 0) >> null + } +}