QPID-1101 : Updated Direct Exchange so it does not modify lists of queues

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@669885 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 20 insertions, 11 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
index 5dcc2cf143..4da639567a 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
@@ -198,7 +198,7 @@ public class DirectExchange extends AbstractExchange
             _logger.debug("Publishing message to queue " + queues);
         }
 
-            payload.enqueue(queues);
+        payload.enqueue(queues);
 
 
     }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java
index 4f1f550e94..9bf82a3730 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java
@@ -23,6 +23,7 @@ package org.apache.qpid.server.exchange;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.ArrayList;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -44,15 +45,15 @@ class Index
         List<AMQQueue> queues = _index.get(key);
         if(queues == null)
         {
-            queues = new CopyOnWriteArrayList<AMQQueue>();
-            //next call is atomic, so there is no race to create the list
-            List<AMQQueue> active = _index.putIfAbsent(key, queues);
-            if(active != null)
-            {
-                //someone added the new one in faster than we did, so use theirs
-                queues = active;
-            }
+            queues = new ArrayList<AMQQueue>();
+        }
+        else
+        {
+            queues = new ArrayList<AMQQueue>(queues);
         }
+        //next call is atomic, so there is no race to create the list
+        _index.put(key, queues);
+
         if(queues.contains(queue))
         {
             return false;
@@ -68,10 +69,18 @@ class Index
         List<AMQQueue> queues = _index.get(key);
         if (queues != null)
         {
+            queues = new ArrayList<AMQQueue>(queues);
             boolean removed = queues.remove(queue);
-            if (queues.size() == 0)
+            if(removed)
             {
-                _index.remove(key);
+                if (queues.size() == 0)
+                {
+                    _index.remove(key);
+                }
+                else
+                {
+                    _index.put(key, queues);
+                }
             }
             return removed;
         }