From 87ead8544910cc35d8cda5d69ff49edf53e483c3 Mon Sep 17 00:00:00 2001 From: Johno Crawford Date: Wed, 2 Aug 2017 09:24:24 +0200 Subject: [PATCH] AMQP-756: Add support for no-local consumers JIRA: https://jira.spring.io/browse/AMQP-756 * Fix Checkstyle * Add `@author` * Mention `noLocal` in the Doc for ListenerContainer * Add IDEA's `out` dir to the `.gitignore` --- .gitignore | 1 + .../config/ListenerContainerFactoryBean.java | 10 +++++ .../AbstractMessageListenerContainer.java | 19 ++++++++ .../listener/BlockingQueueConsumer.java | 34 +++++++++++++- .../DirectMessageListenerContainer.java | 2 +- .../SimpleMessageListenerContainer.java | 2 +- .../listener/BlockingQueueConsumerTests.java | 45 ++++++++++++++----- src/reference/asciidoc/amqp.adoc | 8 ++++ 8 files changed, 107 insertions(+), 14 deletions(-) diff --git a/.gitignore b/.gitignore index 659fa37a..598a88e4 100644 --- a/.gitignore +++ b/.gitignore @@ -12,6 +12,7 @@ .checkstyle bin build +out .DS_Store .springBeans erl_crash.dump diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java index 71e0bb9c..101beea6 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java @@ -52,6 +52,7 @@ import org.springframework.util.backoff.BackOff; * * @author Gary Russell * @author Artem Bilan + * @author Johno Crawford * * @since 2.0 * @@ -103,6 +104,8 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean consumerArgs; + private Boolean noLocal; + private Boolean exclusive; private Boolean defaultRequeueRejected; @@ -252,6 +255,10 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean consumerArgs = new HashMap(); + private final boolean noLocal; + private final boolean exclusive; private final Set deliveryTags = new LinkedHashSet(); @@ -225,11 +228,37 @@ public class BlockingQueueConsumer { * @param exclusive true if the consumer is to be exclusive. * @param queues The queues. */ + public BlockingQueueConsumer(ConnectionFactory connectionFactory, + MessagePropertiesConverter messagePropertiesConverter, + ActiveObjectCounter activeObjectCounter, AcknowledgeMode acknowledgeMode, + boolean transactional, int prefetchCount, boolean defaultRequeueRejected, + Map consumerArgs, boolean exclusive, String... queues) { + this(connectionFactory, messagePropertiesConverter, activeObjectCounter, acknowledgeMode, transactional, + prefetchCount, defaultRequeueRejected, consumerArgs, false, exclusive, queues); + } + + /** + * Create a consumer. The consumer must not attempt to use + * the connection factory or communicate with the broker + * until it is started. + * @param connectionFactory The connection factory. + * @param messagePropertiesConverter The properties converter. + * @param activeObjectCounter The active object counter; used during shutdown. + * @param acknowledgeMode The acknowledge mode. + * @param transactional Whether the channel is transactional. + * @param prefetchCount The prefetch count. + * @param defaultRequeueRejected true to reject requeued messages. + * @param consumerArgs The consumer arguments (e.g. x-priority). + * @param noLocal true if the consumer is to be no-local. + * @param exclusive true if the consumer is to be exclusive. + * @param queues The queues. + * @since 1.7.4 + */ public BlockingQueueConsumer(ConnectionFactory connectionFactory, MessagePropertiesConverter messagePropertiesConverter, ActiveObjectCounter activeObjectCounter, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, boolean defaultRequeueRejected, - Map consumerArgs, boolean exclusive, String... queues) { + Map consumerArgs, boolean noLocal, boolean exclusive, String... queues) { this.connectionFactory = connectionFactory; this.messagePropertiesConverter = messagePropertiesConverter; this.activeObjectCounter = activeObjectCounter; @@ -240,6 +269,7 @@ public class BlockingQueueConsumer { if (consumerArgs != null && consumerArgs.size() > 0) { this.consumerArgs.putAll(consumerArgs); } + this.noLocal = noLocal; this.exclusive = exclusive; this.queues = Arrays.copyOf(queues, queues.length); this.queue = new LinkedBlockingQueue(prefetchCount); @@ -591,7 +621,7 @@ public class BlockingQueueConsumer { private void consumeFromQueue(String queue) throws IOException { String consumerTag = this.channel.basicConsume(queue, this.acknowledgeMode.isAutoAck(), - (this.tagStrategy != null ? this.tagStrategy.createConsumerTag(queue) : ""), false, this.exclusive, + (this.tagStrategy != null ? this.tagStrategy.createConsumerTag(queue) : ""), this.noLocal, this.exclusive, this.consumerArgs, this.consumer); if (consumerTag != null) { this.consumerTags.put(consumerTag, queue); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index a70d3564..15cb941d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java @@ -573,7 +573,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta consumer.consumerTag = channel.basicConsume(queue, getAcknowledgeMode().isAutoAck(), (getConsumerTagStrategy() != null ? getConsumerTagStrategy().createConsumerTag(queue) : ""), - false, isExclusive(), getConsumerArguments(), consumer); + isNoLocal(), isExclusive(), getConsumerArguments(), consumer); } catch (AmqpApplicationContextClosedException e) { throw new AmqpConnectException(e); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 61ff61e7..dd6ca98a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java @@ -646,7 +646,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta int actualPrefetchCount = getPrefetchCount() > this.txSize ? getPrefetchCount() : this.txSize; consumer = new BlockingQueueConsumer(getConnectionFactory(), getMessagePropertiesConverter(), this.cancellationLock, getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount, - isDefaultRequeueRejected(), getConsumerArguments(), isExclusive(), queues); + isDefaultRequeueRejected(), getConsumerArguments(), isNoLocal(), isExclusive(), queues); if (this.declarationRetries != null) { consumer.setDeclarationRetries(this.declarationRetries); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java index f06235fb..c2e200af 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java @@ -23,6 +23,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.BDDMockito.willThrow; @@ -46,7 +47,6 @@ import org.apache.logging.log4j.Level; import org.junit.Rule; import org.junit.Test; import org.mockito.ArgumentCaptor; -import org.mockito.Mockito; import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.core.AcknowledgeMode; @@ -70,6 +70,8 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringChannel; /** * @author Gary Russell * @author Artem Bilan + * @author Johno Crawford + * * @since 1.0.1 * */ @@ -137,20 +139,20 @@ public class BlockingQueueConsumerTests { Channel channel = mock(Channel.class); when(connectionFactory.createConnection()).thenReturn(connection); - when(connection.createChannel(Mockito.anyBoolean())).thenReturn(channel); + when(connection.createChannel(anyBoolean())).thenReturn(channel); when(channel.isOpen()).thenReturn(true); - when(channel.queueDeclarePassive(Mockito.anyString())) + when(channel.queueDeclarePassive(anyString())) .then(invocation -> { String arg = invocation.getArgument(0); if ("good".equals(arg)) { - return Mockito.any(AMQP.Queue.DeclareOk.class); + return any(AMQP.Queue.DeclareOk.class); } else { throw new IOException(); } }); when(channel.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), - anyMap(), any(Consumer.class))).thenReturn("consumerTag"); + anyMap(), any(Consumer.class))).thenReturn("consumerTag"); BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory, new DefaultMessagePropertiesConverter(), new ActiveObjectCounter(), @@ -164,6 +166,29 @@ public class BlockingQueueConsumerTests { verify(channel).basicQos(20); } + @Test + public void testNoLocalConsumerConfiguration() throws Exception { + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + Connection connection = mock(Connection.class); + Channel channel = mock(Channel.class); + + when(connectionFactory.createConnection()).thenReturn(connection); + when(connection.createChannel(anyBoolean())).thenReturn(channel); + when(channel.isOpen()).thenReturn(true); + + final String queue = "testQ"; + final boolean noLocal = true; + + BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory, + new DefaultMessagePropertiesConverter(), new ActiveObjectCounter(), + AcknowledgeMode.AUTO, true, 1, true, null, noLocal, false, queue); + blockingQueueConsumer.start(); + verify(channel) + .basicConsume(eq(queue), eq(AcknowledgeMode.AUTO.isAutoAck()), eq(""), eq(noLocal), + eq(false), anyMap(), any(Consumer.class)); + blockingQueueConsumer.stop(); + } + @Test public void testRecoverAfterDeletedQueueAndLostConnection() throws Exception { ConnectionFactory connectionFactory = mock(ConnectionFactory.class); @@ -236,7 +261,7 @@ public class BlockingQueueConsumerTests { deliveryTags.add(1L); dfa.setPropertyValue("deliveryTags", deliveryTags); blockingQueueConsumer.rollbackOnExceptionIfNecessary(ex); - Mockito.verify(channel).basicNack(1L, true, expectedRequeue); + verify(channel).basicNack(1L, true, expectedRequeue); } @Test @@ -251,10 +276,10 @@ public class BlockingQueueConsumerTests { when(connection.createChannel(anyBoolean())).thenReturn(channel); final AtomicBoolean isOpen = new AtomicBoolean(true); doReturn(isOpen.get()).when(channel).isOpen(); - when(channel.queueDeclarePassive(Mockito.anyString())) + when(channel.queueDeclarePassive(anyString())) .then(invocation -> mock(AMQP.Queue.DeclareOk.class)); when(channel.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), - anyMap(), any(Consumer.class))).thenReturn("consumerTag"); + anyMap(), any(Consumer.class))).thenReturn("consumerTag"); BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory, new DefaultMessagePropertiesConverter(), new ActiveObjectCounter<>(), @@ -283,10 +308,10 @@ public class BlockingQueueConsumerTests { when(connection.createChannel(anyBoolean())).thenReturn(channel); final AtomicBoolean isOpen = new AtomicBoolean(true); doReturn(isOpen.get()).when(channel).isOpen(); - when(channel.queueDeclarePassive(Mockito.anyString())) + when(channel.queueDeclarePassive(anyString())) .then(invocation -> mock(AMQP.Queue.DeclareOk.class)); when(channel.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), - anyMap(), any(Consumer.class))).thenReturn("consumerTag"); + anyMap(), any(Consumer.class))).thenReturn("consumerTag"); BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory, new DefaultMessagePropertiesConverter(), new ActiveObjectCounter(), diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 82bb940e..242b1018 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -4650,6 +4650,14 @@ ManagerRollback a| image::images/tickmark.png[] a| image::images/tickmark.png[] +| noLocal +(N/A) + +| Set to `true` to disable delivery from the server to consumers messages published on the same channel's connection. + +a| image::images/tickmark.png[] +a| image::images/tickmark.png[] + |=== [[listener-concurrency]]