From 738d0b0757b8e7f0b05f0d1505e8c233af9c3d41 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 22 May 2017 10:52:14 -0400 Subject: [PATCH] INT-4256: AMQP: Conversion Errors to ErrorChannel 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` --- .../inbound/AmqpInboundChannelAdapter.java | 188 +++++++++++++--- .../amqp/inbound/AmqpInboundGateway.java | 204 +++++++++++++++--- ...AmqpMessageHeaderErrorMessageStrategy.java | 56 +++++ .../amqp/inbound/InboundEndpointTests.java | 128 +++++++++++ .../ChannelPublishingJmsMessageListener.java | 12 +- 5 files changed, 530 insertions(+), 58 deletions(-) create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java 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 011b29c20d..f93ef9ff35 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -22,13 +22,23 @@ import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException; 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; @@ -40,17 +50,23 @@ import com.rabbitmq.client.Channel; * @author Mark Fisher * @author Gary Russell * @author Artem Bilan + * * @since 2.1 */ 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"); @@ -60,6 +76,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements "configure its own listener implementation."); this.messageListenerContainer = listenerContainer; this.messageListenerContainer.setAutoStartup(false); + setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy()); } @@ -73,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"; @@ -80,22 +122,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements @Override protected void onInit() { - this.messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() { - - @Override - public void onMessage(Message message, Channel channel) throws Exception { - 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); - } - sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build()); - } - - }); + 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(); } @@ -110,26 +146,122 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements this.messageListenerContainer.stop(); } - - /** - * {@inheritDoc} - *

- * Shuts down the listener container. - */ @Override public int beforeShutdown() { this.stop(); return 0; } - - /** - * {@inheritDoc} - *

No-op - */ @Override public int afterShutdown() { return 0; } + /** + * 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(AmqpMessageHeaderErrorMessageStrategy.AMQP_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(final Message message, final Channel channel) throws Exception { + try { + if (AmqpInboundChannelAdapter.this.retryTemplate == null) { + processMessage(message, channel); + } + else { + AmqpInboundChannelAdapter.this.retryTemplate.execute(new RetryCallback() { + + @Override + public Void doWithRetry(RetryContext context) throws RuntimeException { + processMessage(message, channel); + return null; + } + + }, (RecoveryCallback) AmqpInboundChannelAdapter.this.recoveryCallback); + } + } + catch (RuntimeException e) { + if (getErrorChannel() != null) { + getMessagingTemplate().send(getErrorChannel(), buildErrorMessage(null, + new ListenerExecutionFailedException("Message conversion failed", e, message))); + } + else { + 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 d755f757a1..8022c8f768 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -28,12 +28,22 @@ import org.springframework.amqp.core.MessageProperties; 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.exception.ListenerExecutionFailedException; 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; @@ -47,10 +57,14 @@ import com.rabbitmq.client.Channel; * * @author Mark Fisher * @author Artem Bilan + * @author Gary Russell + * * @since 2.1 */ public class AmqpInboundGateway extends MessagingGatewaySupport { + private static final ThreadLocal attributesHolder = new ThreadLocal(); + private final AbstractMessageListenerContainer messageListenerContainer; private final AmqpTemplate amqpTemplate; @@ -63,6 +77,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); } @@ -90,6 +108,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { this.messageListenerContainer.setAutoStartup(false); this.amqpTemplate = amqpTemplate; this.amqpTemplateExplicitlySet = amqpTemplateExplicitlySet; + setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy()); } @@ -132,6 +151,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"; @@ -139,19 +182,124 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { @Override protected void onInit() throws Exception { - this.messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() { - @Override - public void onMessage(Message message, Channel channel) { - Object payload = AmqpInboundGateway.this.amqpMessageConverter.fromMessage(message); - Map headers = - AmqpInboundGateway.this.headerMapper.toHeadersFromRequest(message.getMessageProperties()); + 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 + protected void doStart() { + this.messageListenerContainer.start(); + } + + @Override + protected void doStop() { + this.messageListenerContainer.stop(); + } + + /** + * 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(AmqpMessageHeaderErrorMessageStrategy.AMQP_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(final Message message, final Channel channel) throws Exception { + if (AmqpInboundGateway.this.retryTemplate == null) { + doOnMessage(message, channel); + } + else { + AmqpInboundGateway.this.retryTemplate.execute(new RetryCallback() { + + @Override + public Object doWithRetry(RetryContext context) throws RuntimeException { + 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; + try { + payload = AmqpInboundGateway.this.amqpMessageConverter.fromMessage(message); + headers = AmqpInboundGateway.this.headerMapper.toHeadersFromRequest(message.getMessageProperties()); if (AmqpInboundGateway.this.messageListenerContainer.getAcknowledgeMode() == AcknowledgeMode.MANUAL) { headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag()); headers.put(AmqpHeaders.CHANNEL, channel); } - org.springframework.messaging.Message request = - getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build(); - final org.springframework.messaging.Message reply = sendAndReceiveMessage(request); + } + catch (RuntimeException e) { + if (getErrorChannel() != null) { + AmqpInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null, + new ListenerExecutionFailedException("Message conversion failed", e, message))); + } + else { + throw e; + } + error = true; + } + + if (!error) { + 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(); @@ -170,8 +318,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { String contentEncoding = messageProperties.getContentEncoding(); long contentLength = messageProperties.getContentLength(); String contentType = messageProperties.getContentType(); - AmqpInboundGateway.this.headerMapper.fromHeadersToReply(reply.getHeaders(), - messageProperties); + 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 @@ -183,7 +330,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { messageProperties.setContentType(contentType); } return message; - } + }; }; @@ -203,23 +350,26 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { } } } - - }); - this.messageListenerContainer.afterPropertiesSet(); - if (!this.amqpTemplateExplicitlySet) { - ((RabbitTemplate) this.amqpTemplate).afterPropertiesSet(); } - super.onInit(); - } - @Override - protected void doStart() { - this.messageListenerContainer.start(); - } + @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 + } - @Override - protected void doStop() { - this.messageListenerContainer.stop(); } } 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..807461f22f --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java @@ -0,0 +1,56 @@ +/* + * 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.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 {@code org.springframework.integration.message.EnhancedErrorMessage}. + * + * @author Gary Russell + * @since 4.3.10 + * + */ +public class AmqpMessageHeaderErrorMessageStrategy implements ErrorMessageStrategy { + + /** + * Header name/retry context variable for the raw received message. + */ + public static final String AMQP_MESSAGE = "amqp_message"; // move to AmqpHeaders.MESSAGE in 2.0 + + @SuppressWarnings("deprecation") + @Override + public ErrorMessage buildErrorMessage(Throwable throwable, AttributeAccessor context) { + Object inputMessage = context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY); + Map headers = Collections.singletonMap( + AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE, + context.getAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE)); + return inputMessage instanceof Message + ? new org.springframework.integration.message.EnhancedErrorMessage(throwable, headers, + (Message) inputMessage) + : new ErrorMessage(throwable, headers); + } + +} 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 a7557f9c88..0d2f4c0a5a 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,10 +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.Matchers.anyBoolean; import static org.mockito.Mockito.doAnswer; @@ -39,15 +45,20 @@ 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.rabbit.support.CorrelationData; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +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; @@ -55,8 +66,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; @@ -225,6 +238,121 @@ public class InboundEndpointTests { } + @Test + public void testAdapterConversionError() throws Exception { + Connection connection = mock(Connection.class); + doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean()); + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + when(connectionFactory.createConnection()).thenReturn(connection); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container); + QueueChannel outputChannel = new QueueChannel(); + adapter.setOutputChannel(outputChannel); + QueueChannel errorChannel = new QueueChannel(); + adapter.setErrorChannel(errorChannel); + adapter.setMessageConverter(new MessageConverter() { + + @Override + public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties) + throws MessageConversionException { + throw new MessageConversionException("intended"); + } + + @Override + public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException { + return null; + } + + }); + adapter.afterPropertiesSet(); + ((ChannelAwareMessageListener) container.getMessageListener()).onMessage(null, null); + assertNull(outputChannel.receive(0)); + assertNotNull(errorChannel.receive(0)); + } + + @Test + public void testGatewayConversionError() throws Exception { + Connection connection = mock(Connection.class); + doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean()); + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + when(connectionFactory.createConnection()).thenReturn(connection); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + AmqpInboundGateway adapter = new AmqpInboundGateway(container); + QueueChannel outputChannel = new QueueChannel(); + adapter.setRequestChannel(outputChannel); + QueueChannel errorChannel = new QueueChannel(); + adapter.setErrorChannel(errorChannel); + adapter.setMessageConverter(new MessageConverter() { + + @Override + public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties) + throws MessageConversionException { + throw new MessageConversionException("intended"); + } + + @Override + public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException { + return null; + } + + }); + adapter.afterPropertiesSet(); + ((ChannelAwareMessageListener) container.getMessageListener()).onMessage(null, null); + assertNull(outputChannel.receive(0)); + 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(AmqpMessageHeaderErrorMessageStrategy.AMQP_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(AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE, org.springframework.amqp.core.Message.class); + assertThat(amqpMessage, notNullValue()); + assertNull(errors.receive(0)); + } public static class Foo { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 2bd30b11f3..f2d22da640 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -32,6 +32,7 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; @@ -330,8 +331,9 @@ public class ChannelPublishingJmsMessageListener if (errorChannel == null) { throw e; } - errorChannel.send(this.gatewayDelegate.buildErrorMessage( - new MessagingException("Inbound conversion failed for: " + jmsMessage, e))); + this.gatewayDelegate.getMessagingTemplate().send(errorChannel, + this.gatewayDelegate.buildErrorMessage( + new MessagingException("Inbound conversion failed for: " + jmsMessage, e))); errors = true; } if (!errors) { @@ -510,10 +512,14 @@ public class ChannelPublishingJmsMessageListener return super.sendAndReceiveMessage(request); } - public ErrorMessage buildErrorMessage(Throwable throwable) { + protected ErrorMessage buildErrorMessage(Throwable throwable) { return super.buildErrorMessage(null, throwable); } + protected MessagingTemplate getMessagingTemplate() { + return this.messagingTemplate; + } + @Override public String getComponentType() { if (ChannelPublishingJmsMessageListener.this.expectReply) {