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

Added back support for interacting with a running broker via stub runner

Browse Source
This commit is contained in:
Marcin Grzejszczak
2019-07-09 17:46:51 +02:00
parent ce4140a9ff
commit f48895c189
12 changed files with 544 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,9 +16,6 @@
package org.springframework.cloud.contract.verifier.messaging;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* Core interface that allows you to build, send and receive messages.
*
@@ -29,39 +26,7 @@ import java.util.concurrent.TimeUnit;
* @author Marcin Grzejszczak
* @since 1.0.0
*/
public interface MessageVerifier<M> {
/**
* Sends the message to the given destination.
* @param message to send
* @param destination destination to which the message will be sent
*/
void send(M message, String destination);
/**
* Sends the given payload with headers, to the given destination.
* @param <T> payload type
* @param payload payload to send
* @param headers headers to send
* @param destination destination to which the message will be sent
*/
<T> void send(T payload, Map<String, Object> headers, String destination);
/**
* Receives the message from the given destination. You can provide the timeout for
* receiving that message.
* @param destination destination from which the message will be received
* @param timeout timeout to wait for the message
* @param timeUnit param to define the unit of timeout
* @return received message
*/
M receive(String destination, long timeout, TimeUnit timeUnit);
/**
* Receives the message from the given destination. A default timeout will be applied.
* @param destination destination from which the message will be received
* @return received message
*/
M receive(String destination);
public interface MessageVerifier<M>
extends MessageVerifierSender<M>, MessageVerifierReceiver<M> {
}

50
spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierReceiver.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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;
import java.util.concurrent.TimeUnit;
/**
* Core interface that allows you to receive messages.
*
* Destination is relevant to the underlying implementation. Might be a channel, queue,
* topic etc.
*
* @param <M> message type
* @author Marcin Grzejszczak
* @since 2.2.0
*/
public interface MessageVerifierReceiver<M> {
/**
* Receives the message from the given destination. You can provide the timeout for
* receiving that message.
* @param destination destination from which the message will be received
* @param timeout timeout to wait for the message
* @param timeUnit param to define the unit of timeout
* @return received message
*/
M receive(String destination, long timeout, TimeUnit timeUnit);
/**
* Receives the message from the given destination. A default timeout will be applied.
* @param destination destination from which the message will be received
* @return received message
*/
M receive(String destination);
}

49
spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierSender.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* 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;
import java.util.Map;
/**
* Core interface that allows you to send messages.
*
* Destination is relevant to the underlying implementation. Might be a channel, queue,
* topic etc.
*
* @param <M> message type
* @author Marcin Grzejszczak
* @since 2.2.0
*/
public interface MessageVerifierSender<M> {
/**
* Sends the message to the given destination.
* @param message to send
* @param destination destination to which the message will be sent
*/
void send(M message, String destination);
/**
* Sends the given payload with headers, to the given destination.
* @param <T> payload type
* @param payload payload to send
* @param headers headers to send
* @param destination destination to which the message will be sent
*/
<T> void send(T payload, Map<String, Object> headers, String destination);
}

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

@@ -19,12 +19,14 @@ package org.springframework.cloud.contract.verifier.messaging.stream;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cloud.contract.verifier.messaging.MessageVerifier;
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessage;
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.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
@@ -36,19 +38,12 @@ import org.springframework.util.Assert;
* @author Marcin Grzejszczak
*/
@Configuration
@ConditionalOnClass({ EnableBinding.class, MessageCollector.class })
@ConditionalOnClass(EnableBinding.class)
@ConditionalOnProperty(name = "stubrunner.stream.enabled", havingValue = "true",
matchIfMissing = true)
@AutoConfigureBefore(NoOpContractVerifierAutoConfiguration.class)
public class ContractVerifierStreamAutoConfiguration {
@Bean
@ConditionalOnMissingBean
MessageVerifier<Message<?>> contractVerifierMessageExchange(
ApplicationContext applicationContext) {
return new StreamStubMessages(applicationContext);
}
@Bean
@ConditionalOnMissingBean
public ContractVerifierMessaging<?> contractVerifierMessagingConverter(
@@ -56,6 +51,37 @@ public class ContractVerifierStreamAutoConfiguration {
return new ContractVerifierHelper(exchange);
}
@Configuration
@ConditionalOnClass(MessageCollector.class)
static class MessageCollectorConfiguration {
@Bean
@ConditionalOnMissingBean
MessageVerifier<Message<?>> contractVerifierMessageExchangeWithMessageCollector(
ApplicationContext context) {
DestinationResolver resolver = new DestinationResolver(context);
return new StreamStubMessages(
new StreamFromBinderMappingMessageSender(context, resolver),
new StreamMessageCollectorMessageReceiver(resolver, context));
}
}
@Configuration
@ConditionalOnMissingClass("org.springframework.cloud.stream.test.binder.MessageCollector")
static class NoMessageCollectorClassConfiguration {
@Bean
@ConditionalOnMissingBean
MessageVerifier<Message<?>> contractVerifierMessageExchangeWithNoMessageCollector(
ApplicationContext applicationContext,
BinderAwareChannelResolver resolver) {
return new StreamStubMessages(new StreamStubMessageSender(resolver),
new StreamPollableChannelMessageReceiver(applicationContext));
}
}
}
class ContractVerifierHelper extends ContractVerifierMessaging<Message<?>> {

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

@@ -0,0 +1,23 @@
/*
* 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;
enum DefaultChannels {
INPUT, OUTPUT
}

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

@@ -0,0 +1,88 @@
/*
* 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.HashMap;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.util.StringUtils;
class DestinationResolver {
private static final Log log = LogFactory.getLog(DestinationResolver.class);
private final ApplicationContext context;
DestinationResolver(ApplicationContext context) {
this.context = context;
}
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())) {
if (log.isDebugEnabled()) {
log.debug("Found a channel named [" + entry.getKey()
+ "] with destination [" + destination + "]");
}
channels.put(entry.getKey(), destination);
}
}
if (channels.size() == 1) {
return channels.keySet().iterator().next();
}
else if (channels.size() > 0) {
if (log.isDebugEnabled()) {
log.debug("Found following channels [" + channels
+ "] for destination [" + destination + "]. "
+ "Will pick the one that matches the default channel name or the first one if none is matching");
}
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);
}
if (log.isDebugEnabled()) {
log.debug("No destination named [" + destination
+ "] was found. Assuming that the destination equals the channel name");
}
return destination;
}
}

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

@@ -0,0 +1,69 @@
/*
* 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.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.contract.verifier.messaging.MessageVerifierSender;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
/**
* @author Marcin Grzejszczak
*/
class StreamFromBinderMappingMessageSender implements MessageVerifierSender<Message<?>> {
private static final Log log = LogFactory
.getLog(StreamFromBinderMappingMessageSender.class);
private final ApplicationContext context;
private final DestinationResolver resolver;
private final ContractVerifierStreamMessageBuilder builder = new ContractVerifierStreamMessageBuilder();
StreamFromBinderMappingMessageSender(ApplicationContext context,
DestinationResolver resolver) {
this.context = context;
this.resolver = resolver;
}
@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 {
MessageChannel messageChannel = this.context.getBean(this.resolver
.resolvedDestination(destination, DefaultChannels.OUTPUT),
MessageChannel.class);
messageChannel.send(message);
}
catch (Exception e) {
log.error("Exception occurred while trying to send a message [" + message
+ "] " + "to a channel with name [" + destination + "]", e);
throw e;
}
}
}

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

@@ -0,0 +1,75 @@
/*
* 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.concurrent.TimeUnit;
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.test.binder.MessageCollector;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
class StreamMessageCollectorMessageReceiver
implements MessageVerifierReceiver<Message<?>> {
private static final Log log = LogFactory
.getLog(StreamMessageCollectorMessageReceiver.class);
private final DestinationResolver resolver;
private MessageCollector messageCollector;
private final ApplicationContext context;
StreamMessageCollectorMessageReceiver(DestinationResolver resolver,
ApplicationContext context) {
this.resolver = resolver;
this.context = context;
}
@Override
public Message<?> receive(String destination, long timeout, TimeUnit timeUnit) {
try {
MessageChannel messageChannel = this.context.getBean(
this.resolver.resolvedDestination(destination, DefaultChannels.INPUT),
MessageChannel.class);
return messageCollector().forChannel(messageChannel).poll(timeout, timeUnit);
}
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 MessageCollector messageCollector() {
if (this.messageCollector == null) {
this.messageCollector = context.getBean(MessageCollector.class);
}
return this.messageCollector;
}
@Override
public Message<?> receive(String destination) {
return receive(destination, 5, TimeUnit.SECONDS);
}
}

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

@@ -0,0 +1,64 @@
/*
* 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.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.contract.verifier.messaging.MessageVerifierReceiver;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
class StreamPollableChannelMessageReceiver
implements MessageVerifierReceiver<Message<?>> {
private static final Log log = LogFactory
.getLog(StreamPollableChannelMessageReceiver.class);
private final ApplicationContext context;
private final DestinationResolver destinationResolver;
StreamPollableChannelMessageReceiver(ApplicationContext context) {
this.context = context;
this.destinationResolver = new DestinationResolver(context);
}
@Override
public Message<?> receive(String destination, long timeout, TimeUnit timeUnit) {
try {
PollableChannel messageChannel = this.context.getBean(this.destinationResolver
.resolvedDestination(destination, DefaultChannels.INPUT),
PollableChannel.class);
return messageChannel.receive(timeUnit.toMillis(timeout));
}
catch (Exception e) {
log.error("Exception occurred while trying to read a message from "
+ " a channel with name [" + destination + "]", e);
throw new IllegalStateException(e);
}
}
@Override
public Message<?> receive(String destination) {
return receive(destination, 5, TimeUnit.SECONDS);
}
}

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

@@ -0,0 +1,62 @@
/*
* 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.Map;
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.binding.BinderAwareChannelResolver;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
/**
* @author Marcin Grzejszczak
*/
class StreamStubMessageSender implements MessageVerifierSender<Message<?>> {
private static final Log log = LogFactory.getLog(StreamStubMessageSender.class);
private final BinderAwareChannelResolver resolver;
private final ContractVerifierStreamMessageBuilder builder = new ContractVerifierStreamMessageBuilder();
StreamStubMessageSender(BinderAwareChannelResolver resolver) {
this.resolver = resolver;
}
@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 {
MessageChannel messageChannel = this.resolver.resolveDestination(destination);
messageChannel.send(message);
}
catch (Exception e) {
log.error("Exception occurred while trying to send a message [" + message
+ "] " + "to a channel with name [" + destination + "]", e);
throw e;
}
}
}

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

@@ -16,139 +16,47 @@
package org.springframework.cloud.contract.verifier.messaging.stream;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.contract.verifier.messaging.MessageVerifier;
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.cloud.contract.verifier.messaging.MessageVerifierReceiver;
import org.springframework.cloud.contract.verifier.messaging.MessageVerifierSender;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.StringUtils;
enum DefaultChannels {
INPUT, OUTPUT
}
/**
* @author Marcin Grzejszczak
*/
public class StreamStubMessages implements MessageVerifier<Message<?>> {
private static final Log log = LogFactory.getLog(StreamStubMessages.class);
private final MessageVerifierSender<Message<?>> sender;
private final ApplicationContext context;
private final MessageVerifierReceiver<Message<?>> receiver;
private MessageCollector messageCollector;
private final ContractVerifierStreamMessageBuilder builder = new ContractVerifierStreamMessageBuilder();
public StreamStubMessages(ApplicationContext context) {
this.context = context;
public StreamStubMessages(MessageVerifierSender<Message<?>> sender,
MessageVerifierReceiver<Message<?>> receiver) {
this.sender = sender;
this.receiver = receiver;
}
@Override
public <T> void send(T payload, Map<String, Object> headers, String destination) {
send(this.builder.create(payload, headers), destination);
this.sender.send(payload, headers, destination);
}
@Override
public void send(Message<?> message, String destination) {
try {
MessageChannel messageChannel = this.context.getBean(
resolvedDestination(destination, DefaultChannels.OUTPUT),
MessageChannel.class);
messageChannel.send(message);
}
catch (Exception e) {
log.error("Exception occurred while trying to send a message [" + message
+ "] " + "to a channel with name [" + destination + "]", e);
throw e;
}
this.sender.send(message, destination);
}
@Override
public Message<?> receive(String destination, long timeout, TimeUnit timeUnit) {
try {
MessageChannel messageChannel = this.context.getBean(
resolvedDestination(destination, DefaultChannels.INPUT),
MessageChannel.class);
return messageCollector().forChannel(messageChannel).poll(timeout, timeUnit);
}
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 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())) {
if (log.isDebugEnabled()) {
log.debug("Found a channel named [" + entry.getKey()
+ "] with destination [" + destination + "]");
}
channels.put(entry.getKey(), destination);
}
}
if (channels.size() == 1) {
return channels.keySet().iterator().next();
}
else if (channels.size() > 0) {
if (log.isDebugEnabled()) {
log.debug("Found following channels [" + channels
+ "] for destination [" + destination + "]. "
+ "Will pick the one that matches the default channel name or the first one if none is matching");
}
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);
}
if (log.isDebugEnabled()) {
log.debug("No destination named [" + destination
+ "] was found. Assuming that the destination equals the channel name");
}
return destination;
return this.receiver.receive(destination, timeout, timeUnit);
}
@Override
public Message<?> receive(String destination) {
return receive(destination, 5, TimeUnit.SECONDS);
}
private MessageCollector messageCollector() {
if (this.messageCollector == null) {
this.messageCollector = context.getBean(MessageCollector.class);
}
return this.messageCollector;
return this.receiver.receive(destination);
}
}

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

