PublishSubscribeChannel now wraps a provided TaskExecutor with the ErrorHandlingTaskExecutor if necessary (INT-440). Also, added namespace support for the publish-subscribe-channel's 'error-handler' reference (INT-483).

org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java

@@ -16,15 +16,24 @@
 
 package org.springframework.integration.channel;
 
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.dispatcher.BroadcastingDispatcher;
+import org.springframework.integration.executor.ErrorHandlingTaskExecutor;
+import org.springframework.integration.util.ErrorHandler;
 
 /**
  * A channel that sends Messages to each of its subscribers. 
  * 
  * @author Mark Fisher
  */
-public class PublishSubscribeChannel extends AbstractSubscribableChannel<BroadcastingDispatcher> {
+public class PublishSubscribeChannel extends AbstractSubscribableChannel<BroadcastingDispatcher> implements BeanFactoryAware {
+
+	private volatile TaskExecutor taskExecutor;
+
+	private volatile ErrorHandler errorHandler;
+
 
 	/**
 	 * Create a PublishSubscribeChannel that will use a {@link TaskExecutor}
@@ -32,9 +41,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel<Broadca
 	 */
 	public PublishSubscribeChannel(TaskExecutor taskExecutor) {
 		super(new BroadcastingDispatcher());
-		if (taskExecutor != null) {
-			this.getDispatcher().setTaskExecutor(taskExecutor);
-		}
+		this.taskExecutor = taskExecutor;
 	}
 
 	public PublishSubscribeChannel() {
@@ -42,8 +49,24 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel<Broadca
 	}
 
 
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
 	public void setApplySequence(boolean applySequence) {
 		this.getDispatcher().setApplySequence(applySequence);
 	}
 
+	public void setBeanFactory(BeanFactory beanFactory) {
+		if (this.taskExecutor != null) {
+			if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
+				if (this.errorHandler == null) {
+					this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(beanFactory));
+				}
+				this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler);
+			}
+			this.getDispatcher().setTaskExecutor(this.taskExecutor);
+		}
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java

@@ -37,6 +37,7 @@ public class PublishSubscribeChannelParser extends AbstractChannelParser {
 		if (StringUtils.hasText(taskExecutorRef)) {
 			builder.addConstructorArgReference(taskExecutorRef);
 		}
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
 		return builder;
 	}

org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd

@@ -111,6 +111,7 @@
 						<xsd:element name="interceptors" type="channelInterceptorsType" minOccurs="0" maxOccurs="1"/>
 					</xsd:sequence>
 					<xsd:attribute name="task-executor" type="xsd:string"/>
+					<xsd:attribute name="error-handler" type="xsd:string"/>
 					<xsd:attribute name="apply-sequence" type="xsd:string"/>
 				</xsd:extension>
 			</xsd:complexContent>

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java

@@ -148,11 +148,11 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 						this.transactionManager, this.transactionDefinition);
 			}
 			this.poller = this.createPoller();
-			if (this.taskExecutor != null) {
+			if (this.taskExecutor != null && !(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
 				if (this.errorHandler == null) {
 					this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory()));
 				}
-				this.taskExecutor = new ErrorHandlingTaskExecutor(this.errorHandler, this.taskExecutor);
+				this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler);
 			}
 			this.initialized = true;
 		}

