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

AMQP-493: Add Consumer Tag Naming Strategy

Browse Source 
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) : "")`
This commit is contained in:
Gary Russell
2015-04-23 09:59:31 +01:00
committed by Artem Bilan
parent 6947b31a67
commit fa8cf62440
6 changed files with 140 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-amqp/src/main/java/org/springframework/amqp/support/ConsumerTagStrategy.java Normal file
View File

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

17
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java
View File

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

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

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

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

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

36
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
View File

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

31
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java
View File

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