diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java new file mode 100644 index 0000000000..44106f23a8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java @@ -0,0 +1,65 @@ +/* + * Copyright 2002-2007 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.bus; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter; +import org.springframework.integration.message.Message; + +/** + * The default error channel implementation used by the {@link MessageBus}. + * + * @author Mark Fisher + */ +public class DefaultErrorChannel extends SimpleChannel { + + private final Log logger = LogFactory.getLog(this.getClass()); + + + public DefaultErrorChannel() { + this.addInterceptor(new ErrorLoggingInterceptor()); + } + + + private class ErrorLoggingInterceptor extends ChannelInterceptorAdapter { + + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + Object payload = message.getPayload(); + if (!sent) { + if (logger.isWarnEnabled()) { + logger.warn("DefaultErrorChannel has reached capacity. Are any handlers subscribed?"); + } + } + // if error channel has no subscribers, then errors are at least visible at debug level + if (logger.isDebugEnabled()) { + String errorMessage = "Error Received. Message: " + message.toString(); + if (payload instanceof Throwable) { + logger.debug(errorMessage, (Throwable) payload); + } + else { + logger.debug(errorMessage); + } + } + } + } + +} 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 9114ede538..e6839ecf4c 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 @@ -59,6 +59,8 @@ import org.springframework.util.Assert; */ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lifecycle { + public static final String ERROR_CHANNEL_NAME = "errorChannel"; + private static final int DEFAULT_DISPATCHER_POOL_SIZE = 10; @@ -160,7 +162,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif return; } if (this.getErrorChannel() == null) { - this.setErrorChannel(new SimpleChannel(Integer.MAX_VALUE)); + this.setErrorChannel(new DefaultErrorChannel()); } if (this.executor == null) { this.executor = new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE); @@ -173,11 +175,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public MessageChannel getErrorChannel() { - return this.channelRegistry.getErrorChannel(); + return this.channelRegistry.lookupChannel(ERROR_CHANNEL_NAME); } public void setErrorChannel(MessageChannel errorChannel) { - this.channelRegistry.setErrorChannel(errorChannel); + this.channelRegistry.registerChannel(ERROR_CHANNEL_NAME, errorChannel); } public MessageChannel lookupChannel(String channelName) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java index 98e31f6af1..c5424f00c5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java @@ -29,8 +29,4 @@ public interface ChannelRegistry { MessageChannel lookupChannel(String channelName); - void setErrorChannel(MessageChannel errorChannel); - - MessageChannel getErrorChannel(); - }