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

Ensures that non-transactional channel is an option for AmqpStubMessages; fixes gh-859

Browse Source
This commit is contained in:
Marcin Grzejszczak
2019-02-21 15:46:11 +01:00
parent dc9a6589e1
commit dfb3202dd1
3 changed files with 133 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/ContractVerifierAmqpAutoConfiguration.java
View File

@@ -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<Binding> bindings = emptyList();
@Autowired
private RabbitProperties rabbitProperties;
@Bean
@ConditionalOnMissingBean
public MessageVerifier<Message> contractVerifierMessageExchange() {
return new SpringAmqpStubMessages(this.rabbitTemplate,
new MessageListenerAccessor(this.rabbitListenerEndpointRegistry,
this.simpleMessageListenerContainers, this.bindings));
this.simpleMessageListenerContainers, this.bindings),
this.rabbitProperties);
}
@Bean

53
spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java
View File

@@ -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<Message> {
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<Message> {
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 <T> void send(T payload, Map<String, Object> headers, String destination) {
Message message = org.springframework.amqp.core.MessageBuilder
@@ -128,12 +153,10 @@ public class SpringAmqpStubMessages implements MessageVerifier<Message> {
}
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<Message> {
}
}
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<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);

83
spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessagesSpec.groovy
View File

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