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 3baf50fcf2..3a961aea5d 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 @@ -89,9 +89,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher { } public boolean addHandler(MessageHandler handler) { - synchronized (this.handlerListMonitor) { - return this.handlers.add(handler); - } + return this.handlers.add(handler); } public boolean removeHandler(MessageHandler handler) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSet.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSet.java index 0c359b282e..aed3c95952 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSet.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSet.java @@ -19,7 +19,8 @@ import java.util.Collection; import java.util.Comparator; import java.util.LinkedHashSet; import java.util.Set; -import java.util.TreeSet; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.springframework.core.OrderComparator; import org.springframework.core.Ordered; @@ -31,8 +32,8 @@ import org.springframework.util.CollectionUtils; * Special Set that maintains the following semantics: * All elements that are un-ordered (do not implement {@link Ordered} interface or annotated {@link Order} annotation) * will be stored in the order in which they were added, maintaining the semantics of the {@link LinkedHashSet}. - * However, there is a special {@link Comparator} (instantiated by default) for this implementation of {@link Set}, - * which is aware of the {@link Ordered} interface or and {@link Order} annotation. Those elements will have + * However, for all {@link Ordered} elements a {@link Comparator} (instantiated by default) for this implementation of {@link Set}, + * will be used. Those elements will have * precedence over un-ordered elements. If elements have the same order but themselves do not equal to one another * they will be placed to the right (appended next to) of the element with the same order, thus preserving the order * of the insertion and maintaining {@link LinkedHashSet} semantics. @@ -41,77 +42,111 @@ import org.springframework.util.CollectionUtils; * @since 1.0.3 */ class OrderedAwareLinkedHashSet extends LinkedHashSet { - private TreeSet orderedSet = new TreeSet(new EqualsAwareOrderComparator()); + + OrderComparator comparator = new OrderComparator(); + Lock lock = new ReentrantLock(); /** * Every time when Ordered element is added via this method * this Set will be re-sorted, otherwise the element is simply added to the end of the stack. * If adding multiple objects for performance reasons it is recommended to use * addAll(Collection c) method which first adds all the elements to this set * and then calls reinitializeThis() method. + * Added element must not be null; */ - public boolean add(E o){ - boolean present = this.doAdd(o); - if (o instanceof Ordered){ - this.reinitializeThis(); + public boolean add(E o){ + lock.lock(); + try { + Assert.notNull(o,"Can not add NULL object"); + boolean present = false; + if (o instanceof Ordered){ + present = this.reinitializeThis(o); + } else { + present = super.add(o); + } + return present; + } finally { + lock.unlock(); } - return present; } /** * Adds all elements in this Collection and then resorts this set * via call to the reinitializeThis() method */ - public boolean addAll(Collection c){ - Assert.notNull(c,"Can not merge with NULL set"); + public boolean addAll(Collection c){ + lock.lock(); + try { + Assert.notNull(c,"Can not merge with NULL set"); + for (E object : c) { + this.add(object); + } + return true; + } finally { + lock.unlock(); + } + } + /** + * + */ + public boolean remove(Object o){ + lock.lock(); + try { + return super.remove(o); + } finally { + lock.unlock(); + } + } + /** + * + */ + public boolean removeAll(Collection c){ if (CollectionUtils.isEmpty(c)){ return false; } - for (E object : c) { - this.doAdd(object); - } - this.reinitializeThis(); - // due to the nature of previous method - // at this point collection will always be modified - return true; - } - /** - * - * @param o - * @return - */ - private boolean doAdd(E o){ - if (o instanceof Ordered){ - return orderedSet.add(o); - } else { - return super.add(o); + lock.lock(); + try { + super.removeAll(c); + return true; + } finally { + lock.unlock(); } + } + /** * */ - private void reinitializeThis(){ - E[] tempUnorderedElements = (E[]) super.toArray(); - E[] tempOrderedElements = (E[]) orderedSet.toArray(); + private boolean reinitializeThis(Object adding){ + boolean added = false; + E[] tempUnorderedElements = (E[]) this.toArray(); + if (super.contains(adding)){ + return false; + } super.clear(); - for (E object : tempOrderedElements) { - super.add(object); - } - for (E object : tempUnorderedElements) { - super.add(object); - } - } - /** - * Will reuse most of the functionality of OrderComparator, however if - * elements have the same order, then 1 will be returned, thus positioning - * such element to the right of the existing element. - */ - private static class EqualsAwareOrderComparator extends OrderComparator { - public int compare(Object o1, Object o2) { - int value = super.compare(o1, o2); - // see if objects are not equal - if (value == 0 && !o1.equals(o2)){ - return 1; + + if (tempUnorderedElements.length == 0){ + added = super.add((E) adding); + } else { + Set tempSet = new LinkedHashSet(); + for (E current : tempUnorderedElements) { + if (current instanceof Ordered && adding instanceof Ordered){ + if (((Ordered)adding).getOrder() < ((Ordered)current).getOrder()){ + added = super.add((E) adding); + super.add(current); + } else { + super.add(current); + } + } else { + tempSet.add(current); + } + } + if (!added){ + added = super.add((E) adding); + } + for (Object object : tempSet) { + super.add((E) object); } - return value; } + return added; } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSetTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSetTests.java index 5b34fbbd4e..0e6c79f63c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSetTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSetTests.java @@ -16,6 +16,7 @@ package org.springframework.integration.dispatcher; import java.util.ArrayList; +import java.util.LinkedList; import java.util.List; import org.junit.Test; @@ -130,6 +131,145 @@ public class OrderedAwareLinkedHashSetTests { assertEquals(o10, elements[9]); } + @Test + public void testConcurrent(){ + for(int i = 0; i < 1000; i++){ + this.doConcurrent(); + } + } + private void doConcurrent(){ + final OrderedAwareLinkedHashSet setToTest = new OrderedAwareLinkedHashSet(); + final Object o1 = new Foo(3); + final Object o2 = new Foo(1); + final Object o3 = new Foo(2); + final Object o4 = new Foo(2); + final Object o5 = new Foo(Ordered.LOWEST_PRECEDENCE); + final Object o6 = new Foo(Ordered.LOWEST_PRECEDENCE); + final Object o7 = new Foo(Ordered.HIGHEST_PRECEDENCE); + final Object o8 = new Foo(Ordered.HIGHEST_PRECEDENCE); + final Object o9 = new Foo(4); + final Object o10 = new Foo(2); + Thread t1 = new Thread(new Runnable() { + public void run() { + setToTest.add(o1); + setToTest.add(o3); + setToTest.add(o5); + setToTest.add(o7); + setToTest.add(o9); + } + }); + Thread t2 = new Thread(new Runnable() { + public void run() { + setToTest.add(o2); + setToTest.add(o4); + setToTest.add(o6); + setToTest.add(o8); + setToTest.add(o10); + } + }); + Thread t3 = new Thread(new Runnable() { + public void run() { + setToTest.add(1); + setToTest.add(new Foo(2)); + setToTest.add(3); + setToTest.add(new Foo(9)); + setToTest.add(8); + } + }); + t1.start(); + t2.start(); + t3.start(); + try { + t1.join(); + t2.join(); + t3.join(); + } catch (Exception e) { + e.printStackTrace(); + throw new RuntimeException(e); + } + + + assertEquals(15, setToTest.size()); + } + /** + * Will test addAll operation including the removal and adding an object in the concurrent environment + */ + @Test + public void testConcurrentAll(){ + for(int i = 0; i < 1000; i++){ + this.doConcurrentAll(); + } + } + public void doConcurrentAll(){ + final List tempList = new ArrayList(); + Object o1 = new Foo(3); + Object o2 = new Foo(1); + Object o3 = "Bla"; + Object o4 = new Foo(2); + final Object o5 = new Foo(Ordered.LOWEST_PRECEDENCE); + Object o6 = new Foo(Ordered.LOWEST_PRECEDENCE); + final Object o7 = new Foo(Ordered.HIGHEST_PRECEDENCE); + Object o8 = new Foo(Ordered.HIGHEST_PRECEDENCE); + Object o9 = new Foo(4); + Object o10 = "Baz"; + + tempList.add(o1); + tempList.add(o2); + tempList.add(o3); + tempList.add(o4); + tempList.add(o5); + tempList.add(o6); + tempList.add(o7); + tempList.add(o8); + tempList.add(o9); + tempList.add(o10); + final OrderedAwareLinkedHashSet orderAwareSet = new OrderedAwareLinkedHashSet(); + Thread t1 = new Thread(new Runnable() { + public void run() { + orderAwareSet.addAll(tempList); + orderAwareSet.remove(o5); + orderAwareSet.remove(o7); + } + }); + final List tempList2 = new ArrayList(); + final Foo foo5 = new Foo(5); + Foo foo6 = new Foo(6); + tempList2.add(foo6); + tempList2.add(foo5); + tempList2.add(new Foo(30)); + tempList2.add(new Foo(10)); + tempList2.add(1); + tempList2.add(new Foo(28)); + tempList2.add(10); + tempList2.add(13); + tempList2.add(new Foo(63)); + Thread t2 = new Thread(new Runnable() { + public void run() { + orderAwareSet.addAll(tempList2); + orderAwareSet.remove(foo5); + } + }); + Thread t3 = new Thread(new Runnable() { + public void run() { + orderAwareSet.add("hello"); + orderAwareSet.add("hello again"); + } + }); + + t1.start(); + t2.start(); + t3.start(); + try { + t1.join(); + t2.join(); + t3.join(); + } catch (Exception e) { + e.printStackTrace(); + throw new RuntimeException(e); + } + Object[] elements = orderAwareSet.toArray(); + assertEquals(18, elements.length); + } private static class Foo implements Ordered { private int order; public Foo(int order){