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

Merge branch '1.2.x'

Browse Source
This commit is contained in:
Marcin Grzejszczak
2018-03-20 14:17:20 +01:00
parent 5476e1d127 c7250a8d62
commit b70106e7bc
4 changed files with 57 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -34,10 +34,10 @@ class MessageListenerAccessor {
this.bindings = bindings;
}
List<SimpleMessageListenerContainer> getListenerContainersForDestination(String destination) {
List<SimpleMessageListenerContainer> getListenerContainersForDestination(String destination, String routingKey) {
List<SimpleMessageListenerContainer> listenerContainers = collectListenerContainers();
//we interpret the destination as exchange name and collect all the queues bound to this exchange
Set<String> queueNames = collectQueuesBoundToDestination(destination);
Set<String> queueNames = collectQueuesBoundToDestination(destination, routingKey);
return getListenersByBoundQueues(listenerContainers, queueNames);
}
@@ -56,10 +56,12 @@ class MessageListenerAccessor {
return matchingContainers;
}
private Set<String> collectQueuesBoundToDestination(String destination) {
private Set<String> collectQueuesBoundToDestination(String destination, String routingKey) {
Set<String> queueNames = new HashSet<>();
for (Binding binding: this.bindings) {
if (destination.equals(binding.getExchange()) && DestinationType.QUEUE.equals(binding.getDestinationType())) {
if (destination.equals(binding.getExchange())
&& (routingKey == null || routingKey.equals(binding.getRoutingKey()))
&& DestinationType.QUEUE.equals(binding.getDestinationType())) {
queueNames.add(binding.getDestination());
}
}

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

@@ -29,12 +29,12 @@ import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.contract.verifier.messaging.MessageVerifier;
import org.springframework.util.Assert;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mockingDetails;
@@ -83,12 +83,17 @@ public class SpringAmqpStubMessages implements
if (headers != null && headers.containsKey(DEFAULT_CLASSID_FIELD_NAME)) {
message.getMessageProperties().setHeader(DEFAULT_CLASSID_FIELD_NAME, headers.get(DEFAULT_CLASSID_FIELD_NAME));
}
if (headers != null && headers.containsKey(AmqpHeaders.RECEIVED_ROUTING_KEY)) {
message.getMessageProperties()
.setReceivedRoutingKey((String) headers.get(AmqpHeaders.RECEIVED_ROUTING_KEY));
}
send(message, destination);
}
@Override
public void send(Message message, String destination) {
List<SimpleMessageListenerContainer> listenerContainers = this.messageListenerAccessor.getListenerContainersForDestination(destination);
final String routingKey = message.getMessageProperties().getReceivedRoutingKey();
List<SimpleMessageListenerContainer> listenerContainers = this.messageListenerAccessor.getListenerContainersForDestination(destination, routingKey);
if (listenerContainers.isEmpty()) {
throw new IllegalStateException("no listeners found for destination " + destination);
}
@@ -101,8 +106,9 @@ public class SpringAmqpStubMessages implements
@Override
public Message receive(String destination, long timeout, TimeUnit timeUnit) {
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
verify(this.rabbitTemplate, atLeastOnce()).send(eq(destination), anyString(), messageCaptor.capture(), any(CorrelationData.class));
ArgumentCaptor<String> routingKeyCaptor = ArgumentCaptor.forClass(String.class);
verify(this.rabbitTemplate, atLeastOnce()).send(eq(destination), routingKeyCaptor.capture(),
messageCaptor.capture(), any(CorrelationData.class));
if (messageCaptor.getAllValues().isEmpty()) {
log.info("no messages found on destination {}", destination);
return null;
@@ -110,7 +116,13 @@ public class SpringAmqpStubMessages implements
log.info("multiple messages found on destination {} returning last one - {}", destination);
return messageCaptor.getValue();
}
return messageCaptor.getValue();
Message message = messageCaptor.getValue();
if (!routingKeyCaptor.getValue().isEmpty()) {
log.info("routing key passed {}", routingKeyCaptor.getValue());
message.getMessageProperties()
.setReceivedRoutingKey(routingKeyCaptor.getValue());
}
return message;
}
@Override

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

@@ -18,20 +18,41 @@ class MessageListenerAccessorSpec extends Specification {
def "should get single simple listener container"(){
given:
givenSimpleMessageListenerContainer()
MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [this.listenerContainer], [this.binding])
MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [this.listenerContainer], [this.binding])
when:
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange)
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange, null)
then:
listenerContainersForDestination.size() == 1
listenerContainersForDestination.get(0) == this.listenerContainer
}
def "should get single simple listener container for matching routing key"(){
given:
givenSimpleMessageListenerContainer()
MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [this.listenerContainer], [this.binding])
when:
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange, '#')
then:
listenerContainersForDestination.size() == 1
listenerContainersForDestination.get(0) == this.listenerContainer
}
def "should get empty simple listener container for non matching routing key"(){
given:
givenSimpleMessageListenerContainer()
MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [this.listenerContainer], [this.binding])
when:
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange, 'not matching')
then:
listenerContainersForDestination.isEmpty()
}
def "should get empty listener container list for unknown destination"(){
given:
givenSimpleMessageListenerContainer()
MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [this.listenerContainer], [this.binding])
when:
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination("some-exchange")
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination("some-exchange", null)
then:
listenerContainersForDestination.isEmpty()
}
@@ -42,7 +63,7 @@ class MessageListenerAccessorSpec extends Specification {
this.binding = BindingBuilder.bind(new Queue("some.queue")).to(new DirectExchange(this.exchange)).with("#")
MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(null, [this.listenerContainer], [this.binding])
when:
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange)
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange, null)
then:
listenerContainersForDestination.isEmpty()
}
@@ -54,7 +75,7 @@ class MessageListenerAccessorSpec extends Specification {
rabbitListenerEndpointRegistryMock.getListenerContainers() >> [this.listenerContainer]
MessageListenerAccessor messageListenerAccessor = new MessageListenerAccessor(rabbitListenerEndpointRegistryMock, [], [this.binding])
when:
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange)
List<SimpleMessageListenerContainer> listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange, null)
then:
listenerContainersForDestination.size() == 1
listenerContainersForDestination.get(0) == this.listenerContainer

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

