diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/AmqpRejectAndDontRequeueException.java b/spring-amqp-core/src/main/java/org/springframework/amqp/AmqpRejectAndDontRequeueException.java new file mode 100644 index 00000000..00f87061 --- /dev/null +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/AmqpRejectAndDontRequeueException.java @@ -0,0 +1,41 @@ +/* + * Copyright 2002-2012 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; + +/** + * Exception for listener implementations used to indicate the + * basic.reject will be sent with requeue=false in order to enable + * features such as DLQ. + * @author Gary Russell + * @since 1.0.1 + * + */ +@SuppressWarnings("serial") +public class AmqpRejectAndDontRequeueException extends AmqpException { + + public AmqpRejectAndDontRequeueException(String message, Throwable cause) { + super(message, cause); + } + + public AmqpRejectAndDontRequeueException(String message) { + super(message); + } + + public AmqpRejectAndDontRequeueException(Throwable cause) { + super(cause); + } + +} diff --git a/spring-amqp-parent/pom.xml b/spring-amqp-parent/pom.xml index 1f0ba8a9..b9c01244 100644 --- a/spring-amqp-parent/pom.xml +++ b/spring-amqp-parent/pom.xml @@ -20,7 +20,7 @@ 1.8.4 1.4.3 1.5.3 - 2.7.1 + 2.8.1 3.0.5.RELEASE diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java index f4537e7c..022d8b28 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2011 the original author or authors. + * Copyright 2010-2012 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 @@ -33,6 +33,7 @@ import org.w3c.dom.NodeList; /** * @author Mark Fisher + * @author Gary Russell * @since 1.0 */ class ListenerContainerParser implements BeanDefinitionParser { @@ -83,6 +84,7 @@ class ListenerContainerParser implements BeanDefinitionParser { private static final String ADVICE_CHAIN_ATTRIBUTE = "advice-chain"; + private static final String REQUEUE_REJECTED_ATTRIBUTE = "requeue-rejected"; public BeanDefinition parse(Element element, ParserContext parserContext) { CompositeComponentDefinition compositeDef = new CompositeComponentDefinition(element.getTagName(), @@ -243,6 +245,11 @@ class ListenerContainerParser implements BeanDefinitionParser { containerDef.getPropertyValues().add("txSize", new TypedStringValue(transactionSize)); } + String requeueRejected = containerEle.getAttribute(REQUEUE_REJECTED_ATTRIBUTE); + if (StringUtils.hasText(requeueRejected)) { + containerDef.getPropertyValues().add("defaultRequeueRejected", new TypedStringValue(requeueRejected)); + } + String phase = containerEle.getAttribute(PHASE_ATTRIBUTE); if (StringUtils.hasText(phase)) { containerDef.getPropertyValues().add("phase", phase); 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 a8713d58..06c0d04b 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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 @@ -25,6 +25,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.amqp.AmqpException; +import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; @@ -46,6 +47,7 @@ import com.rabbitmq.utility.Utility; * * @author Mark Pollack * @author Dave Syer + * @author Gary Russell * */ public class BlockingQueueConsumer { @@ -80,6 +82,20 @@ public class BlockingQueueConsumer { private Set deliveryTags = new LinkedHashSet(); + private final boolean defaultRequeuRejected; + + /** + * Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker + * until it is started. RequeueRejected defaults to true. + */ + public BlockingQueueConsumer(ConnectionFactory connectionFactory, + MessagePropertiesConverter messagePropertiesConverter, + ActiveObjectCounter activeObjectCounter, AcknowledgeMode acknowledgeMode, + boolean transactional, int prefetchCount, String... queues) { + this(connectionFactory, messagePropertiesConverter, activeObjectCounter, + acknowledgeMode, transactional, prefetchCount, true, queues); + } + /** * Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker * until it is started. @@ -87,13 +103,14 @@ public class BlockingQueueConsumer { public BlockingQueueConsumer(ConnectionFactory connectionFactory, MessagePropertiesConverter messagePropertiesConverter, ActiveObjectCounter activeObjectCounter, AcknowledgeMode acknowledgeMode, - boolean transactional, int prefetchCount, String... queues) { + boolean transactional, int prefetchCount, boolean defaultRequeueRejected, String... queues) { this.connectionFactory = connectionFactory; this.messagePropertiesConverter = messagePropertiesConverter; this.activeObjectCounter = activeObjectCounter; this.acknowledgeMode = acknowledgeMode; this.transactional = transactional; this.prefetchCount = prefetchCount; + this.defaultRequeuRejected = defaultRequeueRejected; this.queues = queues; } @@ -315,9 +332,17 @@ public class BlockingQueueConsumer { if (logger.isDebugEnabled()) { logger.debug("Rejecting messages"); } + boolean shouldRequeue = this.defaultRequeuRejected; + Throwable t = ex; + while (shouldRequeue && t != null) { + if (t instanceof AmqpRejectAndDontRequeueException) { + shouldRequeue = false; + } + t = t.getCause(); + } for (Long deliveryTag : deliveryTags) { // With newer RabbitMQ brokers could use basicNack here... - channel.basicReject(deliveryTag, true); + channel.basicReject(deliveryTag, shouldRequeue); } if (transactional) { // Need to commit the reject (=nack) 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 36ab6f51..cd3bf4dc 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 @@ -23,6 +23,7 @@ import java.util.concurrent.TimeoutException; import org.aopalliance.aop.Advice; import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIllegalStateException; +import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.ImmediateAcknowledgeAmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; @@ -94,6 +95,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private volatile MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter(); + private volatile boolean defaultRequeueRejected = true; + public static interface ContainerDelegate { void invokeListener(Channel channel, Message message) throws Exception; } @@ -218,6 +221,19 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta this.messagePropertiesConverter = messagePropertiesConverter; } + /** + * Determines 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 + * versions of Rabbit that support dead-lettering, the message must not be requeued in order + * to be sent to the dead letter exchange. Setting to false causes all rejections to not + * be requeued. When true, the default can be overridden by the listener throwing an + * {@link AmqpRejectAndDontRequeueException}. Default true. + * @param defaultRequeueRejected + */ + public void setDefaultRequeueRejected(boolean defaultRequeueRejected) { + this.defaultRequeueRejected = defaultRequeueRejected; + } + /** * Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent * consumers. @@ -378,7 +394,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta // didn't get an ack for delivered messages int actualPrefetchCount = prefetchCount > txSize ? prefetchCount : txSize; consumer = new BlockingQueueConsumer(getConnectionFactory(), this.messagePropertiesConverter, cancellationLock, - getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount, queues); + getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount, this.defaultRequeueRejected, queues); return consumer; } diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd index 6255d7e1..ad08e113 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd @@ -480,6 +480,15 @@ ]]> + + + + + (), + AcknowledgeMode.AUTO, true, 1, "testQ"); + testRequeueOrNotGuts(ex, requeue, channel, blockingQueueConsumer); + } + + private void testRequeueOrNotDefaultNot(Exception ex, boolean requeue) + throws Exception, IOException { + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + Channel channel = mock(Channel.class); + BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory, + new DefaultMessagePropertiesConverter(), new ActiveObjectCounter(), + AcknowledgeMode.AUTO, true, 1, false, "testQ"); + testRequeueOrNotGuts(ex, requeue, channel, blockingQueueConsumer); + } + + private void testRequeueOrNotGuts(Exception ex, boolean requeue, + Channel channel, BlockingQueueConsumer blockingQueueConsumer) + throws Exception, IOException { + DirectFieldAccessor dfa = new DirectFieldAccessor(blockingQueueConsumer); + dfa.setPropertyValue("channel", channel); + Set deliveryTags = new HashSet(); + deliveryTags.add(1L); + dfa.setPropertyValue("deliveryTags", deliveryTags); + blockingQueueConsumer.rollbackOnExceptionIfNecessary(ex); + Mockito.verify(channel).basicReject(1L, requeue); + } + +} diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml index edc86df6..b570e6ab 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml @@ -28,6 +28,10 @@ + + + +