QPID-3603: Fix & clean up in HA code.

- Fix fencepost error in getFirstSafe()
- QueueGuard::attach completes messages before the ReplicatingSubscription postion
- Fix minor test issues in brokertest.py and ha_test.py.
- ReplicatingSubscription check for ready in acknowledge not dispatch.
- HA test fix: retry wait_status retry on ConnectErrors, broker may not be up.

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

2 files changed, 18 insertions, 11 deletions

diff --git a/qpid/cpp/src/tests/brokertest.py b/qpid/cpp/src/tests/brokertest.py
index 93b3868907..b8b642c504 100644
--- a/qpid/cpp/src/tests/brokertest.py
+++ b/qpid/cpp/src/tests/brokertest.py
@@ -76,20 +76,20 @@ def error_line(filename, n=1):
     except: return ""
     return ":\n" + "".join(result)
 
-def retry(function, timeout=10, delay=.01):
+def retry(function, timeout=10, delay=.01, max_delay=1):
     """Call function until it returns a true value or timeout expires.
-    Double the delay for each retry. Returns what function returns if
-    true, None if timeout expires."""
+    Double the delay for each retry up to max_delay.
+    Returns what function returns if true, None if timeout expires."""
     deadline = time.time() + timeout
     ret = None
-    while not ret:
+    while True:
         ret = function()
+        if ret: return ret
         remaining = deadline - time.time()
         if remaining <= 0: return False
         delay = min(delay, remaining)
         time.sleep(delay)
-        delay *= 2
-    return ret
+        delay = min(delay*2, max_delay)
 
 class AtomicCounter:
     def __init__(self):
@@ -657,7 +657,7 @@ class NumberedReceiver(Thread):
             m = self.read_message()
             while m != -1:
                 self.receiver.assert_running()
-                assert m <= self.received, "Missing message %s>%s"%(m, self.received)
+                assert m <= self.received, "%s missing message %s>%s"%(queue, m, self.received)
                 if (m == self.received): # Ignore duplicates
                     self.received += 1
                     if self.sender:
diff --git a/qpid/cpp/src/tests/ha_tests.py b/qpid/cpp/src/tests/ha_tests.py
index 0f9efdf80a..e5a204d03c 100755
--- a/qpid/cpp/src/tests/ha_tests.py
+++ b/qpid/cpp/src/tests/ha_tests.py
@@ -77,10 +77,17 @@ class HaBroker(Broker):
         if not self._agent: self._agent = QmfAgent(self.host_port())
         return self._agent
 
-    def ha_status(self): return self.agent().getHaBroker().status
+    def ha_status(self):
+        hb = self.agent().getHaBroker()
+        hb.update()
+        return hb.status
 
     def wait_status(self, status):
-        assert retry(lambda: self.ha_status() == status), "%s, %r != %r"%(self, self.ha_status(), status)
+        def try_get_status():
+            # Ignore ConnectionError, the broker may not be up yet.
+            try: return self.ha_status() == status;
+            except ConnectionError: return False
+        assert retry(try_get_status, timeout=20), "%s, %r != %r"%(self, self.ha_status(), status)
 
     # FIXME aconway 2012-05-01: do direct python call to qpid-config code.
     def qpid_config(self, args):
@@ -755,12 +762,12 @@ class LongTests(BrokerTest):
                     return receivers[0].received > n + 100
                 # FIXME aconway 2012-05-17: client reconnect sometimes takes > 1 sec.
                 assert retry(enough, 10), "Stalled: %s < %s+100"%(receivers[0].received, n)
-            for s in senders: s.stop()
-            for r in receivers: r.stop()
         except:
             traceback.print_exc()
             raise
         finally:
+            for s in senders: s.stop()
+            for r in receivers: r.stop()
             dead = []
             for i in xrange(3):
                 if not brokers[i].is_running(): dead.append(i)