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

INT-4257: Introduce ErrorMessagePublisher

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

To make a target `ErrorMessage` customizable introduce
`ErrorMessagePublishingRecoveryCallback` and
`ErrorMessageStrategy` to inject

**Cherry-pick to 4.3.x**

Fix `ErrorMessagePublishingRecoveryCallback` generic type for
`RequestHandlerRetryAdvice` compatibility

Polishing

- Support publishing when no retry context is available
- Use a constant for the message context key

Add EnhancedErrorMessage

- contains input message at time of error message generation
- make the default RecovererErrorMessageStrategy public and extensible

Fix `RetryAdviceParserTests` for new `RecoveryCallback` architecture

Rename to `ErrorMessagePublisher`
Decouple ErrorMessagePublisher from Retry

Refactoring for better purpose reflection and JavaDocs

Use `ErrorMessageStrategy` in message producers.

Add JavaDocs to the `DefaultErrorMessageStrategy`

Deprecate EnhancedErrorMessage

Minor Polish for Subclass Use

- make setters final
- allow subclasses to supply the AttributeAccessor

Fix Tests
This commit is contained in:
Artem Bilan
2017-03-10 16:06:56 -05:00
committed by Gary Russell
parent 052aab0e47
commit f7adfc0628
12 changed files with 581 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -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);
}
}

35
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -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;

86
spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java
View File

@@ -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<Object>, BeanFactoryAware {
public class ErrorMessageSendingRecoverer extends ErrorMessagePublisher implements RecoveryCallback<Object> {
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;

11
spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.java
View File

@@ -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 <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
context.setAttribute("message", messageHolder.get());
context.setAttribute(ErrorMessageUtils.FAILED_MESSAGE_CONTEXT_KEY, messageHolder.get());
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
}
}

64
spring-integration-core/src/main/java/org/springframework/integration/message/EnhancedErrorMessage.java Normal file
View File

@@ -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<String, Object> headers) {
super(payload, headers);
this.originalMessage = originalMessage;
}
public Message<?> getOriginalMessage() {
return this.originalMessage;
}
}

47
spring-integration-core/src/main/java/org/springframework/integration/support/DefaultErrorMessageStrategy.java Normal file
View File

@@ -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);
}
}

206
spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessagePublisher.java Normal file
View File

@@ -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.
* <p>
* 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<MessageChannel> 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<MessageChannel> 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);
}
}
}

48
spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessageStrategy.java Normal file
View File

@@ -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.
* <p>
* 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.
* <p>
* 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);
}

72
spring-integration-core/src/main/java/org/springframework/integration/support/ErrorMessageUtils.java Normal file
View File

@@ -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();
}
}
}

6
spring-integration-core/src/test/java/org/springframework/integration/config/xml/RetryAdviceParserTests.java
View File

@@ -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));

4
spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java
View File

@@ -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<String>("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();

19
spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java
View File

@@ -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;