From b5af780ad028f90a7c98f8dbcc16d7deb36f8bc0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 10 Jul 2014 18:43:27 -0400 Subject: [PATCH] AMQP-409 Allow Late Replies to go to a DLQ JIRA: https://jira.spring.io/browse/AMQP-409 When a late reply is received, instead of just logging the event, reject the delivery so that (if so configured) the late reply can be routed to a DLX/DLQ. --- .../amqp/rabbit/core/RabbitTemplate.java | 2 + .../core/FixedReplyQueueDeadLetterTests.java | 230 ++++++++++++++++++ .../JavaConfigFixedReplyQueueTests.java | 2 + src/reference/docbook/amqp.xml | 19 ++ 4 files changed, 253 insertions(+) create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 4cd5f347..861abaac 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit; import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIllegalStateException; +import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.core.Address; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; @@ -1052,6 +1053,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, if (logger.isWarnEnabled()) { logger.warn("Reply received after timeout for " + messageTag); } + throw new AmqpRejectAndDontRequeueException("Reply received after timeout"); } else { // Restore the inbound correlation data diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java new file mode 100644 index 00000000..983209a7 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java @@ -0,0 +1,230 @@ +/* + * Copyright 2014 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.rabbit.core; + +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.DirectExchange; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.FixedReplyQueueDeadLetterTests.FixedReplyQueueDeadLetterConfig; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; +import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * + * @author Gary Russell + * @since 1.3.6 + */ + +@ContextConfiguration(classes=FixedReplyQueueDeadLetterConfig.class) +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class FixedReplyQueueDeadLetterTests { + + @Autowired + private RabbitTemplate rabbitTemplate; + + @Autowired + private DeadListener deadListener; + + @Rule + public BrokerRunning brokerRunning = BrokerRunning.isRunning(); + + /** + * Sends a message to a service that upcases the String and returns as a reply + * using a {@link RabbitTemplate} configured with a fixed reply queue and + * reply listener, configured with JavaConfig. We expect the reply to time + * out and be sent to the DLQ. + * @throws Exception the exception. + */ + @Test + public void test() throws Exception { + assertNull(this.rabbitTemplate.convertSendAndReceive("foo")); + assertTrue(this.deadListener.latch.await(10, TimeUnit.SECONDS)); + } + + @Configuration + public static class FixedReplyQueueDeadLetterConfig { + + @Bean + public ConnectionFactory rabbitConnectionFactory() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + connectionFactory.setHost("localhost"); + return connectionFactory; + } + + /** + * @return Rabbit template with fixed reply queue and small timeout. + */ + @Bean + public RabbitTemplate fixedReplyQRabbitTemplate() { + RabbitTemplate template = new RabbitTemplate(rabbitConnectionFactory()); + template.setExchange(ex().getName()); + template.setRoutingKey("dlx.reply.test"); + template.setReplyQueue(replyQueue()); + template.setReplyTimeout(1); + return template; + } + + /** + * @return The reply listener container - the rabbit template is the listener. + */ + @Bean + public SimpleMessageListenerContainer replyListenerContainer() { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(rabbitConnectionFactory()); + container.setQueues(replyQueue()); + container.setMessageListener(fixedReplyQRabbitTemplate()); + return container; + } + + /** + * @return The listener container that handles the request and returns the reply. + */ + @Bean + public SimpleMessageListenerContainer serviceListenerContainer() { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(rabbitConnectionFactory()); + container.setQueues(requestQueue()); + container.setMessageListener(new MessageListenerAdapter(new PojoListener())); + return container; + } + + /** + * @return The listener container that handles the dead letter. + */ + @Bean + public SimpleMessageListenerContainer dlListenerContainer() { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(rabbitConnectionFactory()); + container.setQueues(dlq()); + container.setMessageListener(new MessageListenerAdapter(deadListener())); + return container; + } + + /** + * @return a non-durable auto-delete exchange. + */ + @Bean + public DirectExchange ex() { + return new DirectExchange("dlx.test.requestEx", false, true); + } + + @Bean + public Binding binding() { + return BindingBuilder.bind(requestQueue()).to(ex()).with("dlx.reply.test"); + } + + /** + * @return a non-durable auto-delete exchange. + */ + @Bean + public DirectExchange dlx() { + return new DirectExchange("reply.dlx", false, true); + } + + /** + * @return an (auto-delete) queue. + */ + @Bean + public Queue requestQueue() { + return new Queue("dlx.test.requestQ", false, false, true); + } + + /** + * @return an (auto-delete) queue configured to go to the dlx. + */ + @Bean + public Queue replyQueue() { + Map args = new HashMap(); + args.put("x-dead-letter-exchange", "reply.dlx"); + return new Queue("dlx.test.replyQ", false, false, true, args); + } + + /** + * @return an (auto-delete) queue. + */ + @Bean + public Queue dlq() { + return new Queue("dlx.test.DLQ", false, false, true); + } + + @Bean + public DeadListener deadListener() { + return new DeadListener(); + } + + /** + * Bind the dlq to the dlx with the reply queue name. + * @return The binding. + */ + @Bean + public Binding dlBinding() { + return BindingBuilder.bind(dlq()).to(dlx()).with(replyQueue().getName()); + } + + /** + * @return an admin to handle the declarations. + */ + @Bean + public RabbitAdmin admin() { + return new RabbitAdmin(rabbitConnectionFactory()); + } + + /** + * Listener that upcases the request, but takes too long. + */ + public static class PojoListener { + + public String handleMessage(String foo) throws Exception { + Thread.sleep(500); + return foo.toUpperCase(); + } + } + + } + + public static class DeadListener { + private final CountDownLatch latch = new CountDownLatch(1); + + public void handleMessage(String foo) { + latch.countDown(); + } + + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java index 5b1ef9f5..f8680928 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java @@ -38,6 +38,7 @@ import org.springframework.amqp.rabbit.test.BrokerRunning; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -51,6 +52,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @ContextConfiguration(classes=FixedReplyQueueConfig.class) @RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class JavaConfigFixedReplyQueueTests { @Autowired diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index 934726d4..fb756627 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -1288,6 +1288,25 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]>this test case. + + + When the reply times out (replyTimeout), the sendAndReceive() methods return + null. + + + Prior to version 1.3.6, late replies for timed out messages were simply logged. Now, if + a late reply is received, it is rejected (the template throws an + AmqpRejectAndDontRequeueException). If the reply queue is configured to send + rejected messages to a dead letter exchange, the reply can be retrieved for later analysis. Simply + bind a queue to the configured dead letter exchange with a routing key equal to the reply queue's + name. + + + Refer to the RabbitMQ Dead Letter Documentation + for more information about configuring dead lettering. You can also take a look at the + FixedReplyQueueDeadLetterTests test case for an example. + +