diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/ContractVerifierAmqpAutoConfiguration.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/ContractVerifierAmqpAutoConfiguration.java index a5d1a84322..6d43f4b8b9 100644 --- a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/ContractVerifierAmqpAutoConfiguration.java +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/ContractVerifierAmqpAutoConfiguration.java @@ -29,6 +29,7 @@ import org.springframework.amqp.support.converter.MessagingMessageConverter; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.AutoConfigureAfter; import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -69,12 +70,16 @@ public class ContractVerifierAmqpAutoConfiguration { @Autowired(required = false) private List bindings = emptyList(); + @Autowired + private RabbitProperties rabbitProperties; + @Bean @ConditionalOnMissingBean public MessageVerifier contractVerifierMessageExchange() { return new SpringAmqpStubMessages(this.rabbitTemplate, new MessageListenerAccessor(this.rabbitListenerEndpointRegistry, - this.simpleMessageListenerContainers, this.bindings)); + this.simpleMessageListenerContainers, this.bindings), + this.rabbitProperties); } @Bean diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java index 4f2de824d6..d5517393f1 100644 --- a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java @@ -20,6 +20,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import com.rabbitmq.client.Channel; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.mockito.ArgumentCaptor; @@ -33,6 +34,7 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.cloud.contract.verifier.messaging.MessageVerifier; import org.springframework.util.Assert; @@ -64,7 +66,9 @@ public class SpringAmqpStubMessages implements MessageVerifier { private final MessageListenerAccessor messageListenerAccessor; - @Autowired + private RabbitProperties rabbitProperties; + + @Deprecated public SpringAmqpStubMessages(RabbitTemplate rabbitTemplate, MessageListenerAccessor messageListenerAccessor) { Assert.notNull(rabbitTemplate, "RabbitTemplate must be set"); @@ -83,6 +87,27 @@ public class SpringAmqpStubMessages implements MessageVerifier { this.messageListenerAccessor = messageListenerAccessor; } + @Autowired + public SpringAmqpStubMessages(RabbitTemplate rabbitTemplate, + MessageListenerAccessor messageListenerAccessor, + RabbitProperties rabbitProperties) { + Assert.notNull(rabbitTemplate, "RabbitTemplate must be set"); + Assert.isTrue( + mockingDetails(rabbitTemplate).isSpy() + || mockingDetails(rabbitTemplate).isMock(), + "StubRunner AMQP will work only if RabbiTemplate is a spy"); // we get + // send + // messages + // by + // capturing + // arguments + // on the + // spy + this.rabbitTemplate = rabbitTemplate; + this.messageListenerAccessor = messageListenerAccessor; + this.rabbitProperties = rabbitProperties; + } + @Override public void send(T payload, Map headers, String destination) { Message message = org.springframework.amqp.core.MessageBuilder @@ -128,12 +153,10 @@ public class SpringAmqpStubMessages implements MessageVerifier { } for (SimpleMessageListenerContainer listenerContainer : listenerContainers) { Object messageListener = listenerContainer.getMessageListener(); - if (messageListener instanceof ChannelAwareMessageListener - && listenerContainer.getConnectionFactory() != null) { + if (isChannelAwareListener(listenerContainer, messageListener)) { try { ((ChannelAwareMessageListener) messageListener).onMessage(message, - listenerContainer.getConnectionFactory().createConnection() - .createChannel(true)); + createChannel(listenerContainer, transactionalChannel())); } catch (Exception e) { throw new RuntimeException(e); @@ -145,6 +168,26 @@ public class SpringAmqpStubMessages implements MessageVerifier { } } + Channel createChannel(SimpleMessageListenerContainer listenerContainer, + boolean transactional) { + return listenerContainer.getConnectionFactory().createConnection() + .createChannel(transactional); + } + + boolean isChannelAwareListener(SimpleMessageListenerContainer listenerContainer, + Object messageListener) { + return messageListener instanceof ChannelAwareMessageListener + && listenerContainer.getConnectionFactory() != null; + } + + private boolean transactionalChannel() { + if (this.rabbitProperties == null) { + // backward compatibility + return true; + } + return !this.rabbitProperties.isPublisherConfirms(); + } + @Override public Message receive(String destination, long timeout, TimeUnit timeUnit) { ArgumentCaptor messageCaptor = ArgumentCaptor.forClass(Message.class); diff --git a/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessagesSpec.groovy b/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessagesSpec.groovy index 14b81526bc..eba5bfc6de 100644 --- a/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessagesSpec.groovy +++ b/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessagesSpec.groovy @@ -16,6 +16,7 @@ package org.springframework.cloud.contract.verifier.messaging.amqp +import com.rabbitmq.client.Channel import org.mockito.exceptions.verification.WantedButNotInvoked import spock.lang.Specification import wiremock.com.google.common.collect.ImmutableMap @@ -29,6 +30,7 @@ import org.springframework.amqp.core.Queue import org.springframework.amqp.rabbit.core.RabbitTemplate import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter +import org.springframework.boot.autoconfigure.amqp.RabbitProperties import static org.mockito.Mockito.mock import static org.springframework.amqp.core.MessageProperties.CONTENT_TYPE_JSON @@ -42,6 +44,7 @@ class SpringAmqpStubMessagesSpec extends Specification { RabbitTemplate rabbitTemplate = mock(RabbitTemplate.class) SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer() MessageListenerAdapter messageListenerAdapter = Mock(MessageListenerAdapter.class) + RabbitProperties rabbitProperties = new RabbitProperties() Message message = Mock(Message.class) String queueName = "test.queue" @@ -55,7 +58,7 @@ class SpringAmqpStubMessagesSpec extends Specification { listenerContainer.setQueueNames(queueName) Binding binding = BindingBuilder.bind(new Queue(queueName)).to(new DirectExchange(exchange)).with(routingKey) MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [listenerContainer], [binding]) - SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor) + SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor, rabbitProperties) when: messageVerifier.send(payload, @@ -73,13 +76,85 @@ class SpringAmqpStubMessagesSpec extends Specification { }) } + def "should send amqp message for non transactional channel"() { + given: + rabbitProperties.setPublisherConfirms(true) + listenerContainer.setMessageListener(messageListenerAdapter) + listenerContainer.setQueueNames(queueName) + Binding binding = BindingBuilder.bind(new Queue(queueName)).to(new DirectExchange(exchange)).with(routingKey) + MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [listenerContainer], [binding]) + boolean createChannelCalled = false + boolean transactionalChannel = false + SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor, rabbitProperties) { + @Override + boolean isChannelAwareListener(SimpleMessageListenerContainer listenerContainer, Object messageListener) { + return true + } + + @Override + Channel createChannel(SimpleMessageListenerContainer listenerContainer, boolean transactional) { + createChannelCalled = true + transactionalChannel = transactional + return null + } + } + + when: + messageVerifier.send(payload, + ImmutableMap.builder() + .put(DEFAULT_CLASSID_FIELD_NAME, "org.example.Some") + .put("amqp_receivedRoutingKey", routingKey) + .put("contentType", CONTENT_TYPE_JSON) + .build(), + exchange) + then: + createChannelCalled + !transactionalChannel + } + + def "should send amqp message for transactional channel"() { + given: + rabbitProperties.setPublisherConfirms(false) + listenerContainer.setMessageListener(messageListenerAdapter) + listenerContainer.setQueueNames(queueName) + Binding binding = BindingBuilder.bind(new Queue(queueName)).to(new DirectExchange(exchange)).with(routingKey) + MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [listenerContainer], [binding]) + boolean createChannelCalled = false + boolean transactionalChannel = false + SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor, rabbitProperties) { + @Override + boolean isChannelAwareListener(SimpleMessageListenerContainer listenerContainer, Object messageListener) { + return true + } + + @Override + Channel createChannel(SimpleMessageListenerContainer listenerContainer, boolean transactional) { + createChannelCalled = true + transactionalChannel = transactional + return null + } + } + + when: + messageVerifier.send(payload, + ImmutableMap.builder() + .put(DEFAULT_CLASSID_FIELD_NAME, "org.example.Some") + .put("amqp_receivedRoutingKey", routingKey) + .put("contentType", CONTENT_TYPE_JSON) + .build(), + exchange) + then: + createChannelCalled + transactionalChannel + } + def "should fail to receive a message if rabbit template wasn't called"() { given: listenerContainer.setMessageListener(messageListenerAdapter) listenerContainer.setQueueNames(queueName) Binding binding = BindingBuilder.bind(new Queue(queueName)).to(new DirectExchange(exchange)).with(routingKey) MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [listenerContainer], [binding]) - SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor) + SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor, rabbitProperties) when: messageVerifier.receive("foo") then: @@ -92,7 +167,7 @@ class SpringAmqpStubMessagesSpec extends Specification { listenerContainer.setQueueNames(queueName) Binding binding = BindingBuilder.bind(new Queue(queueName)).to(new DirectExchange(exchange)).with(routingKey) MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [listenerContainer], [binding]) - SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor) + SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor, rabbitProperties) and: rabbitTemplate.send("foo", "bar", null, null) @@ -106,7 +181,7 @@ class SpringAmqpStubMessagesSpec extends Specification { listenerContainer.setQueueNames(queueName) Binding binding = BindingBuilder.bind(new Queue(queueName)).to(new DirectExchange(exchange)).with(routingKey) MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [listenerContainer], [binding]) - SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor) + SpringAmqpStubMessages messageVerifier = new SpringAmqpStubMessages(rabbitTemplate, messageListenerAccessor, rabbitProperties) message.getMessageProperties() >> new MessageProperties() and: rabbitTemplate.send("foo", "bar", message, null)