SpringCloud/spring-cloud-contract
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added support for the new spring-cloud-stream test binder mechanism

Browse Source 
fixes gh-1278
This commit is contained in:
Marcin Grzejszczak
2019-12-17 11:19:27 +01:00
parent cb2393678d
commit d1daa08e78
6 changed files with 300 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pom.xml
View File

@@ -31,6 +31,8 @@
<spring-cloud-build.version>2.2.1.BUILD-SNAPSHOT</spring-cloud-build.version>
<spring-cloud-zookeeper.version>2.2.1.BUILD-SNAPSHOT</spring-cloud-zookeeper.version>
<spring-cloud-stream.version>Horsham.RELEASE</spring-cloud-stream.version>
<!-- Needs to come from a sc-stream-bom -->
<spring-cloud-stream-test-binder.version>3.0.1.BUILD-SNAPSHOT</spring-cloud-stream-test-binder.version>
<spring-cloud-netflix.version>2.2.1.BUILD-SNAPSHOT</spring-cloud-netflix.version>
<spring-cloud-consul.version>2.2.1.BUILD-SNAPSHOT</spring-cloud-consul.version>
<spring-cloud-commons.version>2.2.1.BUILD-SNAPSHOT</spring-cloud-commons.version>

9
spring-cloud-contract-verifier/pom.xml
View File

@@ -83,6 +83,15 @@
<artifactId>spring-cloud-stream-test-support</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
<type>test-jar</type>
<classifier>test-binder</classifier>
<optional>true</optional>
<!-- TODO: Has to come from a BOM-->
<version>${spring-cloud-stream-test-binder.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>

21
spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/ContractVerifierStreamAutoConfiguration.java
View File

@@ -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<Message<?>> 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

86
spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamInputDestinationMessageSender.java Normal file
View File

@@ -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<Message<?>> {
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 <T> void send(T payload, Map<String, Object> 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;
}
}

83
spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamOutputDestinationMessageReceiver.java Normal file
View File

@@ -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<Message<?>> {
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);
}
}

100
spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessagesWithDestinationsSpec.groovy Normal file
View File

@@ -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
}
}