Skip to content

Commit

Permalink
AMQP-849: RT and DRTMLC - add ErrorHandler
Browse files Browse the repository at this point in the history 
JIRA: https://jira.spring.io/browse/AMQP-849

Support the configuration of an error handler for exceptions when
delivering replies (e.g. late replies).

**cherry-pick to 2.0.x with adjustments to assertJ and docs**

* Polishing - PR Comments

* More polishing

* Replace AssertJ with regular JUnit assertions

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateDirectReplyToContainerIntegrationTests.java
#	src/reference/asciidoc/amqp.adoc
#	src/reference/asciidoc/whats-new.adoc
  • Loading branch information
@garyrussell @artembilan
garyrussell authored and artembilan committed Dec 13, 2018
1 parent f1cc5d7 commit 08f54ce
Show file tree
Hide file tree
Showing 3 changed files with 121 additions and 34 deletions.
84 changes: 51 additions & 33 deletions spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@
import org.springframework.retry.RetryCallback;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;

import com.rabbitmq.client.AMQP;
Expand Down Expand Up @@ -185,60 +186,52 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,

private final AtomicInteger containerInstance = new AtomicInteger();

private volatile String exchange = DEFAULT_EXCHANGE;
private String exchange = DEFAULT_EXCHANGE;

private volatile String routingKey = DEFAULT_ROUTING_KEY;
private String routingKey = DEFAULT_ROUTING_KEY;

// The default queue name that will be used for synchronous receives.
private volatile String queue;

private volatile long receiveTimeout = 0;
private long receiveTimeout = 0;

private volatile long replyTimeout = DEFAULT_REPLY_TIMEOUT;
private long replyTimeout = DEFAULT_REPLY_TIMEOUT;

private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private MessageConverter messageConverter = new SimpleMessageConverter();

private volatile MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
private MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();

private volatile String encoding = DEFAULT_ENCODING;
private String encoding = DEFAULT_ENCODING;

private volatile String replyAddress;
private String replyAddress;

private volatile ConfirmCallback confirmCallback;
private ConfirmCallback confirmCallback;

private volatile ReturnCallback returnCallback;
private ReturnCallback returnCallback;

private volatile Boolean confirmsOrReturnsCapable;

private volatile Expression mandatoryExpression = new ValueExpression<Boolean>(false);

private volatile String correlationKey = null;
private Expression mandatoryExpression = new ValueExpression<Boolean>(false);

private volatile RetryTemplate retryTemplate;
private String correlationKey = null;

private volatile RecoveryCallback<?> recoveryCallback;
private RetryTemplate retryTemplate;

private volatile Expression sendConnectionFactorySelectorExpression;

private volatile Expression receiveConnectionFactorySelectorExpression;

private volatile boolean usingFastReplyTo;
private RecoveryCallback<?> recoveryCallback;

private volatile boolean useDirectReplyToContainer = true;
private Expression sendConnectionFactorySelectorExpression;

private volatile boolean evaluatedFastReplyTo;
private Expression receiveConnectionFactorySelectorExpression;

private volatile boolean useTemporaryReplyQueues;
private boolean useDirectReplyToContainer = true;

private volatile Collection<MessagePostProcessor> beforePublishPostProcessors;
private boolean useTemporaryReplyQueues;

private volatile Collection<MessagePostProcessor> afterReceivePostProcessors;
private Collection<MessagePostProcessor> beforePublishPostProcessors;

private volatile CorrelationDataPostProcessor correlationDataPostProcessor;
private Collection<MessagePostProcessor> afterReceivePostProcessors;

private volatile boolean isListener;
private CorrelationDataPostProcessor correlationDataPostProcessor;

private volatile Expression userIdExpression;
private Expression userIdExpression;

private String beanName = "rabbitTemplate";

Expand All @@ -248,6 +241,18 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,

private boolean usePublisherConnection;

private ErrorHandler replyErrorHandler;

private volatile Boolean confirmsOrReturnsCapable;

private volatile boolean publisherConfirms;

private volatile boolean usingFastReplyTo;

private volatile boolean evaluatedFastReplyTo;

private volatile boolean isListener;

/**
* Convenient constructor for use with setter injection. Don't forget to set the connection factory.
*/
Expand Down Expand Up @@ -694,9 +699,20 @@ public void setUsePublisherConnection(boolean usePublisherConnection) {
this.usePublisherConnection = usePublisherConnection;
}

/**
* When using a direct reply-to container for request/reply operations, set an error
* handler to be invoked when a reply delivery fails (e.g. due to a late reply).
* @param replyErrorHandler the reply error handler
* @since 2.0.11
* @see #setUseDirectReplyToContainer(boolean)
*/
public void setReplyErrorHandler(ErrorHandler replyErrorHandler) {
this.replyErrorHandler = replyErrorHandler;
}

