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

AMQP-727: RoutingConnectionFactory Fixes

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-727

- Bind the lookup key to any thread in the container that might open a connection
  so that the same connection factory will be used if a connection listener is invoked.
- Add a lookup key qualifier to the container so that the same queue name(s) can be used in
  different virtual hosts.

* Fix typos in JavaDocs according PR comments
This commit is contained in:
Gary Russell
2017-03-15 13:04:47 -04:00
committed by Artem Bilan
parent c294fe7c0d
commit 0f2c23d0bb
5 changed files with 197 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -197,6 +197,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
private boolean alwaysRequeueWithTxManagerRollback;
private String lookupKeyQualifier = "";
/**
* {@inheritDoc}
* @since 1.5
@@ -529,7 +531,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
ConnectionFactory connectionFactory = super.getConnectionFactory();
if (connectionFactory instanceof RoutingConnectionFactory) {
ConnectionFactory targetConnectionFactory = ((RoutingConnectionFactory) connectionFactory)
.getTargetConnectionFactory(this.queueNames.toString().replaceAll(" ", ""));
.getTargetConnectionFactory(getRoutingLookupKey());
if (targetConnectionFactory != null) {
return targetConnectionFactory;
}
@@ -537,6 +539,43 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
return connectionFactory;
}
/**
* Set a qualifier that will prefix the connection factory lookup key; default none.
* @param lookupKeyQualifier the qualifier
* @since 1.6.9
* @see #getRoutingLookupKey()
*/
public void setLookupKeyQualifier(String lookupKeyQualifier) {
this.lookupKeyQualifier = lookupKeyQualifier;
}
/**
* Return the lookup key if the connection factory is a
* {@link RoutingConnectionFactory}; null otherwise. The routing key is the
* comma-delimited list of queue names with all spaces removed and bracketed by [...],
* optionally prefixed by a qualifier, e.g. "foo[...]".
* @return the key or null.
* @since 1.6.9
* @see #setLookupKeyQualifier(String)
*/
protected String getRoutingLookupKey() {
return super.getConnectionFactory() instanceof RoutingConnectionFactory
? (this.lookupKeyQualifier + this.queueNames.toString().replaceAll(" ", ""))
: null;
}
/**
* Return the (@link RoutingConnectionFactory} if the connection factory is a
* {@link RoutingConnectionFactory}; null otherwise.
* @return the {@link RoutingConnectionFactory} or null.
* @since 1.6.9
*/
protected RoutingConnectionFactory getRoutingConnectionFactory() {
return super.getConnectionFactory() instanceof RoutingConnectionFactory
? (RoutingConnectionFactory) super.getConnectionFactory()
: null;
}
/**
* The 'id' attribute of the listener.
* @return the id (or the container bean name if no id set).

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

@@ -48,6 +48,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
import org.springframework.amqp.rabbit.connection.ConsumerChannelRegistry;
import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.scheduling.TaskScheduler;
@@ -352,7 +353,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
this.consumersToRestart.clear();
if (this.started) {
if (restartableConsumers.size() > 0) {
redeclareElementsIfNecessary();
doRedeclareElementsIfNecessary();
}
for (SimpleConsumer consumer : restartableConsumers) {
if (this.logger.isDebugEnabled() && restartableConsumers.size() > 0) {
@@ -377,12 +378,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
processMonitorTask();
}, this.monitorInterval);
if (queueNames.length > 0) {
try {
redeclareElementsIfNecessary();
}
catch (Exception e) {
this.logger.error("Failed to redeclare elements", e);
}
doRedeclareElementsIfNecessary();
getTaskExecutor().execute(() -> {
synchronized (this.consumersMonitor) {
@@ -437,6 +433,24 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
}
}
protected void doRedeclareElementsIfNecessary() {
String routingLookupKey = getRoutingLookupKey();
if (routingLookupKey != null) {
SimpleResourceHolder.bind(getRoutingConnectionFactory(), routingLookupKey);
}
try {
redeclareElementsIfNecessary();
}
catch (Exception e) {
this.logger.error("Failed to redeclare elements", e);
}
finally {
if (routingLookupKey != null) {
SimpleResourceHolder.unbind(getRoutingConnectionFactory());
}
}
}
/**
* Subclasses can override this to take additional actions when the monitor task runs.
*/
@@ -480,6 +494,10 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
}
return;
}
String routingLookupKey = getRoutingLookupKey();
if (routingLookupKey != null) {
SimpleResourceHolder.bind(getRoutingConnectionFactory(), routingLookupKey);
}
Connection connection = null; // NOSONAR (close)
try {
connection = getConnectionFactory().createConnection();
@@ -488,6 +506,11 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
this.consumersToRestart.add(new SimpleConsumer(null, null, queue));
throw new AmqpConnectException(e);
}
finally {
if (routingLookupKey != null) {
SimpleResourceHolder.unbind(getRoutingConnectionFactory());
}
}
Channel channel = null;
SimpleConsumer consumer = null;
try {

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

@@ -41,6 +41,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
import org.springframework.amqp.rabbit.connection.ConsumerChannelRegistry;
import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerExecutionException;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
@@ -862,6 +863,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
this.consumer.setLocallyTransacted(isChannelLocallyTransacted());
String routingLookupKey = getRoutingLookupKey();
if (routingLookupKey != null) {
SimpleResourceHolder.bind(getRoutingConnectionFactory(), routingLookupKey);
}
if (this.consumer.getQueueCount() < 1) {
if (logger.isDebugEnabled()) {
logger.debug("Consumer stopping; no queues for " + this.consumer);
@@ -1056,6 +1062,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
restart(this.consumer);
}
if (routingLookupKey != null) {
SimpleResourceHolder.unbind(getRoutingConnectionFactory());
}
}
private void logConsumerException(Throwable t) {

131
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,28 +16,43 @@
package org.springframework.amqp.rabbit.connection;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import com.rabbitmq.client.Channel;
/**
* @author Artem Bilan
* @author Josh Chappelle
* @author Gary Russell
* @since 1.3
*/
public class RoutingConnectionFactoryTests {
@@ -69,9 +84,9 @@ public class RoutingConnectionFactoryTests {
connectionFactory.createConnection();
}
Mockito.verify(connectionFactory1, Mockito.times(2)).createConnection();
Mockito.verify(connectionFactory2).createConnection();
Mockito.verify(defaultConnectionFactory, Mockito.times(2)).createConnection();
verify(connectionFactory1, times(2)).createConnection();
verify(connectionFactory2).createConnection();
verify(defaultConnectionFactory, times(2)).createConnection();
}
@Test
@@ -101,8 +116,8 @@ public class RoutingConnectionFactoryTests {
executorService.shutdown();
assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
Mockito.verify(connectionFactory1, Mockito.times(2)).createConnection();
Mockito.verify(connectionFactory2).createConnection();
verify(connectionFactory1, times(2)).createConnection();
verify(connectionFactory2).createConnection();
}
@Test
@@ -144,8 +159,8 @@ public class RoutingConnectionFactoryTests {
ConnectionFactory targetConnectionFactory = Mockito.mock(ConnectionFactory.class);
routingFactory.addTargetConnectionFactory("1", targetConnectionFactory);
Mockito.verify(targetConnectionFactory,
Mockito.times(2)).addConnectionListener(Mockito.any(ConnectionListener.class));
verify(targetConnectionFactory,
times(2)).addConnectionListener(any(ConnectionListener.class));
}
@Test
@@ -167,24 +182,102 @@ public class RoutingConnectionFactoryTests {
container.afterPropertiesSet();
container.start();
Mockito.verify(connectionFactory1, never()).createConnection();
Mockito.verify(connectionFactory2).createConnection();
Mockito.verify(defaultConnectionFactory, never()).createConnection();
verify(connectionFactory1, never()).createConnection();
verify(connectionFactory2).createConnection();
verify(defaultConnectionFactory, never()).createConnection();
Mockito.reset(connectionFactory1, connectionFactory2, defaultConnectionFactory);
reset(connectionFactory1, connectionFactory2, defaultConnectionFactory);
container.setQueueNames("baz");
Mockito.verify(connectionFactory1).createConnection();
Mockito.verify(connectionFactory2, never()).createConnection();
Mockito.verify(defaultConnectionFactory, never()).createConnection();
verify(connectionFactory1).createConnection();
verify(connectionFactory2, never()).createConnection();
verify(defaultConnectionFactory, never()).createConnection();
Mockito.reset(connectionFactory1, connectionFactory2, defaultConnectionFactory);
reset(connectionFactory1, connectionFactory2, defaultConnectionFactory);
container.setQueueNames("qux");
Mockito.verify(connectionFactory1, never()).createConnection();
Mockito.verify(connectionFactory2, never()).createConnection();
Mockito.verify(defaultConnectionFactory).createConnection();
verify(connectionFactory1, never()).createConnection();
verify(connectionFactory2, never()).createConnection();
verify(defaultConnectionFactory).createConnection();
container.stop();
}
@Test
public void testWithSMLCAndConnectionListener() throws Exception {
ConnectionFactory connectionFactory1 = mock(ConnectionFactory.class);
Map<Object, ConnectionFactory> factories = new HashMap<Object, ConnectionFactory>(2);
factories.put("xxx[foo]", connectionFactory1);
final SimpleRoutingConnectionFactory connectionFactory = new SimpleRoutingConnectionFactory();
final Connection connection = mock(Connection.class);
Channel channel = mock(Channel.class);
given(connection.createChannel(anyBoolean())).willReturn(channel);
final AtomicReference<Object> connectionMakerKey1 = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
willAnswer(i -> {
connectionMakerKey1.set(connectionFactory.determineCurrentLookupKey());
latch.countDown();
return connection;
}).given(connectionFactory1).createConnection();
connectionFactory.setTargetConnectionFactories(factories);
final AtomicReference<Object> connectionMakerKey2 = new AtomicReference<>();
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory) {
@Override
protected synchronized void redeclareElementsIfNecessary() {
connectionMakerKey2.set(connectionFactory.determineCurrentLookupKey());
}
};
container.setQueueNames("foo");
container.setLookupKeyQualifier("xxx");
container.afterPropertiesSet();
container.start();
assertTrue(latch.await(10, TimeUnit.SECONDS));
container.stop();
assertThat(connectionMakerKey1.get(), equalTo("xxx[foo]"));
assertThat(connectionMakerKey2.get(), equalTo("xxx[foo]"));
}
@Test
public void testWithDMLCAndConnectionListener() throws Exception {
ConnectionFactory connectionFactory1 = mock(ConnectionFactory.class);
Map<Object, ConnectionFactory> factories = new HashMap<Object, ConnectionFactory>(2);
factories.put("xxx[foo]", connectionFactory1);
final SimpleRoutingConnectionFactory connectionFactory = new SimpleRoutingConnectionFactory();
final Connection connection = mock(Connection.class);
Channel channel = mock(Channel.class);
given(connection.createChannel(anyBoolean())).willReturn(channel);
final AtomicReference<Object> connectionMakerKey = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
willAnswer(i -> {
connectionMakerKey.set(connectionFactory.determineCurrentLookupKey());
latch.countDown();
return connection;
}).given(connectionFactory1).createConnection();
connectionFactory.setTargetConnectionFactories(factories);
final AtomicReference<Object> connectionMakerKey2 = new AtomicReference<>();
DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory) {
@Override
protected synchronized void redeclareElementsIfNecessary() {
connectionMakerKey2.set(connectionFactory.determineCurrentLookupKey());
}
};
container.setQueueNames("foo");
container.setLookupKeyQualifier("xxx");
container.setShutdownTimeout(10);
container.afterPropertiesSet();
container.start();
assertTrue(latch.await(10, TimeUnit.SECONDS));
container.stop();
assertThat(connectionMakerKey.get(), equalTo("xxx[foo]"));
assertThat(connectionMakerKey2.get(), equalTo("xxx[foo]"));
}
}

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

@@ -466,10 +466,15 @@ But, if `lenientFallback = false`, an `IllegalStateException` is thrown.
The Namespace support also provides the `send-connection-factory-selector-expression` and `receive-connection-factory-selector-expression` attributes on the `<rabbit:template>` component.
Also starting with _version 1.4_, you can configure a routing connection factory in a `SimpleMessageListenerContainer`.
Also starting with _version 1.4_, you can configure a routing connection factory in a listener container.
In that case, the list of queue names is used as the lookup key.
For example, if you configure the container with `setQueueNames("foo", "bar")`, the lookup key will be `"[foo,bar]"` (no spaces).
Starting with _version 1.6.9_ you can add a qualifier to the lookup key using `setLookupKeyQualifier` on the listener container.
This would enable, for example, listening to queues with the same name, but in different virtual host (where you would have a connection factory for each).
For example, with lookup key qualifier `foo` and a container listening to queue `bar`, the lookup key you would register the target connection factory with would be `foo[bar]`.
[[queue-affinity]]
===== Queue Affinity and the LocalizedQueueConnectionFactory