org.springframework.integration/src/main/java/org/springframework/integration/executor/ErrorHandlingTaskExecutor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2007 the original author or authors.
+ * Copyright 2002-2008 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.
@@ -13,57 +13,47 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.integration.executor;
 
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.util.ErrorHandler;
+import org.springframework.util.Assert;
 
 /**
+ * A {@link TaskExecutor} implementation that wraps an existing TaskExecutor
+ * instance in order to catch any exceptions. If an exception is thrown, it
+ * will be handled by the provided {@link ErrorHandler}.
  * 
  * @author Jonas Partner
- *
+ * @author Mark Fisher
  */
 public class ErrorHandlingTaskExecutor implements TaskExecutor {
 
+	private final TaskExecutor taskExecutor;
 
 	private final ErrorHandler errorHandler;
-	
-	private final TaskExecutor taskExecutor;
-	
-	/**
-	 * @param errorChannel
-	 */
-	public ErrorHandlingTaskExecutor(ErrorHandler errorHandler, TaskExecutor taskExecutor) {
-		this.errorHandler = errorHandler;
+
+
+	public ErrorHandlingTaskExecutor(TaskExecutor taskExecutor, ErrorHandler errorHandler) {
+		Assert.notNull(taskExecutor, "taskExecutor must not be null");
+		Assert.notNull(errorHandler, "errorHandler must not be null");
 		this.taskExecutor = taskExecutor;
+		this.errorHandler = errorHandler;
 	}
-	
-	public void execute(Runnable task) {
-		taskExecutor.execute(new ErrorHandlingRunnableWrapper(task,errorHandler));
-	}
-	
-	private static class ErrorHandlingRunnableWrapper implements Runnable {
 
-		private final ErrorHandler errorHandler;
 
-		private final Runnable runnableTarget;
-
-		public ErrorHandlingRunnableWrapper(Runnable runnableTarget,ErrorHandler errorHandler) {
-			this.runnableTarget = runnableTarget;
-			this.errorHandler = errorHandler;
-		}
-
-		public void run() {
-			try {
-				runnableTarget.run();
+	public void execute(final Runnable task) {
+		this.taskExecutor.execute(new Runnable() {
+			public void run() {
+				try {
+					task.run();
+				}
+				catch (Throwable t) {
+					errorHandler.handle(t);
+				}
 			}
-			catch (Throwable t) {
-				errorHandler.handle(t);
-			}
-		}
-
+		});
 	}
-	
-	
-	
+
 }

org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java

@@ -18,6 +18,7 @@ package org.springframework.integration.channel.config;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 import org.junit.Test;
@@ -26,6 +27,7 @@ import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.FatalBeanException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -33,6 +35,7 @@ import org.springframework.integration.config.TestChannelInterceptor;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessageChannel;
 import org.springframework.integration.core.MessagePriority;
+import org.springframework.integration.executor.ErrorHandlingTaskExecutor;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageDeliveryException;
@@ -85,8 +88,12 @@ public class ChannelParserTests {
 		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
 		accessor = new DirectFieldAccessor(accessor.getPropertyValue("dispatcher"));
 		Object taskExecutorProperty = accessor.getPropertyValue("taskExecutor");
+		assertNotNull(taskExecutorProperty);
+		assertEquals(ErrorHandlingTaskExecutor.class, taskExecutorProperty.getClass());		
+		DirectFieldAccessor executorAccessor = new DirectFieldAccessor(taskExecutorProperty);
+		TaskExecutor innerExecutor = (TaskExecutor) executorAccessor.getPropertyValue("taskExecutor");
 		Object taskExecutorBean = context.getBean("taskExecutor");
-		assertEquals(taskExecutorBean, taskExecutorProperty);
+		assertEquals(taskExecutorBean, innerExecutor);
 	}
 
 	@Test

org.springframework.integration/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java

@@ -29,6 +29,8 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.dispatcher.BroadcastingDispatcher;
+import org.springframework.integration.executor.ErrorHandlingTaskExecutor;
+import org.springframework.integration.util.ErrorHandler;
 
 /**
  * @author Mark Fisher
@@ -73,7 +75,22 @@ public class PublishSubscribeChannelParserTests {
 		DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
 		TaskExecutor executor = (TaskExecutor) dispatcherAccessor.getPropertyValue("taskExecutor");
 		assertNotNull(executor);
-		assertEquals(context.getBean("pool"), executor);
+		assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
+		DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
+		TaskExecutor innerExecutor = (TaskExecutor) executorAccessor.getPropertyValue("taskExecutor");
+		assertEquals(context.getBean("pool"), innerExecutor);
+	}
+
+	@Test
+	public void channelWithErrorHandler() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"publishSubscribeChannelParserTests.xml", this.getClass());
+		PublishSubscribeChannel channel = (PublishSubscribeChannel)
+				context.getBean("channelWithErrorHandler");
+		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
+		ErrorHandler errorHandler = (ErrorHandler) accessor.getPropertyValue("errorHandler");
+		assertNotNull(errorHandler);
+		assertEquals(context.getBean("testErrorHandler"), errorHandler);
 	}
 
 }

org.springframework.integration/src/test/java/org/springframework/integration/config/publishSubscribeChannelParserTests.xml

@@ -13,6 +13,10 @@
 
 	<publish-subscribe-channel id="channelWithTaskExecutor" task-executor="pool"/>
 
+	<publish-subscribe-channel id="channelWithErrorHandler" error-handler="testErrorHandler"/>
+
 	<thread-pool-task-executor id="pool"/>
 
+	<beans:bean id="testErrorHandler" class="org.springframework.integration.config.TestErrorHandler"/>
+
 </beans:beans>