From f48895c1893703008e38b3a4e702cd5470c1d7fe Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Tue, 9 Jul 2019 17:46:51 +0200 Subject: [PATCH] Added back support for interacting with a running broker via stub runner --- .../verifier/messaging/MessageVerifier.java | 39 +----- .../messaging/MessageVerifierReceiver.java | 50 ++++++++ .../messaging/MessageVerifierSender.java | 49 ++++++++ ...ntractVerifierStreamAutoConfiguration.java | 42 +++++-- .../messaging/stream/DefaultChannels.java | 23 ++++ .../messaging/stream/DestinationResolver.java | 88 +++++++++++++ .../StreamFromBinderMappingMessageSender.java | 69 +++++++++++ ...StreamMessageCollectorMessageReceiver.java | 75 +++++++++++ .../StreamPollableChannelMessageReceiver.java | 64 ++++++++++ .../stream/StreamStubMessageSender.java | 62 ++++++++++ .../messaging/stream/StreamStubMessages.java | 116 ++---------------- .../stream/StreamStubMessagesSpec.groovy | 24 ++-- 12 files changed, 544 insertions(+), 157 deletions(-) create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierReceiver.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierSender.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/DefaultChannels.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/DestinationResolver.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamFromBinderMappingMessageSender.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamMessageCollectorMessageReceiver.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamPollableChannelMessageReceiver.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessageSender.java diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifier.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifier.java index fd41ca4107..4ef88e01ac 100644 --- a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifier.java +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifier.java @@ -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 { - - /** - * 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 payload type - * @param payload payload to send - * @param headers headers to send - * @param destination destination to which the message will be sent - */ - void send(T payload, Map 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 + extends MessageVerifierSender, MessageVerifierReceiver { } diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierReceiver.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierReceiver.java new file mode 100644 index 0000000000..7a3db801ca --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierReceiver.java @@ -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 message type + * @author Marcin Grzejszczak + * @since 2.2.0 + */ +public interface MessageVerifierReceiver { + + /** + * 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); + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierSender.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierSender.java new file mode 100644 index 0000000000..c6b86ff85a --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/MessageVerifierSender.java @@ -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 message type + * @author Marcin Grzejszczak + * @since 2.2.0 + */ +public interface MessageVerifierSender { + + /** + * 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 payload type + * @param payload payload to send + * @param headers headers to send + * @param destination destination to which the message will be sent + */ + void send(T payload, Map headers, String destination); + +} 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 e5c8d77224..0fb82c79ca 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 @@ -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> 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> 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> contractVerifierMessageExchangeWithNoMessageCollector( + ApplicationContext applicationContext, + BinderAwareChannelResolver resolver) { + return new StreamStubMessages(new StreamStubMessageSender(resolver), + new StreamPollableChannelMessageReceiver(applicationContext)); + } + + } + } class ContractVerifierHelper extends ContractVerifierMessaging> { diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/DefaultChannels.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/DefaultChannels.java new file mode 100644 index 0000000000..c00c13450a --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/DefaultChannels.java @@ -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 + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/DestinationResolver.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/DestinationResolver.java new file mode 100644 index 0000000000..b84b781693 --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/DestinationResolver.java @@ -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 channels = new HashMap<>(); + for (Map.Entry 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; + } + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamFromBinderMappingMessageSender.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamFromBinderMappingMessageSender.java new file mode 100644 index 0000000000..9aaf66f4fb --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamFromBinderMappingMessageSender.java @@ -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> { + + 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 void send(T payload, Map 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; + } + } + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamMessageCollectorMessageReceiver.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamMessageCollectorMessageReceiver.java new file mode 100644 index 0000000000..1ca1279396 --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamMessageCollectorMessageReceiver.java @@ -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> { + + 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); + } + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamPollableChannelMessageReceiver.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamPollableChannelMessageReceiver.java new file mode 100644 index 0000000000..e1305c9290 --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamPollableChannelMessageReceiver.java @@ -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> { + + 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); + } + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessageSender.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessageSender.java new file mode 100644 index 0000000000..ffea55d009 --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/stream/StreamStubMessageSender.java @@ -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> { + + 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 void send(T payload, Map 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; + } + } + +} 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 b68950b537..b5f1ebad7b 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,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> { - private static final Log log = LogFactory.getLog(StreamStubMessages.class); + private final MessageVerifierSender> sender; - private final ApplicationContext context; + private final MessageVerifierReceiver> receiver; - private MessageCollector messageCollector; - - private final ContractVerifierStreamMessageBuilder builder = new ContractVerifierStreamMessageBuilder(); - - public StreamStubMessages(ApplicationContext context) { - this.context = context; + public StreamStubMessages(MessageVerifierSender> sender, + MessageVerifierReceiver> receiver) { + this.sender = sender; + this.receiver = receiver; } @Override public void send(T payload, Map 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 channels = new HashMap<>(); - for (Map.Entry 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); } } 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 index 1723a2c07a..eb182d382a 100644 --- 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 @@ -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: