diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index 9a89f92550..fdf5cd44a7 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -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 attributesHolder = new ThreadLocal(); + private final AbstractMessageListenerContainer messageListenerContainer; private volatile MessageConverter messageConverter = new SimpleMessageConverter(); private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper(); + private RetryTemplate retryTemplate; + + private RecoveryCallback 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 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} - *

- * 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 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) 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 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 messagingMessage = getMessageBuilderFactory() + .withPayload(payload) + .copyHeaders(headers) + .build(); + setAttributesIfNecessary(message, messagingMessage); + sendMessage(messagingMessage); + } + + @Override + public boolean open(RetryContext context, RetryCallback callback) { + attributesHolder.set(context); + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index 98d0ad2aa2..112b4f3929 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -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 attributesHolder = new ThreadLocal(); + 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 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 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) AmqpInboundGateway.this.recoveryCallback); + } + } + + private void doOnMessage(Message message, Channel channel) { boolean error = false; Map 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 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 boolean open(RetryContext context, RetryCallback callback) { + attributesHolder.set(context); + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java new file mode 100644 index 0000000000..35eaca1033 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java @@ -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 headers = + Collections.singletonMap(AmqpHeaders.RAW_MESSAGE, context.getAttribute(AmqpHeaders.RAW_MESSAGE)); + return new ErrorMessage(throwable, headers, inputMessage instanceof Message ? (Message) inputMessage : null); + } + +} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index 84b6366b07..3cb313bfde 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -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;