diff options
Diffstat (limited to 'qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py')
| -rw-r--r-- | qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py | 557 |
1 files changed, 557 insertions, 0 deletions
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py b/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py index 11b2caad08..e49a997a54 100644 --- a/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py +++ b/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py @@ -19,6 +19,7 @@ from qpid.messaging import * from qpid.tests.messaging import Base +import qmf.console from time import sleep # @@ -360,8 +361,564 @@ class MultiConsumerMsgGroupTests(Base): except Empty: pass + def test_transaction(self): + """ Verify behavior when using transactions. + """ + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","A","B","B","A","B"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + s1 = self.conn.session(transactional=True) + c1 = s1.receiver("msg-group-q", options={"capacity":1}) + s2 = self.conn.session(transactional=True) + c2 = s2.receiver("msg-group-q", options={"capacity":1}) + + # C1 gets group A + m1 = c1.fetch(0) + assert m1.properties['THE-GROUP'] == 'A' + assert m1.content['index'] == 0 + + # C2 gets group B + m2 = c2.fetch(0) + assert m2.properties['THE-GROUP'] == 'B' + assert m2.content['index'] == 2 + + s1.acknowledge(m1) # A-0 consumed, A group freed + s2.acknowledge(m2) # B-2 consumed, B group freed + + s1.commit() + s2.rollback() # release B-2 and group B + + ## Q: ["A1","B2","B3","A4","B5"] + + # C2 should be able to get the next A + m3 = c2.fetch(0) + assert m3.properties['THE-GROUP'] == 'A' + assert m3.content['index'] == 1 + + # C1 should be able to get B-2 + m4 = c1.fetch(0) + assert m4.properties['THE-GROUP'] == 'B' + assert m4.content['index'] == 2 + + s2.acknowledge(m3) # C2 consumes A-1 + s1.acknowledge(m4) # C1 consumes B-2 + s1.commit() # C1 consume B-2 occurs, free group B + + ## Q: [["A1",]"B3","A4","B5"] + + # A-1 is still considered owned by C2, since the commit has yet to + # occur, so the next available to C1 would be B-3 + m5 = c1.fetch(0) # B-3 + assert m5.properties['THE-GROUP'] == 'B' + assert m5.content['index'] == 3 + + # and C2 should find A-4 available, since it owns the A group + m6 = c2.fetch(0) # A-4 + assert m6.properties['THE-GROUP'] == 'A' + assert m6.content['index'] == 4 + + s2.acknowledge(m6) # C2 consumes A-4 + + # uh-oh, A-1 and A-4 released, along with A group + s2.rollback() + + ## Q: ["A1",["B3"],"A4","B5"] + m7 = c1.fetch(0) # A-1 is found + assert m7.properties['THE-GROUP'] == 'A' + assert m7.content['index'] == 1 + + ## Q: [["A1"],["B3"],"A4","B5"] + # since C1 "owns" both A and B group, C2 should find nothing available + try: + m8 = c2.fetch(0) + assert False # should not get here + except Empty: + pass + + # C1 next gets A4 + m9 = c1.fetch(0) + assert m9.properties['THE-GROUP'] == 'A' + assert m9.content['index'] == 4 + + s1.acknowledge() + + ## Q: [["A1"],["B3"],["A4"],"B5"] + # even though C1 acknowledges A1,B3, and A4, B5 is still considered + # owned as the commit has yet to take place + try: + m10 = c2.fetch(0) + assert False # should not get here + except Empty: + pass + + # now A1,B3,A4 dequeued, B5 should be free + s1.commit() + + ## Q: ["B5"] + m11 = c2.fetch(0) + assert m11.properties['THE-GROUP'] == 'B' + assert m11.content['index'] == 5 + + s2.acknowledge() + s2.commit() + + def test_query(self): + """ Verify the queue query method against message groups + """ + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","B","C","A","B","C","A"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + s1 = self.setup_session() + c1 = s1.receiver("msg-group-q", options={"capacity":1}) + s2 = self.setup_session() + c2 = s2.receiver("msg-group-q", options={"capacity":1}) + + m1 = c1.fetch(0) + m2 = c2.fetch(0) + + # at this point, group A should be owned by C1, group B by C2, and + # group C should be available + + # now setup a QMF session, so we can call methods + self.qmf_session = qmf.console.Session() + self.qmf_broker = self.qmf_session.addBroker(str(self.broker)) + brokers = self.qmf_session.getObjects(_class="broker") + assert len(brokers) == 1 + broker = brokers[0] + + # verify the query method call's group information + rc = broker.query("queue", "msg-group-q") + assert rc.status == 0 + assert rc.text == "OK" + results = rc.outArgs['results'] + assert 'qpid.message_group_queue' in results + q_info = results['qpid.message_group_queue'] + assert 'group_header_key' in q_info and q_info['group_header_key'] == "THE-GROUP" + assert 'group_state' in q_info and len(q_info['group_state']) == 3 + for g_info in q_info['group_state']: + assert 'group_id' in g_info + if g_info['group_id'] == "A": + assert g_info['msg_count'] == 3 + assert g_info['consumer'] != "" + elif g_info['group_id'] == "B": + assert g_info['msg_count'] == 2 + assert g_info['consumer'] != "" + elif g_info['group_id'] == "C": + assert g_info['msg_count'] == 2 + assert g_info['consumer'] == "" + else: + assert(False) # should never get here + self.qmf_session.delBroker(self.qmf_broker) + + def test_purge_free(self): + """ Verify we can purge a queue of all messages of a given "unowned" + group. + """ + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","B","A","B","C","A"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + # now setup a QMF session, so we can call methods + self.qmf_session = qmf.console.Session() + self.qmf_broker = self.qmf_session.addBroker(str(self.broker)) + queue = self.qmf_session.getObjects(_class="queue", name="msg-group-q")[0] + assert queue + msg_filter = { 'filter_type' : 'header_match_str', + 'filter_params' : { 'header_key' : "THE-GROUP", + 'header_value' : "B" }} + assert queue.msgDepth == 6 + rc = queue.purge(0, msg_filter) + assert rc.status == 0 + queue.update() + assert queue.msgDepth == 4 + + # verify all B's removed.... + s2 = self.setup_session() + b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":1}) + count = 0 + try: + while True: + m2 = b1.fetch(0) + assert m2.properties['THE-GROUP'] != 'B' + count += 1 + except Empty: + pass + assert count == 4 + + self.qmf_session.delBroker(self.qmf_broker) + + def test_purge_acquired(self): + """ Verify we can purge messages from an acquired group. + """ + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","B","A","B","C","A"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + # acquire group "A" + s1 = self.setup_session() + c1 = s1.receiver("msg-group-q", options={"capacity":1}) + m1 = c1.fetch(0) + assert m1.properties['THE-GROUP'] == 'A' + assert m1.content['index'] == 0 + + # now setup a QMF session, so we can purge group A + self.qmf_session = qmf.console.Session() + self.qmf_broker = self.qmf_session.addBroker(str(self.broker)) + queue = self.qmf_session.getObjects(_class="queue", name="msg-group-q")[0] + assert queue + msg_filter = { 'filter_type' : 'header_match_str', + 'filter_params' : { 'header_key' : "THE-GROUP", + 'header_value' : "A" }} + assert queue.msgDepth == 6 + rc = queue.purge(0, msg_filter) + assert rc.status == 0 + queue.update() + queue.msgDepth == 4 # the pending acquired A still counts! + + # verify all other A's removed.... + s2 = self.setup_session() + b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":1}) + count = 0 + try: + while True: + m2 = b1.fetch(0) + assert m2.properties['THE-GROUP'] != 'A' + count += 1 + except Empty: + pass + assert count == 3 # only 3 really available + s1.acknowledge() # ack the consumed A-0 + self.qmf_session.delBroker(self.qmf_broker) + + def test_purge_count(self): + """ Verify we can purge a fixed number of messages from an acquired + group. + """ + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","B","A","B","C","A"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + # acquire group "A" + s1 = self.setup_session() + c1 = s1.receiver("msg-group-q", options={"capacity":1}) + m1 = c1.fetch(0) + assert m1.properties['THE-GROUP'] == 'A' + assert m1.content['index'] == 0 + + # now setup a QMF session, so we can purge group A + self.qmf_session = qmf.console.Session() + self.qmf_broker = self.qmf_session.addBroker(str(self.broker)) + queue = self.qmf_session.getObjects(_class="queue", name="msg-group-q")[0] + assert queue + msg_filter = { 'filter_type' : 'header_match_str', + 'filter_params' : { 'header_key' : "THE-GROUP", + 'header_value' : "A" }} + assert queue.msgDepth == 6 + rc = queue.purge(1, msg_filter) + assert rc.status == 0 + queue.update() + queue.msgDepth == 5 # the pending acquired A still counts! + + # verify all other A's removed.... + s2 = self.setup_session() + b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":1}) + count = 0 + a_count = 0 + try: + while True: + m2 = b1.fetch(0) + if m2.properties['THE-GROUP'] != 'A': + count += 1 + else: + a_count += 1 + except Empty: + pass + assert count == 3 # non-A's + assert a_count == 1 # and one is an A + s1.acknowledge() # ack the consumed A-0 + self.qmf_session.delBroker(self.qmf_broker) + + def test_move_all(self): + """ Verify we can move messages from an acquired group. + """ + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","B","A","B","C","A"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + # set up destination queue + rcvr = self.ssn.receiver("dest-q; {create:always, delete:receiver," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + # acquire group "A" + s1 = self.setup_session() + c1 = s1.receiver("msg-group-q", options={"capacity":1}) + m1 = c1.fetch(0) + assert m1.properties['THE-GROUP'] == 'A' + assert m1.content['index'] == 0 + + # now setup a QMF session, so we can move what's left of group A + self.qmf_session = qmf.console.Session() + self.qmf_broker = self.qmf_session.addBroker(str(self.broker)) + brokers = self.qmf_session.getObjects(_class="broker") + assert len(brokers) == 1 + broker = brokers[0] + msg_filter = { 'filter_type' : 'header_match_str', + 'filter_params' : { 'header_key' : "THE-GROUP", + 'header_value' : "A" }} + rc = broker.queueMoveMessages("msg-group-q", "dest-q", 0, msg_filter) + assert rc.status == 0 + + # verify all other A's removed from msg-group-q + s2 = self.setup_session() + b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":1}) + count = 0 + try: + while True: + m2 = b1.fetch(0) + assert m2.properties['THE-GROUP'] != 'A' + count += 1 + except Empty: + pass + assert count == 3 # only 3 really available + + # verify the moved A's are at the dest-q + s2 = self.setup_session() + b1 = s2.receiver("dest-q; {mode: browse}", options={"capacity":1}) + count = 0 + try: + while True: + m2 = b1.fetch(0) + assert m2.properties['THE-GROUP'] == 'A' + assert m2.content['index'] == 2 or m2.content['index'] == 5 + count += 1 + except Empty: + pass + assert count == 2 # two A's moved + + s1.acknowledge() # ack the consumed A-0 + self.qmf_session.delBroker(self.qmf_broker) + + def test_move_count(self): + """ Verify we can move a fixed number of messages from an acquired group. + """ + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","B","A","B","C","A"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + # set up destination queue + rcvr = self.ssn.receiver("dest-q; {create:always, delete:receiver," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + # now setup a QMF session, so we can move group B + self.qmf_session = qmf.console.Session() + self.qmf_broker = self.qmf_session.addBroker(str(self.broker)) + brokers = self.qmf_session.getObjects(_class="broker") + assert len(brokers) == 1 + broker = brokers[0] + msg_filter = { 'filter_type' : 'header_match_str', + 'filter_params' : { 'header_key' : "THE-GROUP", + 'header_value' : "B" }} + rc = broker.queueMoveMessages("msg-group-q", "dest-q", 3, msg_filter) + assert rc.status == 0 + + # verify all B's removed from msg-group-q + s2 = self.setup_session() + b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":1}) + count = 0 + try: + while True: + m2 = b1.fetch(0) + assert m2.properties['THE-GROUP'] != 'B' + count += 1 + except Empty: + pass + assert count == 4 + + # verify the moved B's are at the dest-q + s2 = self.setup_session() + b1 = s2.receiver("dest-q; {mode: browse}", options={"capacity":1}) + count = 0 + try: + while True: + m2 = b1.fetch(0) + assert m2.properties['THE-GROUP'] == 'B' + assert m2.content['index'] == 1 or m2.content['index'] == 3 + count += 1 + except Empty: + pass + assert count == 2 + self.qmf_session.delBroker(self.qmf_broker) + def test_reroute(self): + """ Verify we can reroute messages from an acquired group. + """ + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","B","A","B","C","A"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + # create a topic exchange for the reroute + rcvr = self.ssn.receiver("reroute-q; {create: always, delete:receiver," + + " node: {type: topic}}") + + # acquire group "A" + s1 = self.setup_session() + c1 = s1.receiver("msg-group-q", options={"capacity":1}) + m1 = c1.fetch(0) + assert m1.properties['THE-GROUP'] == 'A' + assert m1.content['index'] == 0 + + # now setup a QMF session, so we can reroute group A + self.qmf_session = qmf.console.Session() + self.qmf_broker = self.qmf_session.addBroker(str(self.broker)) + queue = self.qmf_session.getObjects(_class="queue", name="msg-group-q")[0] + assert queue + msg_filter = { 'filter_type' : 'header_match_str', + 'filter_params' : { 'header_key' : "THE-GROUP", + 'header_value' : "A" }} + assert queue.msgDepth == 6 + rc = queue.reroute(0, False, "reroute-q", msg_filter) + assert rc.status == 0 + queue.update() + queue.msgDepth == 4 # the pending acquired A still counts! + + # verify all other A's removed.... + s2 = self.setup_session() + b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":1}) + count = 0 + try: + while True: + m2 = b1.fetch(0) + assert m2.properties['THE-GROUP'] != 'A' + count += 1 + except Empty: + pass + assert count == 3 # only 3 really available + + # and what of reroute-q? + count = 0 + try: + while True: + m2 = rcvr.fetch(0) + assert m2.properties['THE-GROUP'] == 'A' + assert m2.content['index'] == 2 or m2.content['index'] == 5 + count += 1 + except Empty: + pass + assert count == 2 + + s1.acknowledge() # ack the consumed A-0 + self.qmf_session.delBroker(self.qmf_broker) + + def test_queue_delete(self): + """ Test deleting a queue while consumers are active. + """ + + ## Create a msg group queue + + snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," + + " node: {x-declare: {arguments:" + + " {'qpid.group_header_key':'THE-GROUP'}}}}") + + groups = ["A","B","A","B","C"] + messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups] + index = 0 + for m in messages: + m.content['index'] = index + index += 1 + snd.send(m) + + ## Queue = A-0, B-1, A-2, b-3, C-4 + ## Owners= ---, ---, ---, ---, --- + + # create consumers + s1 = self.setup_session() + c1 = s1.receiver("msg-group-q", options={"capacity":1}) + s2 = self.setup_session() + c2 = s2.receiver("msg-group-q", options={"capacity":1}) + + # C1 should acquire A-0 + m1 = c1.fetch(0); + assert m1.properties['THE-GROUP'] == 'A' + assert m1.content['index'] == 0 + + # c2 acquires B-1 + m2 = c2.fetch(0) + assert m2.properties['THE-GROUP'] == 'B' + assert m2.content['index'] == 1 + + # with group A and B owned, and C free, delete the + # queue + snd.close() + self.ssn.close() class StickyConsumerMsgGroupTests(Base): """ |