In ClientChannel: Use subscribe and flush in place of get; use per-subscriber flow control for managing prefetches.

In brokers Session: set credit to 0 when subscription is created (modified python tests accordingly)



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

1 files changed, 7 insertions, 5 deletions

diff --git a/qpid/python/tests_0-10/broker.py b/qpid/python/tests_0-10/broker.py
index 0eb71287ec..0df7eb09fa 100644
--- a/qpid/python/tests_0-10/broker.py
+++ b/qpid/python/tests_0-10/broker.py
@@ -35,7 +35,7 @@ class BrokerTests(TestBase):
 
         # No ack consumer
         ctag = "tag1"
-        ch.message_subscribe(queue = "myqueue", destination = ctag, confirm_mode = 0)
+        self.subscribe(ch, queue = "myqueue", destination = ctag)
         body = "test no-ack"
         ch.message_transfer(content = Content(body, properties = {"routing_key" : "myqueue"}))
         msg = self.client.queue(ctag).get(timeout = 5)
@@ -44,7 +44,9 @@ class BrokerTests(TestBase):
         # Acknowledging consumer
         self.queue_declare(ch, queue = "otherqueue")
         ctag = "tag2"
-        ch.message_subscribe(queue = "otherqueue", destination = ctag, confirm_mode = 1)
+        self.subscribe(ch, queue = "otherqueue", destination = ctag, confirm_mode = 1)
+        ch.message_flow(destination=ctag, unit=0, value=0xFFFFFFFF)
+        ch.message_flow(destination=ctag, unit=1, value=0xFFFFFFFF)
         body = "test ack"
         ch.message_transfer(content = Content(body, properties = {"routing_key" : "otherqueue"}))
         msg = self.client.queue(ctag).get(timeout = 5)
@@ -60,7 +62,7 @@ class BrokerTests(TestBase):
         self.queue_declare(channel, queue="test-queue") 
         channel.queue_bind(queue="test-queue", exchange="test-exchange", routing_key="key")
         consumer_tag = "tag1"
-        channel.message_subscribe(queue="test-queue", destination=consumer_tag, confirm_mode = 0)
+        self.subscribe(queue="test-queue", destination=consumer_tag)
         queue = self.client.queue(consumer_tag)
 
         body = "Immediate Delivery"
@@ -84,7 +86,7 @@ class BrokerTests(TestBase):
         channel.message_transfer(destination="test-exchange", content = Content(body, properties = {"routing_key" : "key"}))
 
         consumer_tag = "tag1"
-        channel.message_subscribe(queue="test-queue", destination=consumer_tag, confirm_mode = 0)
+        self.subscribe(queue="test-queue", destination=consumer_tag)
         queue = self.client.queue(consumer_tag)
         msg = queue.get(timeout=5)
         self.assert_(msg.content.body == body)
@@ -111,7 +113,7 @@ class BrokerTests(TestBase):
     def test_channel_flow(self):
         channel = self.channel
         channel.queue_declare(queue="flow_test_queue", exclusive=True)
-        channel.message_subscribe(destination="my-tag", queue="flow_test_queue")
+        self.subscribe(destination="my-tag", queue="flow_test_queue")
         incoming = self.client.queue("my-tag")
         
         channel.channel_flow(active=False)