MessageBus now properly registers error-channel set by 'setErrorChannel' (INT-164). DefaultChannelRegistry no longer has any awareness of an "error channel".

spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java

@@ -90,6 +90,8 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 
 	private volatile boolean initialized;
 
+	private volatile boolean initializing;
+
 	private volatile boolean starting;
 
 	private volatile boolean running;
@@ -173,31 +175,32 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 
 	public void initialize() {
 		synchronized (this.lifecycleMonitor) {
-			if (this.initialized) {
+			if (this.initialized || this.initializing) {
 				return;
 			}
-			if (this.getErrorChannel() == null) {
-				this.setErrorChannel(new DefaultErrorChannel());
-			}
+			this.initializing = true;
 			if (this.defaultConcurrencyPolicy == null) {
 				this.defaultConcurrencyPolicy = new ConcurrencyPolicy();
 			}
 			if (this.executor == null) {
 				this.executor = new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE);
 			}
-			SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(this.executor);
-			scheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
-			this.taskScheduler = scheduler;
+			this.taskScheduler = new SimpleMessagingTaskScheduler(this.executor);
+			if (this.getErrorChannel() == null) {
+				this.setErrorChannel(new DefaultErrorChannel());
+			}
+			this.taskScheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
 			this.initialized = true;
+			this.initializing = false;
 		}
 	}
 
 	public MessageChannel getErrorChannel() {
-		return this.channelRegistry.lookupChannel(ERROR_CHANNEL_NAME);
+		return this.lookupChannel(ERROR_CHANNEL_NAME);
 	}
 
 	public void setErrorChannel(MessageChannel errorChannel) {
-		this.channelRegistry.registerChannel(ERROR_CHANNEL_NAME, errorChannel);
+		this.registerChannel(ERROR_CHANNEL_NAME, errorChannel);
 	}
 
 	public MessageChannel lookupChannel(String channelName) {

spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java

@@ -30,16 +30,6 @@ public class DefaultChannelRegistry implements ChannelRegistry {
 
 	private final Map<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
 
-	private volatile MessageChannel errorChannel;
-
-
-	public void setErrorChannel(MessageChannel errorChannel) {
-		this.errorChannel = errorChannel;
-	}
-
-	public MessageChannel getErrorChannel() {
-		return this.errorChannel;
-	}
 
 	public MessageChannel lookupChannel(String channelName) {
 		return this.channels.get(channelName);

spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java

@@ -19,6 +19,7 @@ package org.springframework.integration.scheduling;
 import java.util.concurrent.ScheduledFuture;
 
 import org.springframework.context.Lifecycle;
+import org.springframework.integration.util.ErrorHandler;
 import org.springframework.scheduling.SchedulingTaskExecutor;
 
 /**
@@ -30,4 +31,6 @@ public interface MessagingTaskScheduler extends SchedulingTaskExecutor, Lifecycl
 
 	ScheduledFuture<?> schedule(Runnable task);
 
+	void setErrorHandler(ErrorHandler errorHandler);
+
 }

spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java

@@ -233,6 +233,33 @@ public class MessageBusTests {
 		assertTrue(exceptionThrown);
 	}
 
+	@Test
+	public void testErrorChannelRegistration() {
+		MessageChannel errorChannel = new SimpleChannel();
+		MessageBus bus = new MessageBus();
+		bus.setErrorChannel(errorChannel);
+		assertEquals(errorChannel, bus.getErrorChannel());
+	}
+
+	@Test
+	public void testHandlerSubscribedToErrorChannel() throws InterruptedException {
+		MessageChannel errorChannel = new SimpleChannel();
+		MessageBus bus = new MessageBus();
+		bus.setErrorChannel(errorChannel);
+		final CountDownLatch latch = new CountDownLatch(1);
+		MessageHandler handler = new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				latch.countDown();
+				return null;
+			}
+		};
+		bus.registerHandler("testHandler", handler, new Subscription(MessageBus.ERROR_CHANNEL_NAME));
+		bus.start();
+		errorChannel.send(new ErrorMessage(new RuntimeException("test-exception")));
+		latch.await(1000, TimeUnit.MILLISECONDS);
+		assertEquals("handler should have received error message", 0, latch.getCount());
+	}
+
 
 	private static class FailingSource implements PollableSource<Object> {
 

spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java

@@ -73,20 +73,4 @@ public class DefaultChannelRegistryTests {
 		assertNull(unregisteredChannel);
 	}
 
-	@Test
-	public void testNoErrorChannelByDefault() {
-		DefaultChannelRegistry registry = new DefaultChannelRegistry();
-		assertNull(registry.getErrorChannel());
-	}
-
-	@Test
-	public void testRegisteredErrorChannel() {
-		SimpleChannel errorChannel = new SimpleChannel();
-		DefaultChannelRegistry registry = new DefaultChannelRegistry();
-		registry.setErrorChannel(errorChannel);
-		MessageChannel result = registry.getErrorChannel();
-		assertNotNull(result);
-		assertSame(errorChannel, result);
-	}
-
 }