@@ -4,6 +4,7 @@ import wiremock.com.google.common.collect.ImmutableMap
import org.springframework.amqp.core.Binding
import org.springframework.amqp.core.BindingBuilder
import org.springframework.amqp.core.DirectExchange
import org.springframework.amqp.core.Message
import org.springframework.amqp.core.Queue
import org.springframework.amqp.rabbit.core.RabbitTemplate
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer
@@ -25,12 +26,13 @@ class SpringAmqpStubMessagesSpec extends Specification {
String queueName = "test.queue"
String exchange = "test-exchange"
String payload = '''{"name":"some"}'''
String routingKey = "resource.created"
def "should send amqp message with type id"() {
given:
listenerContainer.setMessageListener(messageListenerAdapter)
listenerContainer.setQueueNames(queueName)
Binding binding = BindingBuilder.bind(new Queue(queueName)).to(new DirectExchange(exchange)).with("#")
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)
@@ -38,13 +40,15 @@ class SpringAmqpStubMessagesSpec extends Specification {
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:
1 * messageListenerAdapter.onMessage({
it.getMessageProperties().getContentType() == CONTENT_TYPE_JSON &&
it.getMessageProperties().getHeaders().get(DEFAULT_CLASSID_FIELD_NAME) == "org.example.Some"
1 * messageListenerAdapter.onMessage({ Message msg ->
msg.getMessageProperties().getReceivedRoutingKey() == "resource.created" &&
msg.getMessageProperties().getContentType() == CONTENT_TYPE_JSON &&
msg.getMessageProperties().getHeaders().get(DEFAULT_CLASSID_FIELD_NAME) == "org.example.Some"
})
}
}