You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by tr...@apache.org on 2013/08/08 16:15:48 UTC
svn commit: r1511797 - in /qpid/trunk/qpid/extras/dispatch: src/container.c
tests/system_tests_one_router.py
Author: tross
Date: Thu Aug 8 14:15:47 2013
New Revision: 1511797
URL: http://svn.apache.org/r1511797
Log:
QPID-5045 - Added tests for message REJECT and for the three-ack delivery pattern.
Modified:
qpid/trunk/qpid/extras/dispatch/src/container.c
qpid/trunk/qpid/extras/dispatch/tests/system_tests_one_router.py
Modified: qpid/trunk/qpid/extras/dispatch/src/container.c
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/src/container.c?rev=1511797&r1=1511796&r2=1511797&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/src/container.c (original)
+++ qpid/trunk/qpid/extras/dispatch/src/container.c Thu Aug 8 14:15:47 2013
@@ -325,8 +325,10 @@ static int process_handler(dx_container_
}
//
- // Step 2.5: Traverse all of the links on the connection looking for
- // links. Call the attached node's writable handler for such links.
+ // Step 2.5: Call the attached node's writable handler for all active links
+ // on the connection. Note that in Dispatch, links are considered
+ // bidirectional. Incoming and outgoing only pertains to deliveries and
+ // deliveries are a subset of the traffic that flows both directions on links.
//
pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE);
while (pn_link) {
Modified: qpid/trunk/qpid/extras/dispatch/tests/system_tests_one_router.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/tests/system_tests_one_router.py?rev=1511797&r1=1511796&r2=1511797&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/tests/system_tests_one_router.py (original)
+++ qpid/trunk/qpid/extras/dispatch/tests/system_tests_one_router.py Thu Aug 8 14:15:47 2013
@@ -37,11 +37,14 @@ class RouterTest(unittest.TestCase):
self.router.terminate()
self.router.wait()
- def subscribe(self, messenger, address):
- messenger.subscribe(address)
+ def flush(self, messenger):
while messenger.work(100):
pass
+ def subscribe(self, messenger, address):
+ messenger.subscribe(address)
+ self.flush(messenger)
+
def test_0_discard(self):
addr = "amqp://0.0.0.0:20000/discard/1"
M1 = Messenger()
@@ -153,9 +156,12 @@ class RouterTest(unittest.TestCase):
tm.address = addr
tm.body = {'number': 0}
+
+ ##
+ ## Test ACCEPT
+ ##
tx_tracker = M1.put(tm)
M1.send(0)
-
M2.recv(1)
rx_tracker = M2.get(rm)
self.assertEqual(0, rm.body['number'])
@@ -164,13 +170,29 @@ class RouterTest(unittest.TestCase):
M2.accept(rx_tracker)
M2.settle(rx_tracker)
- while M2.work(100):
- pass
- while M1.work(100):
- pass
+ self.flush(M2)
+ self.flush(M1)
self.assertEqual(ACCEPTED, M1.status(tx_tracker))
+ ##
+ ## Test REJECT
+ ##
+ tx_tracker = M1.put(tm)
+ M1.send(0)
+ M2.recv(1)
+ rx_tracker = M2.get(rm)
+ self.assertEqual(0, rm.body['number'])
+ self.assertEqual(PENDING, M1.status(tx_tracker))
+
+ M2.reject(rx_tracker)
+ M2.settle(rx_tracker)
+
+ self.flush(M2)
+ self.flush(M1)
+
+ self.assertEqual(REJECTED, M1.status(tx_tracker))
+
M1.stop()
M2.stop()
@@ -178,13 +200,63 @@ class RouterTest(unittest.TestCase):
# def test_4_unsettled_undeliverable(self):
# pass
-# def test_4_three_ack(self):
-# pass
+ def test_5_three_ack(self):
+ addr = "amqp://0.0.0.0:20000/three_ack/1"
+ M1 = Messenger()
+ M2 = Messenger()
+
+ M1.timeout = 1000
+ M2.timeout = 1000
+ M1.outgoing_window = 5
+ M2.incoming_window = 5
+
+ M1.start()
+ M2.start()
+ self.subscribe(M2, addr)
+
+ tm = Message()
+ rm = Message()
+
+ tm.address = addr
+ tm.body = {'number': 200}
+
+ tx_tracker = M1.put(tm)
+ M1.send(0)
+ M2.recv(1)
+ rx_tracker = M2.get(rm)
+ self.assertEqual(200, rm.body['number'])
+ self.assertEqual(PENDING, M1.status(tx_tracker))
+
+ M2.accept(rx_tracker)
+
+ self.flush(M2)
+ self.flush(M1)
+
+ self.assertEqual(ACCEPTED, M1.status(tx_tracker))
+
+ M1.settle(tx_tracker)
+
+ self.flush(M1)
+ self.flush(M2)
+
+ ##
+ ## We need a way to verify on M2 (receiver) that the tracker has been
+ ## settled on the M1 (sender). [ See PROTON-395 ]
+ ##
+
+ M2.settle(rx_tracker)
+
+ self.flush(M2)
+ self.flush(M1)
+
+ M1.stop()
+ M2.stop()
+
-# def test_5_link_route_sender(self):
+# def test_6_link_route_sender(self):
# pass
-# def test_6_link_route_receiver(self):
+# def test_7_link_route_receiver(self):
# pass
if __name__ == '__main__':
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org