QPID-2874 Clustered broker crashes in assertion in cluster/ExpiryPolicy.cpp

- Added missing lock to ExpiryPolicy
- 1-N mapping for expiry ID to mapping when receiving an update.
- Regression test.

A fan-out message (sent to multiple queues e.g. by fanout or topic
exchange) is a single message on multiple queues with a single expiry
ID. During an update however each instance is sent as a separate
message so we need to allow 1-N mapping of expiry ID to message during
update.

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

1 files changed, 14 insertions, 0 deletions

diff --git a/python/qpid/brokertest.py b/python/qpid/brokertest.py
index 14be7c43c9..208bfd6000 100644
--- a/python/qpid/brokertest.py
+++ b/python/qpid/brokertest.py
@@ -523,6 +523,20 @@ class BrokerTest(TestCase):
         cluster = Cluster(self, count, args, expect=expect, wait=wait)
         return cluster
 
+    def assert_browse(self, session, queue, expect_contents, timeout=0):
+        """Assert that the contents of messages on queue (as retrieved
+        using session and timeout) exactly match the strings in
+        expect_contents"""
+
+        r = session.receiver("%s;{mode:browse}"%(queue))
+        actual_contents = []
+        try:
+            for c in expect_contents: actual_contents.append(r.fetch(timeout=timeout).content)
+            while True: actual_contents.append(r.fetch(timeout=0).content) # Check for extra messages.
+        except messaging.Empty: pass
+        r.close()
+        self.assertEqual(expect_contents, actual_contents)
+
 class RethrownException(Exception):
     """Captures the stack trace of the current exception to be thrown later"""
     def __init__(self, msg=""):