Added a DefaultErrorChannel implementation to be used by the MessageBus. Provided a key for the error channel name: MessageBus.ERROR_CHANNEL_NAME (INT-119). Removed the getErrorChannel and setErrorChannel() methods from the ChannelRegistry interface (since the key can be used to do a lookup). MessageBus still provides convenience methods.

This commit is contained in:
Mark Fisher
2008-02-22 19:51:00 +00:00
parent 701743c979
commit aa66647d7e
3 changed files with 70 additions and 7 deletions

View File

@@ -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);
}
}
}
}
}

View File

@@ -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) {

View File

@@ -29,8 +29,4 @@ public interface ChannelRegistry {
MessageChannel lookupChannel(String channelName);
void setErrorChannel(MessageChannel errorChannel);
MessageChannel getErrorChannel();
}