@@ -45,7 +45,8 @@ class StreamStubMessagesSpec extends Specification {
applicationContext.getBean(BindingServiceProperties) >> properties
applicationContext.getBean(MessageCollector) >> collector
and:
StreamStubMessages messages = new StreamStubMessages(applicationContext)
DestinationResolver resolver = new DestinationResolver(applicationContext)
StreamMessageCollectorMessageReceiver messages = new StreamMessageCollectorMessageReceiver(resolver, applicationContext)
when:
messages.receive("verifications")
then:
@@ -68,7 +69,8 @@ class StreamStubMessagesSpec extends Specification {
applicationContext.getBean(BindingServiceProperties) >> properties
applicationContext.getBean(MessageCollector) >> collector
and:
StreamStubMessages messages = new StreamStubMessages(applicationContext)
DestinationResolver resolver = new DestinationResolver(applicationContext)
StreamFromBinderMappingMessageSender messages = new StreamFromBinderMappingMessageSender(applicationContext, resolver)
when:
messages.send("foo", [:], "verifications")
then:
@@ -89,7 +91,8 @@ class StreamStubMessagesSpec extends Specification {
applicationContext.getBean(BindingServiceProperties) >> properties
applicationContext.getBean(MessageCollector) >> collector
and:
StreamStubMessages messages = new StreamStubMessages(applicationContext)
DestinationResolver resolver = new DestinationResolver(applicationContext)
StreamStubMessages messages = new StreamStubMessages(new StreamFromBinderMappingMessageSender(applicationContext, resolver), new StreamMessageCollectorMessageReceiver(resolver, applicationContext))
when:
messageInteraction(messages)
then:
@@ -113,7 +116,8 @@ class StreamStubMessagesSpec extends Specification {
applicationContext.getBean(BindingServiceProperties) >> properties
applicationContext.getBean(MessageCollector) >> collector
and:
StreamStubMessages messages = new StreamStubMessages(applicationContext)
DestinationResolver resolver = new DestinationResolver(applicationContext)
StreamStubMessages messages = new StreamStubMessages(new StreamFromBinderMappingMessageSender(applicationContext, resolver), new StreamMessageCollectorMessageReceiver(resolver, applicationContext))
when:
messageInteraction(messages)
then:
@@ -138,7 +142,8 @@ class StreamStubMessagesSpec extends Specification {
applicationContext.getBean(BindingServiceProperties) >> properties
applicationContext.getBean(MessageCollector) >> collector
and:
StreamStubMessages messages = new StreamStubMessages(applicationContext)
DestinationResolver resolver = new DestinationResolver(applicationContext)
StreamStubMessages messages = new StreamStubMessages(new StreamFromBinderMappingMessageSender(applicationContext, resolver), new StreamMessageCollectorMessageReceiver(resolver, applicationContext))
when:
messages.receive("verificationsChannel")
then:
@@ -161,7 +166,8 @@ class StreamStubMessagesSpec extends Specification {
applicationContext.getBean(BindingServiceProperties) >> properties
applicationContext.getBean(MessageCollector) >> collector
and:
StreamStubMessages messages = new StreamStubMessages(applicationContext)
DestinationResolver resolver = new DestinationResolver(applicationContext)
StreamStubMessages messages = new StreamStubMessages(new StreamFromBinderMappingMessageSender(applicationContext, resolver), new StreamMessageCollectorMessageReceiver(resolver, applicationContext))
when:
messages.send("foo", [:], "verificationsChannel")
then:
@@ -182,7 +188,8 @@ class StreamStubMessagesSpec extends Specification {
applicationContext.getBean(BindingServiceProperties) >> properties
applicationContext.getBean(MessageCollector) >> collector
and:
StreamStubMessages messages = new StreamStubMessages(applicationContext)
DestinationResolver resolver = new DestinationResolver(applicationContext)
StreamStubMessages messages = new StreamStubMessages(new StreamFromBinderMappingMessageSender(applicationContext, resolver), new StreamMessageCollectorMessageReceiver(resolver, applicationContext))
when:
messageInteraction(messages)
then:
@@ -206,7 +213,8 @@ class StreamStubMessagesSpec extends Specification {
applicationContext.getBean(BindingServiceProperties) >> properties
applicationContext.getBean(MessageCollector) >> collector
and:
StreamStubMessages messages = new StreamStubMessages(applicationContext)
DestinationResolver resolver = new DestinationResolver(applicationContext)
StreamStubMessages messages = new StreamStubMessages(new StreamFromBinderMappingMessageSender(applicationContext, resolver), new StreamMessageCollectorMessageReceiver(resolver, applicationContext))
when:
messageInteraction(messages)
then: