From c7250a8d6291e36bb6c05738f17cb2a1ce5ac0db Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Tue, 20 Mar 2018 14:17:05 +0100 Subject: [PATCH] Added support for `RoutingKey` in AMQP stub messages fixed via @KazuoD suggestion fixes gh-585 --- .../amqp/MessageListenerAccessor.java | 10 +++--- .../amqp/SpringAmqpStubMessages.java | 22 ++++++++++--- .../amqp/MessageListenerAccessorSpec.groovy | 31 ++++++++++++++++--- .../amqp/SpringAmqpStubMessagesSpec.groovy | 12 ++++--- 4 files changed, 57 insertions(+), 18 deletions(-) diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/MessageListenerAccessor.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/MessageListenerAccessor.java index bfe38df2ff..181161a0b3 100644 --- a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/MessageListenerAccessor.java +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/MessageListenerAccessor.java @@ -34,10 +34,10 @@ class MessageListenerAccessor { this.bindings = bindings; } - List getListenerContainersForDestination(String destination) { + List getListenerContainersForDestination(String destination, String routingKey) { List listenerContainers = collectListenerContainers(); //we interpret the destination as exchange name and collect all the queues bound to this exchange - Set queueNames = collectQueuesBoundToDestination(destination); + Set queueNames = collectQueuesBoundToDestination(destination, routingKey); return getListenersByBoundQueues(listenerContainers, queueNames); } @@ -56,10 +56,12 @@ class MessageListenerAccessor { return matchingContainers; } - private Set collectQueuesBoundToDestination(String destination) { + private Set collectQueuesBoundToDestination(String destination, String routingKey) { Set 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()); } } 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 32a0599cee..6d4a271abd 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 @@ -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; @@ -82,12 +82,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 listenerContainers = this.messageListenerAccessor.getListenerContainersForDestination(destination); + final String routingKey = message.getMessageProperties().getReceivedRoutingKey(); + List listenerContainers = this.messageListenerAccessor.getListenerContainersForDestination(destination, routingKey); if (listenerContainers.isEmpty()) { throw new IllegalStateException("no listeners found for destination " + destination); } @@ -100,8 +105,9 @@ public class SpringAmqpStubMessages implements @Override public Message receive(String destination, long timeout, TimeUnit timeUnit) { ArgumentCaptor messageCaptor = ArgumentCaptor.forClass(Message.class); - verify(this.rabbitTemplate, atLeastOnce()).send(eq(destination), anyString(), messageCaptor.capture(), any(CorrelationData.class)); - + ArgumentCaptor 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; @@ -109,7 +115,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 diff --git a/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/MessageListenerAccessorSpec.groovy b/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/MessageListenerAccessorSpec.groovy index 2f409057ea..da594aca34 100644 --- a/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/MessageListenerAccessorSpec.groovy +++ b/spring-cloud-contract-verifier/src/test/groovy/org/springframework/cloud/contract/verifier/messaging/amqp/MessageListenerAccessorSpec.groovy @@ -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 listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange) + List 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 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 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 listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination("some-exchange") + List 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 listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange) + List 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 listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange) + List listenerContainersForDestination = messageListenerAccessor.getListenerContainersForDestination(this.exchange, null) then: listenerContainersForDestination.size() == 1 listenerContainersForDestination.get(0) == this.listenerContainer 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 66a47a0892..e15205d30b 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 @@ -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" }) } }