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

MessagePublishingErrorHandler now retrieves the 'errorChannel' header from the 'failedMessage' of a MessagingException (if available). It resolves it if necessary (if it is a channel name instead of MessageChannel instance) by delegating to a ChannelResolver.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-16 02:36:25 +00:00
parent 2f009b607b
commit ae698b61fd
2 changed files with 42 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
View File

@@ -34,6 +34,8 @@ import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
@@ -197,8 +199,10 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
synchronized (this.lifecycleMonitor) {
this.activateEndpoints();
if (this.taskScheduler instanceof SimpleTaskScheduler) {
((SimpleTaskScheduler) this.taskScheduler).setErrorHandler(
new MessagePublishingErrorHandler(this.lookupChannel(ERROR_CHANNEL_BEAN_NAME)));
ChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(channelResolver);
errorHandler.setDefaultErrorChannel(this.lookupChannel(ERROR_CHANNEL_BEAN_NAME));
((SimpleTaskScheduler) this.taskScheduler).setErrorHandler(errorHandler);
}
this.taskScheduler.start();
}

50
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
View File

@@ -19,10 +19,12 @@ package org.springframework.integration.channel;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.util.Assert;
/**
* {@link ErrorHandler} implementation that sends an {@link ErrorMessage} to a
@@ -34,40 +36,60 @@ public class MessagePublishingErrorHandler implements ErrorHandler {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile MessageChannel errorChannel;
private final ChannelResolver channelResolver;
private final long sendTimeout = 1000;
private volatile MessageChannel defaultErrorChannel;
private volatile long sendTimeout = 1000;
public MessagePublishingErrorHandler() {
}
public MessagePublishingErrorHandler(MessageChannel errorChannel) {
this.errorChannel = errorChannel;
public MessagePublishingErrorHandler(ChannelResolver channelResolver) {
Assert.notNull(channelResolver, "channelResolver must not be null");
this.channelResolver = channelResolver;
}
public void setErrorChannel(MessageChannel errorChannel) {
this.errorChannel = errorChannel;
public void setDefaultErrorChannel(MessageChannel defaultErrorChannel) {
this.defaultErrorChannel = defaultErrorChannel;
}
public void setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
}
public final void handle(Throwable t) {
Message<?> failedMessage = null;
if (logger.isWarnEnabled()) {
if (t instanceof MessagingException) {
logger.warn("failure occurred in messaging task with message: "
+ ((MessagingException) t).getFailedMessage(), t);
failedMessage = ((MessagingException) t).getFailedMessage();
logger.warn("failure occurred in messaging task with message: " + failedMessage, t);
}
else {
logger.warn("failure occurred in messaging task", t);
}
}
if (this.errorChannel != null) {
MessageChannel errorChannel = null;
if (failedMessage != null) {
Object errorChannelHeader = failedMessage.getHeaders().getErrorChannel();
if (errorChannelHeader != null) {
if (errorChannelHeader instanceof MessageChannel) {
errorChannel = (MessageChannel) errorChannelHeader;
}
else if (errorChannelHeader instanceof String) {
errorChannel = this.channelResolver.resolveChannelName((String) errorChannelHeader);
}
}
}
if (errorChannel == null) {
errorChannel = this.defaultErrorChannel;
}
if (errorChannel != null) {
try {
if (this.sendTimeout >= 0) {
this.errorChannel.send(new ErrorMessage(t), this.sendTimeout);
errorChannel.send(new ErrorMessage(t), this.sendTimeout);
}
else {
this.errorChannel.send(new ErrorMessage(t));
errorChannel.send(new ErrorMessage(t));
}
}
catch (Throwable ignore) { // message will be logged only