diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 86d7515b..004b5fd9 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -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). 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 93cfab25..c283c240 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 @@ -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 { 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 fca2ebc4..dd8b6121 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 @@ -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) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java index bd828558..9a0df5b6 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java @@ -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 factories = new HashMap(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 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 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 factories = new HashMap(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 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 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]")); + } } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 277a40d1..ca02fcbe 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -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 `` 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