From 56713aa052a9db0007a1478bae2ad58fbfb96f4e Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 26 Jun 2009 04:51:40 +0000 Subject: [PATCH] INT-684 Moved ordering of MessageHandlers up to the base AbstractDispatcher so that all subclasses will have ordering (rather than the FailOverDispatcher only). The actual strategy for invoking within that order still varies (e.g. round-robin vs. failover) --- .../dispatcher/AbstractDispatcher.java | 74 ++++++++++++++++--- .../dispatcher/BroadcastingDispatcher.java | 6 +- .../dispatcher/FailOverDispatcher.java | 29 +------- 3 files changed, 68 insertions(+), 41 deletions(-) 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(); } }