From ae698b61fdfa4fed08ddb25f4d8848ad79241a2f Mon Sep 17 00:00:00 2001
From: Mark Fisher <markfisher@vmware.com>
Date: Thu, 16 Oct 2008 02:36:25 +0000
Subject: [PATCH] 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.

---
 .../integration/bus/DefaultMessageBus.java    |  8 ++-
 .../MessagePublishingErrorHandler.java        | 50 +++++++++++++------
 2 files changed, 42 insertions(+), 16 deletions(-)

diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
index 8bd867654b..9352a7a474 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
@@ -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();
 		}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
index 8dc2872899..2e4ac2e097 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
@@ -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