diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index e990f4236a..870d0d650f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -18,10 +18,15 @@ package org.springframework.integration.dispatcher; import java.util.ArrayList; import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedList; import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + +import org.springframework.core.OrderComparator; +import org.springframework.core.Ordered; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageHandler; @@ -31,6 +36,14 @@ import org.springframework.util.Assert; /** * Base class for {@link MessageDispatcher} implementations. * + *

The {@link Comparator} that determines the order may be provided via the + * {@link #setComparator(Comparator)} method. If none is provided, the default + * will be an instance of {@link OrderComparator}, and any {@link MessageHandler} + * that implements {@link org.springframework.core.Ordered} and has an order + * value other than LOWEST_PRECEDENCE will be ordered accordingly. Any other + * handlers will be placed at the end of the list in their original + * insertion-order. + * * @author Mark Fisher * @author Iwein Fuld */ @@ -38,10 +51,26 @@ public abstract class AbstractDispatcher implements MessageDispatcher { protected final Log logger = LogFactory.getLog(this.getClass()); - private final List handlers = new ArrayList(); + private volatile List handlers = new LinkedList(); + + @SuppressWarnings("unchecked") + private volatile Comparator comparator = new OrderComparator(); private volatile TaskExecutor taskExecutor; + private final Object handlerListMonitor = new Object(); + + + /** + * Provide a {@link Comparator} that will determine the sorting order of + * the handler list. If no comparator is configured, the default will be + * an instance of {@link OrderComparator}. + */ + public void setComparator(Comparator comparator) { + Assert.notNull(comparator, "comparator must not be null"); + this.comparator = comparator; + } + /** * Specify a {@link TaskExecutor} for invoking the handlers. If none is * provided, the invocation will occur in the thread that runs this polling @@ -56,18 +85,44 @@ public abstract class AbstractDispatcher implements MessageDispatcher { } protected List getHandlers() { - return Collections.unmodifiableList(handlers); + return Collections.unmodifiableList(this.handlers); } public boolean addHandler(MessageHandler handler) { if (this.handlers.contains(handler)) { return false; } - return this.handlers.add(handler); + if (!this.hasOrder(handler)) { + synchronized (this.handlerListMonitor) { + return this.handlers.add(handler); + } + } + synchronized (this.handlerListMonitor) { + List orderedHandlers = new ArrayList(); + orderedHandlers.add(handler); + List handlerList = new ArrayList(handlers); + for (MessageHandler nextHandler : handlerList) { + if (this.hasOrder(nextHandler)) { + orderedHandlers.add(nextHandler); + } + } + handlerList.removeAll(orderedHandlers); + Collections.sort(orderedHandlers, this.comparator); + orderedHandlers.addAll(handlerList); + this.handlers = orderedHandlers; + return true; + } } public boolean removeHandler(MessageHandler handler) { - return this.handlers.remove(handler); + synchronized (this.handlerListMonitor) { + return this.handlers.remove(handler); + } + } + + private boolean hasOrder(MessageHandler handler) { + return handler instanceof Ordered + && ((Ordered) handler).getOrder() < Ordered.LOWEST_PRECEDENCE; } public String toString() { @@ -87,14 +142,11 @@ public abstract class AbstractDispatcher implements MessageDispatcher { } catch (MessageRejectedException e) { if (logger.isDebugEnabled()) { - logger - .debug( - "Handler '" - + handler - + "' rejected Message, if other handlers are available this dispatcher may try to send to those.", - e); + logger.debug("Handler '" + handler + "' rejected Message, " + + "if other handlers are available this dispatcher may try to send to those.", e); } return false; } } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index 74f34a5350..73c6cbdc09 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -16,6 +16,7 @@ package org.springframework.integration.dispatcher; +import java.util.List; import java.util.UUID; import org.springframework.core.task.TaskExecutor; @@ -49,8 +50,9 @@ public class BroadcastingDispatcher extends AbstractDispatcher { public boolean dispatch(Message message) { int sequenceNumber = 1; - int sequenceSize = getHandlers().size(); - for (final MessageHandler handler : getHandlers()) { + List handlers = this.getHandlers(); + int sequenceSize = handlers.size(); + for (final MessageHandler handler : handlers) { final Message messageToSend = (!this.applySequence) ? message : MessageBuilder.fromMessage(message) .setSequenceNumber(sequenceNumber++) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java index 4823e2f5e7..0c4b63dce3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java @@ -16,51 +16,24 @@ package org.springframework.integration.dispatcher; -import java.util.ArrayList; -import java.util.Collections; -import java.util.Comparator; import java.util.Iterator; import java.util.List; -import org.springframework.core.OrderComparator; import org.springframework.integration.message.MessageHandler; -import org.springframework.util.Assert; /** * {@link AbstractUnicastDispatcher} that will try its handlers in the * same order every dispatch. * - *

The {@link Comparator} that determines the order may be provided via the - * {@link #setComparator(Comparator)} method. If none is provided, the default - * will be an instance of {@link OrderComparator}, and any {@link MessageHandler} - * that implements {@link org.springframework.core.Ordered} will be ordered - * accordingly. Any other handlers will be placed at the end of the list. - * * @author Mark Fisher * @author Iwein Fuld */ public class FailOverDispatcher extends AbstractUnicastDispatcher { - @SuppressWarnings("unchecked") - private Comparator comparator = new OrderComparator(); - - - /** - * Provide a {@link Comparator} that will determine the sorting order of - * the handler list and hence the order of the failover chain. If no - * comparator is configured, the default will be an instance of - * {@link OrderComparator}. - */ - public void setComparator(Comparator comparator) { - Assert.notNull(comparator, "comparator must not be null"); - this.comparator = comparator; - } @Override protected Iterator getHandlerIterator(List handlers) { - List handlerList = new ArrayList(handlers); - Collections.sort(handlerList, this.comparator); - return handlerList.iterator(); + return handlers.iterator(); } }