Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-756: Add support for no-local consumers

Browse Source 
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`
This commit is contained in:
Johno Crawford
2017-08-02 09:24:24 +02:00
committed by Artem Bilan
parent 2ec3b48c34
commit 87ead85449
8 changed files with 107 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -12,6 +12,7 @@
.checkstyle
bin
build
out
.DS_Store
.springBeans
erl_crash.dump

10
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java
View File

@@ -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<AbstractMe
private Map<String, Object> consumerArgs;
private Boolean noLocal;
private Boolean exclusive;
private Boolean defaultRequeueRejected;
@@ -252,6 +255,10 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
this.consumerArgs = args;
}
public void setNoLocal(Boolean noLocal) {
this.noLocal = noLocal;
}
public void setExclusive(boolean exclusive) {
this.exclusive = exclusive;
}
@@ -442,6 +449,9 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
if (this.consumerArgs != null) {
container.setConsumerArguments(this.consumerArgs);
}
if (this.noLocal != null) {
container.setNoLocal(this.noLocal);
}
if (this.exclusive != null) {
container.setExclusive(this.exclusive);
}

19
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View File

@@ -91,6 +91,7 @@ import com.rabbitmq.client.ShutdownSignalException;
* @author James Carr
* @author Gary Russell
* @author Alex Panchenko
* @author Johno Crawford
*/
public abstract class AbstractMessageListenerContainer extends RabbitAccessor
implements MessageListenerContainer, ApplicationContextAware, BeanNameAware, DisposableBean,
@@ -189,6 +190,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
private volatile boolean exclusive;
private volatile boolean noLocal;
private volatile boolean defaultRequeueRejected = true;
private volatile int prefetchCount = DEFAULT_PREFETCH_COUNT;
@@ -666,6 +669,22 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
return this.exclusive;
}
/**
* Set to true for an no-local consumer.
* @param noLocal true for an no-local consumer.
*/
public void setNoLocal(boolean noLocal) {
this.noLocal = noLocal;
}
/**
* Return whether the consumers should be no-local.
* @return true for no-local consumers.
*/
protected boolean isNoLocal() {
return this.noLocal;
}
/**
* Set the default behavior when a message is rejected, for example because the listener
* threw an exception. When true, messages will be requeued, when false, they will not. For

34
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -77,6 +77,7 @@ import com.rabbitmq.utility.Utility;
* @author Casper Mout
* @author Artem Bilan
* @author Alex Panchenko
* @author Johno Crawford
*/
public class BlockingQueueConsumer {
@@ -115,6 +116,8 @@ public class BlockingQueueConsumer {
private final Map<String, Object> consumerArgs = new HashMap<String, Object>();
private final boolean noLocal;
private final boolean exclusive;
private final Set<Long> deliveryTags = new LinkedHashSet<Long>();
@@ -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<BlockingQueueConsumer> activeObjectCounter, AcknowledgeMode acknowledgeMode,
boolean transactional, int prefetchCount, boolean defaultRequeueRejected,
Map<String, Object> 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<BlockingQueueConsumer> activeObjectCounter, AcknowledgeMode acknowledgeMode,
boolean transactional, int prefetchCount, boolean defaultRequeueRejected,
Map<String, Object> consumerArgs, boolean exclusive, String... queues) {
Map<String, Object> 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<Delivery>(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);

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java
View File

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

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

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

45
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java
View File

@@ -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<BlockingQueueConsumer>(),
@@ -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<BlockingQueueConsumer>(),
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<BlockingQueueConsumer>(),

8
src/reference/asciidoc/amqp.adoc
View File

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