From fa8cf62440f9e3406dcd887fc2ff3e426956e7a7 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 23 Apr 2015 09:59:31 +0100 Subject: [PATCH] AMQP-493: Add Consumer Tag Naming Strategy JIRA: https://jira.spring.io/browse/AMQP-493 Allow the user to generate custom consumer tags instead of the default server-generated tag. Remove `ServerGeneratesConsumerTagStrategy` and replace it with ternary operator: `(this.tagStrategy != null ? this.tagStrategy.createConsumerTag(queue) : "")` --- .../amqp/support/ConsumerTagStrategy.java | 36 +++++++++++++++++++ .../SimpleRabbitListenerContainerFactory.java | 17 ++++++++- .../listener/BlockingQueueConsumer.java | 15 +++++++- .../SimpleMessageListenerContainer.java | 16 +++++++++ .../EnableRabbitIntegrationTests.java | 36 ++++++++++++++++--- ...BlockingQueueConsumerIntegrationTests.java | 31 +++++++++++++--- 6 files changed, 140 insertions(+), 11 deletions(-) create mode 100644 spring-amqp/src/main/java/org/springframework/amqp/support/ConsumerTagStrategy.java diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/ConsumerTagStrategy.java b/spring-amqp/src/main/java/org/springframework/amqp/support/ConsumerTagStrategy.java new file mode 100644 index 00000000..24c8ebbd --- /dev/null +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/ConsumerTagStrategy.java @@ -0,0 +1,36 @@ +/* + * Copyright 2015 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.amqp.support; + +/** + * A strategy interface to determine the consumer tag to be used when issuing a + * {@code basicConsume} operation. + * + * @author Gary Russell + * @since 1.4.5 + * + */ +public interface ConsumerTagStrategy { + + /** + * Create the consumer tag, optionally based on the queue name that the consumer + * will listen to. Consumer tags must be unique. + * @param queue The queue name that this consumer will listen to. + * @return The consumer tag. + */ + String createConsumerTag(String queue); + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java index 74190c59..24aa32b7 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -22,6 +22,7 @@ import org.aopalliance.aop.Advice; import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.support.ConsumerTagStrategy; import org.springframework.transaction.PlatformTransactionManager; /** @@ -32,6 +33,7 @@ import org.springframework.transaction.PlatformTransactionManager; * for those that are used to build such container definition manually. * * @author Stephane Nicoll + * @author Gary Russell * @since 1.4 */ public class SimpleRabbitListenerContainerFactory @@ -67,6 +69,8 @@ public class SimpleRabbitListenerContainerFactory private Boolean missingQueuesFatal; + private ConsumerTagStrategy consumerTagStrategy; + /** * @param taskExecutor the {@link Executor} to use. * @see SimpleMessageListenerContainer#setTaskExecutor @@ -187,6 +191,14 @@ public class SimpleRabbitListenerContainerFactory this.missingQueuesFatal = missingQueuesFatal; } + /** + * @param consumerTagStrategy the consumerTagStrategy to set + * @see SimpleMessageListenerContainer#setConsumerTagStrategy(ConsumerTagStrategy) + */ + public void setConsumerTagStrategy(ConsumerTagStrategy consumerTagStrategy) { + this.consumerTagStrategy = consumerTagStrategy; + } + @Override protected SimpleMessageListenerContainer createContainerInstance() { return new SimpleMessageListenerContainer(); @@ -241,6 +253,9 @@ public class SimpleRabbitListenerContainerFactory if (this.missingQueuesFatal != null) { instance.setMissingQueuesFatal(this.missingQueuesFatal); } + if (this.consumerTagStrategy != null) { + instance.setConsumerTagStrategy(this.consumerTagStrategy); + } } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 968c7dec..55fb7ce1 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -48,6 +48,7 @@ import org.springframework.amqp.rabbit.listener.exception.ConsumerCancelledExcep import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; +import org.springframework.amqp.support.ConsumerTagStrategy; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.AMQP.BasicProperties; @@ -121,6 +122,8 @@ public class BlockingQueueConsumer { private long lastRetryDeclaration; + private ConsumerTagStrategy tagStrategy; + /** * Create a consumer. The consumer must not attempt to use * the connection factory or communicate with the broker @@ -267,6 +270,15 @@ public class BlockingQueueConsumer { this.retryDeclarationInterval = retryDeclarationInterval; } + /** + * Set the {@link ConsumerTagStrategy} to use when generating consumer tags. + * @param tagStrategy the tagStrategy to set + * @since 1.4.5 + */ + public void setTagStrategy(ConsumerTagStrategy tagStrategy) { + this.tagStrategy = tagStrategy; + } + protected void basicCancel() { for (String consumerTag : this.consumerTags.keySet()) { try { @@ -494,7 +506,8 @@ public class BlockingQueueConsumer { } private void consumeFromQueue(String queue) throws IOException { - String consumerTag = this.channel.basicConsume(queue, this.acknowledgeMode.isAutoAck(), "", false, this.exclusive, + String consumerTag = this.channel.basicConsume(queue, this.acknowledgeMode.isAutoAck(), + (this.tagStrategy != null ? this.tagStrategy.createConsumerTag(queue) : ""), false, 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/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 3e7c1693..57233630 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 @@ -51,6 +51,7 @@ import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupEx import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; +import org.springframework.amqp.support.ConsumerTagStrategy; import org.springframework.aop.Pointcut; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.DefaultPointcutAdvisor; @@ -157,6 +158,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private volatile boolean autoDeclare = true; + private volatile ConsumerTagStrategy consumerTagStrategy; + public interface ContainerDelegate { void invokeListener(Channel channel, Message message) throws Exception; } @@ -582,6 +585,16 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta this.retryDeclarationInterval = retryDeclarationInterval; } + /** + * Set the implementation of {@link ConsumerTagStrategy} to generate consumer tags. + * By default, the RabbitMQ server generates consumer tags. + * @param consumerTagStrategy the consumerTagStrategy to set. + * @since 1.4.5 + */ + public void setConsumerTagStrategy(ConsumerTagStrategy consumerTagStrategy) { + this.consumerTagStrategy = consumerTagStrategy; + } + /** * Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent * consumers. @@ -897,6 +910,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta if (this.retryDeclarationInterval != null) { consumer.setRetryDeclarationInterval(this.retryDeclarationInterval); } + if (this.consumerTagStrategy != null) { + consumer.setTagStrategy(this.consumerTagStrategy); + } return consumer; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index 69b2795f..4ea653fb 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -24,10 +24,12 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.util.Date; +import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import org.hamcrest.Matchers; import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; @@ -46,6 +48,8 @@ import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.test.BrokerRunning; import org.springframework.amqp.rabbit.test.MessageTestUtils; +import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.amqp.support.ConsumerTagStrategy; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -62,6 +66,7 @@ import org.springframework.util.ErrorHandler; * * @author Stephane Nicoll * @author Artem Bilan + * @author Gary Russell * @since 1.4 */ @ContextConfiguration(classes = EnableRabbitIntegrationTests.EnableRabbitConfig.class) @@ -83,6 +88,9 @@ public class EnableRabbitIntegrationTests { @Autowired private AtomicReference errorHandlerError; + @Autowired + private String tagPrefix; + @Test public void simpleEndpoint() { assertEquals("FOO", rabbitTemplate.convertSendAndReceive("test.simple", "foo")); @@ -114,7 +122,7 @@ public class EnableRabbitIntegrationTests { Message reply = rabbitTemplate.sendAndReceive("test.reply", request); assertEquals("Wrong reply", "content", MessageTestUtils.extractText(reply)); assertEquals("Wrong foo header", "fooValue", reply.getMessageProperties().getHeaders().get("foo")); - assertEquals("Wrong bar header", "barValue", reply.getMessageProperties().getHeaders().get("bar")); + assertThat((String) reply.getMessageProperties().getHeaders().get("bar"), Matchers.startsWith(tagPrefix)); } @Test @@ -165,9 +173,10 @@ public class EnableRabbitIntegrationTests { } @RabbitListener(queues = "test.reply") - public org.springframework.messaging.Message reply(String payload, @Header String foo) { + public org.springframework.messaging.Message reply(String payload, @Header String foo, + @Header(AmqpHeaders.CONSUMER_TAG) String tag) { return MessageBuilder.withPayload(payload) - .setHeader("foo", foo).setHeader("bar", "barValue").build(); + .setHeader("foo", foo).setHeader("bar", tag).build(); } @RabbitListener(queues = "test.sendTo") @@ -187,14 +196,33 @@ public class EnableRabbitIntegrationTests { @EnableRabbit public static class EnableRabbitConfig { + private int increment; + @Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(rabbitConnectionFactory()); factory.setErrorHandler(errorHandler()); + factory.setConsumerTagStrategy(consumerTagStrategy()); return factory; } + @Bean + public String tagPrefix() { + return UUID.randomUUID().toString(); + } + + @Bean + public ConsumerTagStrategy consumerTagStrategy() { + return new ConsumerTagStrategy() { + + @Override + public String createConsumerTag(String queue) { + return tagPrefix() + increment++; + } + }; + } + @Bean public CountDownLatch errorHandlerLatch() { return new CountDownLatch(1); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java index d33ef090..70307a3a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2014 the original author or authors. + * Copyright 2010-2015 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. You may obtain a copy of the License at @@ -12,8 +12,12 @@ */ package org.springframework.amqp.rabbit.listener; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import java.util.Map; +import java.util.UUID; + import org.apache.log4j.Level; import org.junit.Rule; import org.junit.Test; @@ -26,19 +30,24 @@ import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter import org.springframework.amqp.rabbit.test.BrokerRunning; import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; +import org.springframework.amqp.support.ConsumerTagStrategy; +import org.springframework.amqp.utils.test.TestUtils; /** * @author Dave Syer * @author Gunnar Hillert + * @author Gary Russell * @since 1.0 * */ public class BlockingQueueConsumerIntegrationTests { - private static Queue queue = new Queue("test.queue"); + private static Queue queue1 = new Queue("test.queue1"); + + private static Queue queue2 = new Queue("test.queue2"); @Rule - public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue); + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue1, queue2); @Rule public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.INFO, RabbitTemplate.class, @@ -56,13 +65,25 @@ public class BlockingQueueConsumerIntegrationTests { BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory, new DefaultMessagePropertiesConverter(), new ActiveObjectCounter(), - AcknowledgeMode.AUTO, true, 1, queue.getName()); + AcknowledgeMode.AUTO, true, 1, queue1.getName(), queue2.getName()); + final String consumerTagPrefix = UUID.randomUUID().toString(); + blockingQueueConsumer.setTagStrategy(new ConsumerTagStrategy() { + + @Override + public String createConsumerTag(String queue) { + return consumerTagPrefix + '#' + queue; + } + }); blockingQueueConsumer.start(); + assertNotNull(TestUtils.getPropertyValue(blockingQueueConsumer, "consumerTags", Map.class).get( + consumerTagPrefix + "#" + queue1.getName())); + assertNotNull(TestUtils.getPropertyValue(blockingQueueConsumer, "consumerTags", Map.class).get( + consumerTagPrefix + "#" + queue2.getName())); // TODO: make this into a proper assertion. An exception can be thrown here by the Rabbit client and printed to // stderr without being rethrown (so hard to make a test fail). blockingQueueConsumer.stop(); - assertNull(template.receiveAndConvert(queue.getName())); + assertNull(template.receiveAndConvert(queue1.getName())); connectionFactory.destroy(); }