/**
* Invoked by the container during startup so it can verify the queue is correctly
* configured (if a simple reply queue name is used instead of exchange/routingKey.
* configured (if a simple reply queue name is used instead of exchange/routingKey).
* @return the queue name, if configured.
* @since 1.5
*/
Expand Down Expand Up @@ -1654,6 +1670,9 @@ private Message doSendAndReceiveWithDirect(String exchange, String routingKey, M
container.setAfterReceivePostProcessors(this.afterReceivePostProcessors
.toArray(new MessagePostProcessor[this.afterReceivePostProcessors.size()]));
}
if (this.replyErrorHandler != null) {
container.setErrorHandler(this.replyErrorHandler);
}
container.start();
this.directReplyToContainers.put(connectionFactory, container);
this.replyAddress = Address.AMQ_RABBITMQ_REPLY_TO;
Expand Down Expand Up @@ -2231,8 +2250,7 @@ public void onMessage(Message message) {
.getHeaders().get(this.correlationKey);
}
if (messageTag == null) {
logger.error("No correlation header in reply");
return;
throw new AmqpRejectAndDontRequeueException("No correlation header in reply");
}

PendingReply pendingReply = this.replyHolder.get(messageTag);
Expand Down
67 changes: 66 additions & 1 deletion ...pringframework/amqp/rabbit/core/RabbitTemplateDirectReplyToContainerIntegrationTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,16 +16,32 @@

package org.springframework.amqp.rabbit.core;

import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;

import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.junit.Test;

import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.util.ErrorHandler;

/**
* @author Gary Russell
Expand All @@ -46,17 +62,66 @@ protected RabbitTemplate createSendAndReceiveRabbitTemplate(ConnectionFactory co

@SuppressWarnings("unchecked")
@Test
public void channelReleasedOnTimeout() {
public void channelReleasedOnTimeout() throws Exception {
final CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
RabbitTemplate template = createSendAndReceiveRabbitTemplate(connectionFactory);
template.setReplyTimeout(1);
AtomicReference<Throwable> exception = new AtomicReference<>();
CountDownLatch latch = new CountDownLatch(1);
ErrorHandler replyErrorHandler = t -> {
exception.set(t);
latch.countDown();
};
template.setReplyErrorHandler(replyErrorHandler);
Object reply = template.convertSendAndReceive(ROUTE, "foo");
assertNull(reply);
Object container = TestUtils.getPropertyValue(template, "directReplyToContainers", Map.class)
.get(template.isUsePublisherConnection()
? connectionFactory.getPublisherConnectionFactory()
: connectionFactory);
assertEquals(0, TestUtils.getPropertyValue(container, "inUseConsumerChannels", Map.class).size());
assertSame(replyErrorHandler, TestUtils.getPropertyValue(container, "errorHandler"));
Message replyMessage = new Message("foo".getBytes(), new MessageProperties());
try {
template.onMessage(replyMessage);
}
catch (Exception e) {
assertThat(e, instanceOf(AmqpRejectAndDontRequeueException.class));
assertEquals("No correlation header in reply", e.getMessage());
}

replyMessage.getMessageProperties().setCorrelationId("foo");

try {
template.onMessage(replyMessage);
}
catch (Exception e) {
assertThat(e, instanceOf(AmqpRejectAndDontRequeueException.class));
assertEquals("Reply received after timeout", e.getMessage());
}

ExecutorService executor = Executors.newFixedThreadPool(1);
// Set up a consumer to respond to our producer
executor.submit(() -> {
Message message = template.receive(ROUTE, 10_000);
assertNotNull("No message received", message);
template.send(message.getMessageProperties().getReplyTo(), replyMessage);
return message;
});
while (template.receive(ROUTE, 100) != null) {
// empty
}
reply = template.convertSendAndReceive(ROUTE, "foo");
assertNull(reply);
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertThat(exception.get(), instanceOf(ListenerExecutionFailedException.class));
assertEquals("Reply received after timeout", exception.get().getCause().getMessage());
assertArrayEquals(replyMessage.getBody(),
((ListenerExecutionFailedException) exception.get()).getFailedMessage().getBody());
assertEquals(0, TestUtils.getPropertyValue(container, "inUseConsumerChannels", Map.class).size());
executor.shutdownNow();
template.stop();
connectionFactory.destroy();
}

}
4 changes: 4 additions & 0 deletions src/reference/asciidoc/amqp.adoc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2993,6 +2993,10 @@ NOTE: This feature uses publisher returns and is enabled by setting `publisherRe
`CachingConnectionFactory` (see <<cf-pub-conf-ret>>).
Also, you must not have registered your own `ReturnCallback` with the `RabbitTemplate`.

Starting with versions 2.0.11, 2.1.3, when using the default `DirectReplyToMessageListenerContainer`, you can add an error handler by setting the template's `replyErrorHandler` property.
This error handler will be invoked for any failed deliveries, such as late replies and messages received without a correlation header.
The exception passed in is a `ListenerExecutionFailedException` which has a `failedMessage` property.

[[direct-reply-to]]
===== RabbitMQ Direct reply-to

Expand Down

0 comments on commit 08f54ce

Please sign in to comment.