qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kgiu...@apache.org
Subject svn commit: r1099979 - in /qpid/trunk/qpid/cpp/src/tests: cluster_tests.py queue_flow_limit_tests.py
Date Thu, 05 May 2011 21:41:40 GMT
Author: kgiusti
Date: Thu May  5 21:41:40 2011
New Revision: 1099979

URL: http://svn.apache.org/viewvc?rev=1099979&view=rev
Log:
QPID-3244: unit test for the fix.

Modified:
    qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
    qpid/trunk/qpid/cpp/src/tests/queue_flow_limit_tests.py

Modified: qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_tests.py?rev=1099979&r1=1099978&r2=1099979&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_tests.py Thu May  5 21:41:40 2011
@@ -532,6 +532,54 @@ acl allow all all
         assert not sender.isAlive()
         assert sender.done
 
+    def test_blocked_queue_delete(self):
+        """Verify that producers which are blocked on a queue due to flow
+        control are unblocked when that queue is deleted.
+        """
+
+        cluster = self.cluster(2)
+        cluster[0].startQmf()
+        cluster[1].startQmf()
+
+        # configure a queue with a specific flow limit on first broker
+        ssn0 = cluster[0].connect().session()
+        s0 = ssn0.sender("flq; {create:always, node:{type:queue, x-declare:{arguments:{'qpid.flow_stop_count':5,
'qpid.flow_resume_count':3}}}}")
+        q1 = [q for q in cluster[0].qmf_session.getObjects(_class="queue") if q.name == "flq"][0]
+        oid = q1.getObjectId()
+        self.assertEqual(q1.name, "flq")
+        self.assertEqual(q1.arguments, {u'qpid.flow_stop_count': 5L, u'qpid.flow_resume_count':
3L})
+        assert not q1.flowStopped
+        self.assertEqual(q1.flowStoppedCount, 0)
+
+        # fill the queue on one broker until flow control is active
+        for x in range(5): s0.send(Message(str(x)))
+        sender = ShortTests.BlockedSend(s0, Message(str(6)))
+        sender.start()                  # Tests that sender does block
+        # Verify the broker queue goes into a flowStopped state
+        deadline = time.time() + 1
+        while not q1.flowStopped and time.time() < deadline: q1.update()
+        assert q1.flowStopped
+        self.assertEqual(q1.flowStoppedCount, 1)
+        sender.assert_blocked()         # Still blocked
+
+        # Now verify the  both brokers in cluster have same configuration
+        qs = cluster[1].qmf_session.getObjects(_objectId=oid)
+        self.assertEqual(len(qs), 1)
+        q2 = qs[0]
+        self.assertEqual(q2.name, "flq")
+        self.assertEqual(q2.arguments, {u'qpid.flow_stop_count': 5L, u'qpid.flow_resume_count':
3L})
+        assert q2.flowStopped
+        self.assertEqual(q2.flowStoppedCount, 1)
+
+        # now delete the blocked queue from other broker
+        ssn1 = cluster[1].connect().session()
+        self.evaluate_address(ssn1, "flq;{delete:always}")
+        sender.wait()                   # Verify no longer blocked.
+
+        ssn0.connection.close()
+        ssn1.connection.close()
+        cluster_test_logs.verify_logs()
+
 
     def test_alternate_exchange_update(self):
         """Verify that alternate-exchange on exchanges and queues is propagated to new members
of a cluster. """

Modified: qpid/trunk/qpid/cpp/src/tests/queue_flow_limit_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/queue_flow_limit_tests.py?rev=1099979&r1=1099978&r2=1099979&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/queue_flow_limit_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/queue_flow_limit_tests.py Thu May  5 21:41:40 2011
@@ -325,5 +325,47 @@ class QueueFlowLimitTests(TestBase010):
         self.verify_limit(TestQ(oid))
 
 
+    def test_blocked_queue_delete(self):
+        """ Verify that blocked senders are unblocked when a queue that is flow
+        controlled is deleted.
+        """
+
+        class BlockedSender(Thread):
+            def __init__(self, tester, queue, count, capacity=10):
+                self.tester = tester
+                self.queue = queue
+                self.count = count
+                self.capacity = capacity
+                Thread.__init__(self)
+                self.done = False
+                self.start()
+            def run(self):
+                # spawn qpid-send
+                p = self.tester._start_qpid_send(self.queue,
+                                                 self.count,
+                                                 self.capacity)
+                p.close()  # waits for qpid-send to complete
+                self.done = True
+
+        self.startQmf();
+        oid = self._create_queue("kill-q", stop_size=10, resume_size=2)
+        q = self.qmf.getObjects(_objectId=oid)[0]
+        self.failIf(q.flowStopped)
+
+        sender = BlockedSender(self, "kill-q", count=100)
+        # wait for flow control
+        deadline = time() + 10
+        while (not q.flowStopped) and time() < deadline:
+            q.update()
+
+        self.failUnless(q.flowStopped)
+        self.failIf(sender.done)   # sender blocked
+
+        self._delete_queue("kill-q")
+        sender.join(5)
+        self.failIf(sender.isAlive())
+        self.failUnless(sender.done)
+
+
 
 



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message