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

INT-4256: AMQP: Conversion Errors to ErrorChannel

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4256

Also fix the JMS endpoint to use the `MessagingTemplate` instead of sending to the
error channel directly (ignored the send result).

Missing commit

Polishing - PR Comments

* Some additional polishing: remove extra `ifs`; make internal
classes `protected` for possible inheritors

Conflicts:
	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java

AMQP: Add Support for SCSt Error Handling

See: https://github.com/spring-cloud/spring-cloud-stream/issues/913

Add retry within `onMessage` so we have access to the converted message
as well as the original spring-amqp message, which is added to the
`ErrorMessage` as a header.

Similar to the work in spring-integration-kafka.

* Polishing JavaDocs
* Rename `AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_DATA`
to `AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE`

Conflicts:
	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java

* Use `AmqpHeaders.RAW_MESSAGE` as a retry context key
* Simple polishing for Java 8 code style
This commit is contained in:
Gary Russell
2017-05-22 10:52:14 -04:00
committed by Artem Bilan
parent b1f74577db
commit 914094e51d
4 changed files with 393 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -26,10 +26,19 @@ import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFaile
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
import com.rabbitmq.client.Channel;
@@ -47,12 +56,17 @@ import com.rabbitmq.client.Channel;
public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
OrderlyShutdownCapable {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
private final AbstractMessageListenerContainer messageListenerContainer;
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback;
public AmqpInboundChannelAdapter(AbstractMessageListenerContainer listenerContainer) {
Assert.notNull(listenerContainer, "listenerContainer must not be null");
@@ -62,6 +76,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
"configure its own listener implementation.");
this.messageListenerContainer = listenerContainer;
this.messageListenerContainer.setAutoStartup(false);
setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy());
}
@@ -75,6 +90,31 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
this.headerMapper = headerMapper;
}
/**
* Set a {@link RetryTemplate} to use for retrying a message delivery within the
* adapter. Unlike adding retry at the container level, this can be used with an
* {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the
* error channel after retries are exhausted. You generally should not configure an
* error channel when using retry here, use a {@link RecoveryCallback} instead.
* @param retryTemplate the template.
* @since 4.3.10.
* @see #setRecoveryCallback(RecoveryCallback)
*/
public void setRetryTemplate(RetryTemplate retryTemplate) {
this.retryTemplate = retryTemplate;
}
/**
* Set a {@link RecoveryCallback} when using retry within the adapter.
* @param recoveryCallback the callback.
* @since 4.3.10
* @see #setRetryTemplate(RetryTemplate)
*/
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
this.recoveryCallback = recoveryCallback;
}
@Override
public String getComponentType() {
return "amqp:inbound-channel-adapter";
@@ -82,7 +122,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
@Override
protected void onInit() {
this.messageListenerContainer.setMessageListener(new Listener());
if (this.retryTemplate != null) {
Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+ "send an error message when retries are exhausted");
}
Listener messageListener = new Listener();
if (this.retryTemplate != null) {
this.retryTemplate.registerListener(messageListener);
}
this.messageListenerContainer.setMessageListener(messageListener);
this.messageListenerContainer.afterPropertiesSet();
super.onInit();
}
@@ -97,42 +146,68 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
this.messageListenerContainer.stop();
}
/**
* {@inheritDoc}
* <p>
* Shuts down the listener container.
*/
@Override
public int beforeShutdown() {
this.stop();
return 0;
}
@Override
public int afterShutdown() {
return 0;
}
protected class Listener implements ChannelAwareMessageListener {
/**
* If there's a retry template, it will set the attributes holder via the listener. If
* there's no retry template, but there's an error channel, we create a new attributes
* holder here. If an attributes holder exists (by either method), we set the
* attributes for use by the {@link ErrorMessageStrategy}.
* @param amqpMessage the AMQP message to use.
* @param message the Spring Messaging message to use.
* @since 4.3.10
*/
private void setAttributesIfNecessary(Message amqpMessage, org.springframework.messaging.Message<?> message) {
boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
boolean needAttributes = needHolder || this.retryTemplate != null;
if (needHolder) {
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(AmqpHeaders.RAW_MESSAGE, amqpMessage);
}
}
}
@Override
protected AttributeAccessor getErrorMessageAttributes(org.springframework.messaging.Message<?> message) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes == null) {
return super.getErrorMessageAttributes(message);
}
else {
return attributes;
}
}
protected class Listener implements ChannelAwareMessageListener, RetryListener {
@SuppressWarnings("unchecked")
@Override
public void onMessage(Message message, Channel channel) throws Exception {
public void onMessage(final Message message, final Channel channel) throws Exception {
try {
Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message);
Map<String, Object> headers = AmqpInboundChannelAdapter.this.headerMapper
.toHeadersFromRequest(message.getMessageProperties());
if (AmqpInboundChannelAdapter.this.messageListenerContainer.getAcknowledgeMode()
== AcknowledgeMode.MANUAL) {
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
headers.put(AmqpHeaders.CHANNEL, channel);
if (AmqpInboundChannelAdapter.this.retryTemplate == null) {
processMessage(message, channel);
}
else {
AmqpInboundChannelAdapter.this.retryTemplate.execute(context -> {
processMessage(message, channel);
return null;
},
(RecoveryCallback<Object>) AmqpInboundChannelAdapter.this.recoveryCallback);
}
sendMessage(
getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build());
}
catch (RuntimeException e) {
if (getErrorChannel() != null) {
@@ -143,6 +218,44 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
throw e;
}
}
finally {
attributesHolder.remove();
}
}
private void processMessage(Message message, Channel channel) {
Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message);
Map<String, Object> headers = AmqpInboundChannelAdapter.this.headerMapper
.toHeadersFromRequest(message.getMessageProperties());
if (AmqpInboundChannelAdapter.this.messageListenerContainer.getAcknowledgeMode()
== AcknowledgeMode.MANUAL) {
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
headers.put(AmqpHeaders.CHANNEL, channel);
}
final org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build();
setAttributesIfNecessary(message, messagingMessage);
sendMessage(messagingMessage);
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
attributesHolder.set(context);
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
}

177
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -31,9 +31,18 @@ import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFaile
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -53,6 +62,8 @@ import com.rabbitmq.client.Channel;
*/
public class AmqpInboundGateway extends MessagingGatewaySupport {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
private final AbstractMessageListenerContainer messageListenerContainer;
private final AmqpTemplate amqpTemplate;
@@ -65,6 +76,10 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
private Address defaultReplyTo;
private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback;
public AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer) {
this(listenerContainer, new RabbitTemplate(listenerContainer.getConnectionFactory()), false);
}
@@ -92,6 +107,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
this.messageListenerContainer.setAutoStartup(false);
this.amqpTemplate = amqpTemplate;
this.amqpTemplateExplicitlySet = amqpTemplateExplicitlySet;
setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy());
}
@@ -134,6 +150,30 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
this.defaultReplyTo = new Address(defaultReplyTo);
}
/**
* Set a {@link RetryTemplate} to use for retrying a message delivery within the
* gateway. Unlike adding retry at the container level, this can be used with an
* {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the
* error channel after retries are exhausted. You generally should not configure an
* error channel when using retry here, use a {@link RecoveryCallback} instead.
* @param retryTemplate the template.
* @since 4.3.10.
* @see #setRecoveryCallback(RecoveryCallback)
*/
public void setRetryTemplate(RetryTemplate retryTemplate) {
this.retryTemplate = retryTemplate;
}
/**
* Set a {@link RecoveryCallback} when using retry within the gateway.
* @param recoveryCallback the callback.
* @since 4.3.10
* @see #setRetryTemplate(RetryTemplate)
*/
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
this.recoveryCallback = recoveryCallback;
}
@Override
public String getComponentType() {
return "amqp:inbound-gateway";
@@ -141,12 +181,26 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
@Override
protected void onInit() throws Exception {
this.messageListenerContainer.setMessageListener(new Listener());
if (this.retryTemplate != null) {
Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+ "send an error message when retries are exhausted");
}
Listener messageListener = new Listener();
if (this.retryTemplate != null) {
this.retryTemplate.registerListener(messageListener);
}
this.messageListenerContainer.setMessageListener(messageListener);
this.messageListenerContainer.afterPropertiesSet();
if (!this.amqpTemplateExplicitlySet) {
((RabbitTemplate) this.amqpTemplate).afterPropertiesSet();
}
super.onInit();
if (this.retryTemplate != null && getErrorChannel() != null) {
logger.warn("Usually, when using a RetryTemplate you should use an ErrorMessageSendingRecoverer and not "
+ "provide an errorChannel. Using an errorChannel could defeat retry and will receive an error "
+ "message for each delivery attempt.");
}
}
@Override
@@ -159,10 +213,59 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
this.messageListenerContainer.stop();
}
protected class Listener implements ChannelAwareMessageListener {
/**
* If there's a retry template, it will set the attributes holder via the listener. If
* there's no retry template, but there's an error channel, we create a new attributes
* holder here. If an attributes holder exists (by either method), we set the
* attributes for use by the {@link ErrorMessageStrategy}.
* @param amqpMessage the AMQP message to use.
* @param message the Spring Messaging message to use.
* @since 4.3.10
*/
private void setAttributesIfNecessary(Message amqpMessage, org.springframework.messaging.Message<?> message) {
boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
boolean needAttributes = needHolder || this.retryTemplate != null;
if (needHolder) {
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(AmqpHeaders.RAW_MESSAGE, amqpMessage);
}
}
}
@Override
protected AttributeAccessor getErrorMessageAttributes(org.springframework.messaging.Message<?> message) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes == null) {
return super.getErrorMessageAttributes(message);
}
else {
return attributes;
}
}
protected class Listener implements ChannelAwareMessageListener, RetryListener {
@SuppressWarnings("unchecked")
@Override
public void onMessage(Message message, Channel channel) throws Exception {
public void onMessage(final Message message, final Channel channel) throws Exception {
if (AmqpInboundGateway.this.retryTemplate == null) {
doOnMessage(message, channel);
}
else {
AmqpInboundGateway.this.retryTemplate.execute(context -> {
doOnMessage(message, channel);
return null;
},
(RecoveryCallback<Object>) AmqpInboundGateway.this.recoveryCallback);
}
}
private void doOnMessage(Message message, Channel channel) {
boolean error = false;
Map<String, Object> headers = null;
Object payload = null;
@@ -186,12 +289,12 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
}
if (!error) {
final org.springframework.messaging.Message<?> reply =
sendAndReceiveMessage(
getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build());
org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build();
setAttributesIfNecessary(message, messagingMessage);
final org.springframework.messaging.Message<?> reply = sendAndReceiveMessage(messagingMessage);
if (reply != null) {
Address replyTo;
String replyToProperty = message.getMessageProperties().getReplyTo();
@@ -202,24 +305,26 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
replyTo = AmqpInboundGateway.this.defaultReplyTo;
}
MessagePostProcessor messagePostProcessor = message1 -> {
MessageProperties messageProperties = message1.getMessageProperties();
String contentEncoding = messageProperties.getContentEncoding();
long contentLength = messageProperties.getContentLength();
String contentType = messageProperties.getContentType();
AmqpInboundGateway.this.headerMapper.fromHeadersToReply(reply.getHeaders(), messageProperties);
// clear the replyTo from the original message since we are using it now
messageProperties.setReplyTo(null);
// reset the content-* properties as determined by the MessageConverter
if (StringUtils.hasText(contentEncoding)) {
messageProperties.setContentEncoding(contentEncoding);
}
messageProperties.setContentLength(contentLength);
if (contentType != null) {
messageProperties.setContentType(contentType);
}
return message1;
};
MessagePostProcessor messagePostProcessor =
message1 -> {
MessageProperties messageProperties = message1.getMessageProperties();
String contentEncoding = messageProperties.getContentEncoding();
long contentLength = messageProperties.getContentLength();
String contentType = messageProperties.getContentType();
AmqpInboundGateway.this.headerMapper.fromHeadersToReply(reply.getHeaders(),
messageProperties);
// clear the replyTo from the original message since we are using it now
messageProperties.setReplyTo(null);
// reset the content-* properties as determined by the MessageConverter
if (StringUtils.hasText(contentEncoding)) {
messageProperties.setContentEncoding(contentEncoding);
}
messageProperties.setContentLength(contentLength);
if (contentType != null) {
messageProperties.setContentType(contentType);
}
return message1;
};
if (replyTo != null) {
AmqpInboundGateway.this.amqpTemplate.convertAndSend(replyTo.getExchangeName(),
@@ -239,6 +344,24 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
}
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
attributesHolder.set(context);
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
}
}

49
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* Copyright 2017 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.integration.amqp.support;
import java.util.Collections;
import java.util.Map;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
/**
* An {@link ErrorMessageStrategy} extension that adds the raw AMQP message as
* a header to the {@link ErrorMessage}.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.3.10
*
*/
public class AmqpMessageHeaderErrorMessageStrategy implements ErrorMessageStrategy {
@Override
public ErrorMessage buildErrorMessage(Throwable throwable, AttributeAccessor context) {
Object inputMessage = context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY);
Map<String, Object> headers =
Collections.singletonMap(AmqpHeaders.RAW_MESSAGE, context.getAttribute(AmqpHeaders.RAW_MESSAGE));
return new ErrorMessage(throwable, headers, inputMessage instanceof Message ? (Message<?>) inputMessage : null);
}
}

59
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -16,12 +16,16 @@
package org.springframework.integration.amqp.inbound;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
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 static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
@@ -45,6 +49,7 @@ import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
@@ -52,9 +57,11 @@ import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
import org.springframework.integration.json.JsonToObjectTransformer;
import org.springframework.integration.json.ObjectToJsonTransformer;
import org.springframework.integration.mapping.support.JsonHeaders;
@@ -62,8 +69,10 @@ import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.integration.transformer.Transformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.retry.support.RetryTemplate;
import com.rabbitmq.client.Channel;
@@ -280,6 +289,56 @@ public class InboundEndpointTests {
assertNotNull(errorChannel.receive(0));
}
@Test
public void testRetryWithinOnMessageAdapter() throws Exception {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
AbstractMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
adapter.setOutputChannel(new DirectChannel());
adapter.setRetryTemplate(new RetryTemplate());
QueueChannel errors = new QueueChannel();
ErrorMessageSendingRecoverer recoveryCallback = new ErrorMessageSendingRecoverer(errors);
recoveryCallback.setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy());
adapter.setRecoveryCallback(recoveryCallback);
adapter.afterPropertiesSet();
ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
listener.onMessage(org.springframework.amqp.core.MessageBuilder.withBody("foo".getBytes())
.andProperties(new MessageProperties()).build(), null);
Message<?> errorMessage = errors.receive(0);
assertNotNull(errorMessage);
assertThat(errorMessage.getPayload(), instanceOf(MessagingException.class));
assertThat(((MessagingException) errorMessage.getPayload()).getMessage(), containsString("Dispatcher has no"));
org.springframework.amqp.core.Message amqpMessage = errorMessage.getHeaders()
.get(AmqpHeaders.RAW_MESSAGE, org.springframework.amqp.core.Message.class);
assertThat(amqpMessage, notNullValue());
assertNull(errors.receive(0));
}
@Test
public void testRetryWithinOnMessageGateway() throws Exception {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
AbstractMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
AmqpInboundGateway adapter = new AmqpInboundGateway(container);
adapter.setRequestChannel(new DirectChannel());
adapter.setRetryTemplate(new RetryTemplate());
QueueChannel errors = new QueueChannel();
ErrorMessageSendingRecoverer recoveryCallback = new ErrorMessageSendingRecoverer(errors);
recoveryCallback.setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy());
adapter.setRecoveryCallback(recoveryCallback);
adapter.afterPropertiesSet();
ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
listener.onMessage(org.springframework.amqp.core.MessageBuilder.withBody("foo".getBytes())
.andProperties(new MessageProperties()).build(), null);
Message<?> errorMessage = errors.receive(0);
assertNotNull(errorMessage);
assertThat(errorMessage.getPayload(), instanceOf(MessagingException.class));
assertThat(((MessagingException) errorMessage.getPayload()).getMessage(), containsString("Dispatcher has no"));
org.springframework.amqp.core.Message amqpMessage = errorMessage.getHeaders()
.get(AmqpHeaders.RAW_MESSAGE, org.springframework.amqp.core.Message.class);
assertThat(amqpMessage, notNullValue());
assertNull(errors.receive(0));
}
public static class Foo {
private String bar;