diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 183c6a4c6b..130acd8c73 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java index d531a3caed..c9d6605329 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java @@ -30,16 +30,6 @@ public class DefaultChannelRegistry implements ChannelRegistry { private final Map channels = new ConcurrentHashMap(); - 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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java index e611a5c70e..7ac632a370 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java +++ b/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); + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 7e94846265..3afebedb0b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/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 { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java index e66c5c9767..581a8a1227 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java +++ b/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); - } - }