diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 74e393db9a..c467ea1b1a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.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. @@ -17,14 +17,17 @@ package org.springframework.integration.endpoint; import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.core.AttributeAccessor; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.history.MessageHistory; +import org.springframework.integration.support.DefaultErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.integration.support.management.TrackableComponent; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -41,6 +44,8 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + private volatile MessageChannel outputChannel; private volatile String outputChannelName; @@ -127,6 +132,17 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements this.shouldTrack = shouldTrack; } + /** + * Set an {@link ErrorMessageStrategy} to use to build an error message when a exception occurs. + * Default is the {@link DefaultErrorMessageStrategy}. + * @param errorMessageStrategy the {@link ErrorMessageStrategy}. + * @since 4.3.10 + */ + public final void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) { + Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' cannot be null"); + this.errorMessageStrategy = errorMessageStrategy; + } + protected MessagingTemplate getMessagingTemplate() { return this.messagingTemplate; } @@ -173,12 +189,25 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements catch (RuntimeException e) { MessageChannel errorChannel = getErrorChannel(); if (errorChannel != null) { - this.messagingTemplate.send(errorChannel, new ErrorMessage(e)); + this.messagingTemplate.send(errorChannel, this.errorMessageStrategy.buildErrorMessage(e, + getErrorMessageAttributes(message))); } - else { + else { throw e; } } } + /** + * Populate an {@link AttributeAccessor} to be used when building an error message + * with the {@link #setErrorMessageStrategy(ErrorMessageStrategy) + * errorMessageStrategy}. + * @param message the message. + * @return the attributes. + * @since 4.3.10 + */ + protected AttributeAccessor getErrorMessageAttributes(Message message) { + return ErrorMessageUtils.getAttributeAccessor(message, null); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 31dac5d3a3..1a3ab775cf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -18,6 +18,7 @@ package org.springframework.integration.gateway; import java.util.concurrent.atomic.AtomicLong; +import org.springframework.core.AttributeAccessor; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractEndpoint; @@ -27,7 +28,10 @@ import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.history.HistoryWritingMessagePostProcessor; import org.springframework.integration.mapping.InboundMessageMapper; import org.springframework.integration.mapping.OutboundMessageMapper; +import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.DefaultMessageBuilderFactory; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.integration.support.management.IntegrationManagedResource; @@ -70,6 +74,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint private final AtomicLong messageCount = new AtomicLong(); + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + private volatile MessageChannel requestChannel; private volatile String requestChannelName; @@ -289,6 +295,17 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return this.countsEnabled; } + /** + * Set an {@link ErrorMessageStrategy} to use to build an error message when a exception occurs. + * Default is the {@link DefaultErrorMessageStrategy}. + * @param errorMessageStrategy the {@link ErrorMessageStrategy}. + * @since 4.3.10 + */ + public final void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) { + Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' cannot be null"); + this.errorMessageStrategy = errorMessageStrategy; + } + @Override protected void onInit() throws Exception { Assert.state(!(this.requestChannelName != null && this.requestChannel != null), @@ -423,6 +440,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } Object reply = null; Throwable error = null; + Message requestMessage = null; try { if (this.countsEnabled) { this.messageCount.incrementAndGet(); @@ -435,7 +453,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } } else { - Message requestMessage = (object instanceof Message) + requestMessage = (object instanceof Message) ? (Message) object : this.requestMapper.toMessage(object); requestMessage = this.historyWritingPostProcessor.postProcessMessage(requestMessage); reply = this.messagingTemplate.sendAndReceive(requestChannel, requestMessage); @@ -462,7 +480,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint if (error != null) { MessageChannel errorChannel = getErrorChannel(); if (errorChannel != null) { - Message errorMessage = new ErrorMessage(error); + Message errorMessage = this.errorMessageStrategy.buildErrorMessage(error, + getErrorMessageAttributes(requestMessage)); Message errorFlowReply = null; try { errorFlowReply = this.messagingTemplate.sendAndReceive(errorChannel, errorMessage); @@ -499,6 +518,18 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return reply; } + /** + * Populate an {@link AttributeAccessor} to be used when building an error message + * with the {@link #setErrorMessageStrategy(ErrorMessageStrategy) + * errorMessageStrategy}. + * @param message the message. + * @return the attributes. + * @since 4.3.10 + */ + protected AttributeAccessor getErrorMessageAttributes(Message message) { + return ErrorMessageUtils.getAttributeAccessor(message, null); + } + private void rethrow(Throwable t, String description) { if (t instanceof RuntimeException) { throw (RuntimeException) t; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java index 6c03b20502..dcfa4a2b7b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.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. @@ -16,72 +16,76 @@ package org.springframework.integration.handler.advice; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.integration.core.MessagingTemplate; +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.support.DefaultErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessagePublisher; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.support.ErrorMessage; import org.springframework.retry.RecoveryCallback; import org.springframework.retry.RetryContext; -import org.springframework.util.Assert; /** - * RecoveryCallback that sends the final throwable as an ErrorMessage after + * A {@link RecoveryCallback} that sends the final throwable as an + * {@link org.springframework.messaging.support.ErrorMessage} after * retry exhaustion. * * @author Gary Russell * @author Artem Bilan + * * @since 2.2 * */ -public class ErrorMessageSendingRecoverer implements RecoveryCallback, BeanFactoryAware { +public class ErrorMessageSendingRecoverer extends ErrorMessagePublisher implements RecoveryCallback { - private final static Log logger = LogFactory.getLog(ErrorMessageSendingRecoverer.class); - - private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + /** + * Construct instance with the default {@code errorChannel} + * to publish recovery error message. + * @since 4.3.10 + */ + public ErrorMessageSendingRecoverer() { + this(null); + } + /** + * Construct instance based on the provided message channel. + * The {@link DefaultErrorMessageStrategy} is used for building error message to publish. + * @param channel the message channel to publish error messages on recovery action. + */ public ErrorMessageSendingRecoverer(MessageChannel channel) { - Assert.notNull(channel, "channel cannot be null"); - this.messagingTemplate.setDefaultDestination(channel); + this(channel, new DefaultErrorMessageStrategy()); } - public void setSendTimeout(long sendTimeout) { - this.messagingTemplate.setSendTimeout(sendTimeout); - } - - @Override - public void setBeanFactory(BeanFactory beanFactory) throws BeansException { - this.messagingTemplate.setBeanFactory(beanFactory); + /** + * Construct instance based on the provided message channel and {@link ErrorMessageStrategy}. + * @param channel the message channel to publish error messages on recovery action. + * @param errorMessageStrategy the {@link ErrorMessageStrategy} + * to build error message for publishing. + * @since 4.3.10 + */ + public ErrorMessageSendingRecoverer(MessageChannel channel, ErrorMessageStrategy errorMessageStrategy) { + setChannel(channel); + setErrorMessageStrategy(errorMessageStrategy); } @Override public Object recover(RetryContext context) throws Exception { - Throwable lastThrowable = context.getLastThrowable(); - if (lastThrowable == null) { - lastThrowable = new RetryExceptionNotAvailableException( - (Message) context.getAttribute("message"), - "No retry exception available; " + - "this can occur, for example, if the RetryPolicy allowed zero attempts to execute the handler; " + - "RetryContext: " + context.toString()); - } - else if (!(lastThrowable instanceof MessagingException)) { - lastThrowable = new MessagingException((Message) context.getAttribute("message"), - lastThrowable.getMessage(), lastThrowable); - } - if (logger.isDebugEnabled()) { - String supplement = ":failedMessage:" + ((MessagingException) lastThrowable).getFailedMessage(); - logger.debug("Sending ErrorMessage " + supplement, lastThrowable); - } - this.messagingTemplate.send(new ErrorMessage(lastThrowable)); + publish(context.getLastThrowable(), context); return null; } + @Override + protected Throwable payloadWhenNull(AttributeAccessor context) { + return new RetryExceptionNotAvailableException( + (Message) context.getAttribute(ErrorMessageUtils.FAILED_MESSAGE_CONTEXT_KEY), + "No retry exception available; " + + "this can occur, for example, if the RetryPolicy allowed zero attempts " + + "to execute the handler; " + + "RetryContext: " + context.toString()); + } + public static class RetryExceptionNotAvailableException extends MessagingException { private static final long serialVersionUID = 1L; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.java index 1656362d95..feceb35a69 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.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. @@ -16,6 +16,7 @@ package org.springframework.integration.handler.advice; +import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.retry.RecoveryCallback; @@ -111,16 +112,18 @@ public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice @Override public boolean open(RetryContext context, RetryCallback callback) { - context.setAttribute("message", messageHolder.get()); + context.setAttribute(ErrorMessageUtils.FAILED_MESSAGE_CONTEXT_KEY, messageHolder.get()); return true; } @Override - public void close(RetryContext context, RetryCallback callback, Throwable throwable) { + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { } @Override - public void onError(RetryContext context, RetryCallback callback, Throwable throwable) { + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/EnhancedErrorMessage.java b/spring-integration-core/src/main/java/org/springframework/integration/message/EnhancedErrorMessage.java new file mode 100644 index 0000000000..f1b5e399b2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/EnhancedErrorMessage.java @@ -0,0 +1,64 @@ +/* + * 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.message; + +import java.util.Map; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.ErrorMessage; + +/** + * An error message that is enhanced with a message that is available at the + * stack frame where the error message is generated. Typically this will be a + * message that begins a subflow, whereas if the {@link Throwable} payload is + * a {@link org.springframework.messaging.MessagingException}, its failedMessage + * property will contain the message at the point where the exception occurred. + * + * @author Gary Russell + * + * @since 4.3.10 + * + * @deprecated since 4.3.10 in favor of direct {@link ErrorMessage} usage since 5.0. + */ +@Deprecated +public class EnhancedErrorMessage extends ErrorMessage { + + private static final long serialVersionUID = 5857673472822628678L; + + private final Message originalMessage; + + public EnhancedErrorMessage(Message originalMessage, Throwable payload) { + super(payload); + this.originalMessage = originalMessage; + } + + public EnhancedErrorMessage(Message originalMessage, Throwable payload, MessageHeaders headers) { + super(payload, headers); + this.originalMessage = originalMessage; + } + + public EnhancedErrorMessage(Message originalMessage, Throwable payload, Map headers) { + super(payload, headers); + this.originalMessage = originalMessage; + } + + public Message getOriginalMessage() { + return this.originalMessage; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/DefaultErrorMessageStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/support/DefaultErrorMessageStrategy.java new file mode 100644 index 0000000000..da5853e64d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/DefaultErrorMessageStrategy.java @@ -0,0 +1,47 @@ +/* + * 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.support; + +import org.springframework.core.AttributeAccessor; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.ErrorMessage; + +/** + * A simple {@link ErrorMessageStrategy} implementations which produces + * a error message with original message if the {@link AttributeAccessor} has + * {@link ErrorMessageUtils#INPUT_MESSAGE_CONTEXT_KEY} attribute. + * Otherwise plain {@link ErrorMessage} with the {@code throwable} as {@code payload}. + * + * @author Gary Russell + * @author Artem Bilan + * + * @since 4.3.10 + * + * @see ErrorMessageUtils + */ +public class DefaultErrorMessageStrategy implements ErrorMessageStrategy { + + @Override + @SuppressWarnings("deprecation") + public ErrorMessage buildErrorMessage(Throwable throwable, AttributeAccessor attributes) { + Object inputMessage = attributes.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY); + return inputMessage instanceof Message + ? new org.springframework.integration.message.EnhancedErrorMessage((Message) inputMessage, throwable) + : new ErrorMessage(throwable); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessagePublisher.java b/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessagePublisher.java new file mode 100644 index 0000000000..44d1b2c267 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessagePublisher.java @@ -0,0 +1,206 @@ +/* + * 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.support; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.support.channel.BeanFactoryChannelResolver; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.core.DestinationResolver; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.util.Assert; + +/** + * The component which can be used as general purpose of errors publishing. + * Can be called or extended in any error handling or retry scenarios. + *

+ * An {@link ErrorMessageStrategy} can be used to provide customization for the target + * {@link ErrorMessage} based on the {@link AttributeAccessor} (or the message and/or + * throwable when using the other {@code publish()} methods). + * + * @author Artem Bilan + * @author Gary Russell + * + * @since 4.3.10 + */ +public class ErrorMessagePublisher implements BeanFactoryAware { + + protected final Log logger = LogFactory.getLog(getClass()); + + protected final MessagingTemplate messagingTemplate = new MessagingTemplate(); + + private DestinationResolver channelResolver; + + private MessageChannel channel; + + private String channelName; + + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + + public final void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) { + Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' must not be null"); + this.errorMessageStrategy = errorMessageStrategy; + } + + public final void setChannel(MessageChannel channel) { + this.channel = channel; + } + + public void setChannelName(String channelName) { + this.channelName = channelName; + } + + public ErrorMessageStrategy getErrorMessageStrategy() { + return this.errorMessageStrategy; + } + + public MessageChannel getChannel() { + populateChannel(); + return this.channel; + } + + public void setSendTimeout(long sendTimeout) { + this.messagingTemplate.setSendTimeout(sendTimeout); + } + + public void setChannelResolver(DestinationResolver channelResolver) { + this.channelResolver = channelResolver; + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) { + Assert.notNull(beanFactory, "beanFactory must not be null"); + if (this.channelResolver == null) { + this.channelResolver = new BeanFactoryChannelResolver(beanFactory); + } + } + + /** + * Publish an error message for the supplied exception. + * @param exception the exception. + */ + public void publish(MessagingException exception) { + publish(null, exception.getFailedMessage(), exception); + } + + /** + * Publish an error message for the supplied message and throwable. If the throwable + * is already a {@link MessagingException} containing the message in its + * {@code failedMessage} property, use {@link #publish(MessagingException)} instead. + * @param failedMessage the message. + * @param throwable the throwable. + */ + public void publish(Message failedMessage, Throwable throwable) { + publish(null, failedMessage, throwable); + } + + /** + * Publish an error message for the supplied exception. + * @param inputMessage the message that started the subflow. + * @param exception the exception. + */ + public void publish(Message inputMessage, MessagingException exception) { + publish(inputMessage, exception.getFailedMessage(), exception); + } + + /** + * Publish an error message for the supplied message and throwable. If the throwable + * is already a {@link MessagingException} containing the message in its + * {@code failedMessage} property, use {@link #publish(MessagingException)} instead. + * @param inputMessage the message that started the subflow. + * @param failedMessage the message. + * @param throwable the throwable. + */ + public void publish(Message inputMessage, Message failedMessage, Throwable throwable) { + publish(throwable, ErrorMessageUtils.getAttributeAccessor(inputMessage, failedMessage)); + } + + /** + * Publish an error message for the supplied throwable and context. + * The {@link #errorMessageStrategy} is used to build a {@link ErrorMessage} + * to publish. + * @param throwable the throwable. May be null. + * @param context the context for {@link ErrorMessage} properties. + */ + public void publish(Throwable throwable, AttributeAccessor context) { + populateChannel(); + Throwable payload = determinePayload(throwable, context); + ErrorMessage errorMessage = this.errorMessageStrategy.buildErrorMessage(payload, context); + if (this.logger.isDebugEnabled() && payload instanceof MessagingException) { + MessagingException exception = (MessagingException) errorMessage.getPayload(); + this.logger.debug("Sending ErrorMessage: failedMessage: " + exception.getFailedMessage(), exception); + } + this.messagingTemplate.send(errorMessage); + } + + /** + * Build a {@code Throwable payload} for future {@link ErrorMessage}. + * @param throwable the error to determine an {@link ErrorMessage} payload. Can be null. + * @param context the context for error. + * @return the throwable for the {@link ErrorMessage} payload + * @see ErrorMessageUtils + */ + protected Throwable determinePayload(Throwable throwable, AttributeAccessor context) { + Throwable lastThrowable = throwable; + if (lastThrowable == null) { + lastThrowable = payloadWhenNull(context); + } + else if (!(lastThrowable instanceof MessagingException)) { + lastThrowable = new MessagingException( + (Message) context.getAttribute(ErrorMessageUtils.FAILED_MESSAGE_CONTEXT_KEY), + lastThrowable.getMessage(), lastThrowable); + } + return lastThrowable; + } + + /** + * Build a {@code Throwable payload} based on the provided context + * for future {@link ErrorMessage} when there is original {@code Throwable}. + * @param context the {@link AttributeAccessor} to use for exception properties. + * @return the {@code Throwable} for an {@link ErrorMessage} payload. + * @see ErrorMessageUtils + */ + protected Throwable payloadWhenNull(AttributeAccessor context) { + return new MessagingException((Message) context.getAttribute(ErrorMessageUtils.FAILED_MESSAGE_CONTEXT_KEY), + "No root cause exception available"); + } + + private void populateChannel() { + if (this.messagingTemplate.getDefaultDestination() == null) { + if (this.channel == null) { + String recoveryChannelName = this.channelName; + if (recoveryChannelName == null) { + recoveryChannelName = IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME; + } + if (this.channelResolver != null) { + this.channel = this.channelResolver.resolveDestination(recoveryChannelName); + } + } + + this.messagingTemplate.setDefaultChannel(this.channel); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessageStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessageStrategy.java new file mode 100644 index 0000000000..0f540a997e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessageStrategy.java @@ -0,0 +1,48 @@ +/* + * 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.support; + +import org.springframework.core.AttributeAccessor; +import org.springframework.messaging.support.ErrorMessage; + +/** + * A strategy to build an {@link ErrorMessage} based on the provided + * {@link Throwable} and {@link AttributeAccessor} as a context. + *

+ * The {@code Throwable payload} is typically {@link org.springframework.messaging.MessagingException} + * which {@code failedMessage} property can be used to determine a cause of the error. + *

+ * This strategy can be used for the + * {@link org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer} + * for {@link org.springframework.retry.RetryContext} access. + * + * @author Artem Bilan + * @author Gary Russell + * + * @since 4.3.10 + */ +public interface ErrorMessageStrategy { + + /** + * Build the error message. + * @param payload the payload. + * @param attributes the attributes. + * @return the ErrorMessage. + */ + ErrorMessage buildErrorMessage(Throwable payload, AttributeAccessor attributes); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessageUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessageUtils.java new file mode 100644 index 0000000000..f11012b94a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessageUtils.java @@ -0,0 +1,72 @@ +/* + * 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.support; + +import org.springframework.core.AttributeAccessor; +import org.springframework.core.AttributeAccessorSupport; +import org.springframework.messaging.Message; + +/** + * Utilities for building error messages. + * + * @author Gary Russell + * + * @since 4.3.10 + * + */ +public final class ErrorMessageUtils { + + /** + * The context key for the message object. + */ + public static final String FAILED_MESSAGE_CONTEXT_KEY = "message"; + /** + * The context key for the message object. + */ + public static final String INPUT_MESSAGE_CONTEXT_KEY = "inputMessage"; + + private ErrorMessageUtils() { + super(); + } + + /** + * Return a {@link AttributeAccessor} for the provided arguments. + * @param inputMessage the input message. + * @param failedMessage the failed message. + * @return the context. + */ + public static AttributeAccessor getAttributeAccessor(Message inputMessage, Message failedMessage) { + AttributeAccessorSupport attributes = new ErrorMessageAttributes(); + if (inputMessage != null) { + attributes.setAttribute(INPUT_MESSAGE_CONTEXT_KEY, inputMessage); + } + if (failedMessage != null) { + attributes.setAttribute(FAILED_MESSAGE_CONTEXT_KEY, failedMessage); + } + return attributes; + } + + @SuppressWarnings("serial") + private static class ErrorMessageAttributes extends AttributeAccessorSupport { + + ErrorMessageAttributes() { + super(); + } + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/RetryAdviceParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/RetryAdviceParserTests.java index a43fe8b325..34432fa293 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/RetryAdviceParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/RetryAdviceParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -37,6 +37,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gary Russell + * @author Artem Bilan + * * @since 4.0 * */ @@ -99,7 +101,7 @@ public class RetryAdviceParserTests { assertNull(TestUtils.getPropertyValue(a1, "recoveryCallback")); assertNotNull(TestUtils.getPropertyValue(a7, "recoveryCallback")); - assertSame(this.foo, TestUtils.getPropertyValue(a7, "recoveryCallback.messagingTemplate.defaultDestination")); + assertSame(this.foo, TestUtils.getPropertyValue(a7, "recoveryCallback.channel")); assertEquals(4567L, TestUtils.getPropertyValue(a7, "recoveryCallback.messagingTemplate.sendTimeout")); assertSame(this.a1, TestUtils.getPropertyValue(this.handler1, "adviceChain", List.class).get(0)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java index 20143f9f93..0db9388b6d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java @@ -16,9 +16,11 @@ package org.springframework.integration.endpoint; +import static org.hamcrest.CoreMatchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; @@ -115,7 +117,7 @@ public class MessageProducerSupportTests { mps.start(); Message message = new GenericMessage("hello"); mps.sendMessage(message); - assertEquals(ErrorMessage.class, errorService.lastMessage.getClass()); + assertThat(errorService.lastMessage, instanceOf(ErrorMessage.class)); ErrorMessage errorMessage = (ErrorMessage) errorService.lastMessage; assertEquals(MessageDeliveryException.class, errorMessage.getPayload().getClass()); MessageDeliveryException exception = (MessageDeliveryException) errorMessage.getPayload(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java index 3c8d2fafe8..3075c8b637 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java @@ -16,6 +16,8 @@ package org.springframework.integration.handler.advice; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -1111,6 +1113,23 @@ public class AdvisedMessageHandlerTests { } + @Test + @SuppressWarnings("deprecation") + public void enhancedRecoverer() throws Exception { + QueueChannel channel = new QueueChannel(); + ErrorMessageSendingRecoverer recoverer = new ErrorMessageSendingRecoverer(channel); + recoverer.publish(new GenericMessage<>("foo"), new GenericMessage<>("bar"), new RuntimeException("baz")); + Message error = channel.receive(0); + assertThat(error, instanceOf(org.springframework.integration.message.EnhancedErrorMessage.class)); + assertThat(error.getPayload(), instanceOf(MessagingException.class)); + MessagingException payload = (MessagingException) error.getPayload(); + assertThat(payload.getCause(), instanceOf(RuntimeException.class)); + assertThat(payload.getCause().getMessage(), equalTo("baz")); + assertThat(payload.getFailedMessage().getPayload(), equalTo("bar")); + assertThat(((org.springframework.integration.message.EnhancedErrorMessage) error).getOriginalMessage() + .getPayload(), equalTo("foo")); + } + private interface Bar { Object handleRequestMessage(Message message) throws Throwable;