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.
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
7b49bde611
commit
b5af780ad0
@@ -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
|
||||
|
||||
@@ -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<String, Object> args = new HashMap<String, Object>();
|
||||
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();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -1288,6 +1288,25 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]></programlist
|
||||
<ulink url="https://github.com/spring-projects/spring-amqp/tree/master/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java"
|
||||
>this test case</ulink>.
|
||||
</para>
|
||||
<important>
|
||||
<para>
|
||||
When the reply times out (<code>replyTimeout</code>), the <code>sendAndReceive()</code> methods return
|
||||
null.
|
||||
</para>
|
||||
<para>
|
||||
Prior to <emphasis>version 1.3.6</emphasis>, late replies for timed out messages were simply logged. Now, if
|
||||
a late reply is received, it is rejected (the template throws an
|
||||
<classname>AmqpRejectAndDontRequeueException</classname>). 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.
|
||||
</para>
|
||||
<para>
|
||||
Refer to the <ulink url="https://www.rabbitmq.com/dlx.html">RabbitMQ Dead Letter Documentation</ulink>
|
||||
for more information about configuring dead lettering. You can also take a look at the
|
||||
<classname>FixedReplyQueueDeadLetterTests</classname> test case for an example.
|
||||
</para>
|
||||
</important>
|
||||
</section>
|
||||
</section>
|
||||
<section id="remoting">
|
||||
|
||||
Reference in New Issue
Block a user