Author: aconway
Date: Fri Mar 16 15:54:11 2007
New Revision: 519171
URL: http://svn.apache.org/viewvc?view=rev&rev=519171
Log:
Merged revisions 500305 via svnmerge from
https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9
........
r500305 | gsim | 2007-01-26 13:51:21 -0500 (Fri, 26 Jan 2007) | 3 lines
Updates to use message class in place of basic.
........
Modified:
incubator/qpid/trunk/qpid/ (props changed)
incubator/qpid/trunk/qpid/python/java_failing.txt
incubator/qpid/trunk/qpid/python/qpid/codec.py
incubator/qpid/trunk/qpid/python/qpid/connection.py
incubator/qpid/trunk/qpid/python/qpid/peer.py
incubator/qpid/trunk/qpid/python/qpid/testlib.py
incubator/qpid/trunk/qpid/python/tests/broker.py
incubator/qpid/trunk/qpid/python/tests/example.py
incubator/qpid/trunk/qpid/python/tests/exchange.py
incubator/qpid/trunk/qpid/python/tests/queue.py
incubator/qpid/trunk/qpid/python/tests/testlib.py
incubator/qpid/trunk/qpid/python/tests/tx.py
Propchange: incubator/qpid/trunk/qpid/
------------------------------------------------------------------------------
--- svnmerge-integrated (original)
+++ svnmerge-integrated Fri Mar 16 15:54:11 2007
@@ -1 +1 @@
-/incubator/qpid/branches/qpid.0-9:1-492620,496593,497277
+/incubator/qpid/branches/qpid.0-9:1-492620,496593,497277,500305
Modified: incubator/qpid/trunk/qpid/python/java_failing.txt
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/java_failing.txt?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/java_failing.txt (original)
+++ incubator/qpid/trunk/qpid/python/java_failing.txt Fri Mar 16 15:54:11 2007
@@ -1,2 +1,19 @@
+tests.basic.BasicTests.test_ack
+tests.basic.BasicTests.test_cancel
+tests.basic.BasicTests.test_consume_exclusive
+tests.basic.BasicTests.test_consume_no_local
+tests.basic.BasicTests.test_consume_queue_errors
+tests.basic.BasicTests.test_consume_unique_consumers
+tests.basic.BasicTests.test_get
+tests.basic.BasicTests.test_qos_prefetch_count
+tests.basic.BasicTests.test_qos_prefetch_size
+tests.basic.BasicTests.test_recover_requeue
tests.exchange.RecommendedTypesRuleTests.testTopic
tests.exchange.RequiredInstancesRuleTests.testAmqTopic
+tests.queue.QueueTests.test_declare_exclusive
+tests.queue.QueueTests.test_declare_passive
+tests.queue.QueueTests.test_delete_ifempty
+tests.queue.QueueTests.test_delete_ifunused
+tests.queue.QueueTests.test_delete_simple
+tests.queue.QueueTests.test_purge
+tests.testlib.TestBaseTest.testMessageProperties
Modified: incubator/qpid/trunk/qpid/python/qpid/codec.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/codec.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/codec.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/codec.py Fri Mar 16 15:54:11 2007
@@ -157,14 +157,15 @@
def encode_table(self, tbl):
enc = StringIO()
codec = Codec(enc)
- for key, value in tbl.items():
- codec.encode_shortstr(key)
- if isinstance(value, basestring):
- codec.write("S")
- codec.encode_longstr(value)
- else:
- codec.write("I")
- codec.encode_long(value)
+ if tbl:
+ for key, value in tbl.items():
+ codec.encode_shortstr(key)
+ if isinstance(value, basestring):
+ codec.write("S")
+ codec.encode_longstr(value)
+ else:
+ codec.write("I")
+ codec.encode_long(value)
s = enc.getvalue()
self.encode_long(len(s))
self.write(s)
Modified: incubator/qpid/trunk/qpid/python/qpid/connection.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/connection.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/connection.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/connection.py Fri Mar 16 15:54:11 2007
@@ -204,6 +204,9 @@
method = Method.decode(spec, dec, size - 20)
return Request(id, mark, method)
+ def __str__(self):
+ return "[%s] Request(%s) %s" % (self.channel, self.id, self.method)
+
class Response(Frame):
type = "frame_response"
Modified: incubator/qpid/trunk/qpid/python/qpid/peer.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/peer.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/peer.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/peer.py Fri Mar 16 15:54:11 2007
@@ -237,6 +237,8 @@
frame = Method(type, type.arguments(*args, **kwargs))
if self.reliable:
self.request(frame, self.queue_response, content)
+ if not frame.method.responses:
+ return None
try:
resp = self.responses.get()
return Message(self, resp)
Modified: incubator/qpid/trunk/qpid/python/qpid/testlib.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/testlib.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/testlib.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/testlib.py Fri Mar 16 15:54:11 2007
@@ -80,7 +80,7 @@
def __init__(self):
# Defaults
self.setBroker("localhost")
- self.specfile = "../specs/amqp.0-8.xml"
+ self.spec = "../specs/amqp.0-8.xml"
self.verbose = 1
self.ignore = []
@@ -203,8 +203,11 @@
def consume(self, queueName):
"""Consume from named queue returns the Queue object."""
- reply = self.channel.basic_consume(queue=queueName, no_ack=True)
- return self.client.queue(reply.consumer_tag)
+ if not "uniqueTag" in dir(self): self.uniqueTag = 1
+ else: self.uniqueTag += 1
+ consumer_tag = "tag" + str(self.uniqueTag)
+ self.channel.message_consume(queue=queueName,
destination=consumer_tag, no_ack=True)
+ return self.client.queue(consumer_tag)
def assertEmpty(self, queue):
"""Assert that the queue is empty"""
@@ -218,12 +221,12 @@
Publish to exchange and assert queue.get() returns the same message.
"""
body = self.uniqueString()
- self.channel.basic_publish(exchange=exchange,
- content=Content(body,
properties=properties),
- routing_key=routing_key)
+ self.channel.message_transfer(destination=exchange,
+ body=body,
application_headers=properties,
+ routing_key=routing_key)
msg = queue.get(timeout=1)
- self.assertEqual(body, msg.content.body)
- if (properties): self.assertEqual(properties, msg.content.properties)
+ self.assertEqual(body, msg.body)
+ if (properties): self.assertEqual(properties, msg.application_headers)
def assertPublishConsume(self, queue="", exchange="", routing_key="",
properties=None):
"""
Modified: incubator/qpid/trunk/qpid/python/tests/broker.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/broker.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/broker.py (original)
+++ incubator/qpid/trunk/qpid/python/tests/broker.py Fri Mar 16 15:54:11 2007
@@ -24,7 +24,7 @@
class BrokerTests(TestBase):
"""Tests for basic Broker functionality"""
- def test_amqp_basic_13(self):
+ def test_ack_and_no_ack(self):
"""
First, this test tries to receive a message with a no-ack
consumer. Second, this test tries to explicitely receive and
@@ -34,41 +34,44 @@
self.queue_declare(ch, queue = "myqueue")
# No ack consumer
- ctag = ch.basic_consume(queue = "myqueue", no_ack = True).consumer_tag
+ ctag = "tag1"
+ ch.message_consume(queue = "myqueue", destination = ctag, no_ack =
True)
body = "test no-ack"
- ch.basic_publish(routing_key = "myqueue", content = Content(body))
+ ch.message_transfer(routing_key = "myqueue", body = body)
msg = self.client.queue(ctag).get(timeout = 5)
- self.assert_(msg.content.body == body)
+ self.assert_(msg.body == body)
- # Acknowleding consumer
+ # Acknowledging consumer
self.queue_declare(ch, queue = "otherqueue")
- ctag = ch.basic_consume(queue = "otherqueue", no_ack =
False).consumer_tag
+ ctag = "tag2"
+ ch.message_consume(queue = "otherqueue", destination = ctag, no_ack =
False)
body = "test ack"
- ch.basic_publish(routing_key = "otherqueue", content = Content(body))
+ ch.message_transfer(routing_key = "otherqueue", body = body)
msg = self.client.queue(ctag).get(timeout = 5)
- ch.basic_ack(delivery_tag = msg.delivery_tag)
- self.assert_(msg.content.body == body)
+ msg.ok()
+ self.assert_(msg.body == body)
- def test_basic_delivery_immediate(self):
+ def test_simple_delivery_immediate(self):
"""
- Test basic message delivery where consume is issued before publish
+ Test simple message delivery where consume is issued before publish
"""
channel = self.channel
self.exchange_declare(channel, exchange="test-exchange", type="direct")
self.queue_declare(channel, queue="test-queue")
channel.queue_bind(queue="test-queue", exchange="test-exchange",
routing_key="key")
- reply = channel.basic_consume(queue="test-queue", no_ack=True)
- queue = self.client.queue(reply.consumer_tag)
+ consumer_tag = "tag1"
+ channel.message_consume(queue="test-queue", destination=consumer_tag,
no_ack=True)
+ queue = self.client.queue(consumer_tag)
body = "Immediate Delivery"
- channel.basic_publish(exchange="test-exchange", routing_key="key",
content=Content(body), immediate=True)
+ channel.message_transfer(destination="test-exchange",
routing_key="key", body=body, immediate=True)
msg = queue.get(timeout=5)
- self.assert_(msg.content.body == body)
+ self.assert_(msg.body == body)
# TODO: Ensure we fail if immediate=True and there's no consumer.
- def test_basic_delivery_queued(self):
+ def test_simple_delivery_queued(self):
"""
Test basic message delivery where publish is issued before consume
(i.e. requires queueing of the message)
@@ -78,11 +81,13 @@
self.queue_declare(channel, queue="test-queue")
channel.queue_bind(queue="test-queue", exchange="test-exchange",
routing_key="key")
body = "Queued Delivery"
- channel.basic_publish(exchange="test-exchange", routing_key="key",
content=Content(body))
- reply = channel.basic_consume(queue="test-queue", no_ack=True)
- queue = self.client.queue(reply.consumer_tag)
+ channel.message_transfer(destination="test-exchange",
routing_key="key", body=body)
+
+ consumer_tag = "tag1"
+ channel.message_consume(queue="test-queue", destination=consumer_tag,
no_ack=True)
+ queue = self.client.queue(consumer_tag)
msg = queue.get(timeout=5)
- self.assert_(msg.content.body == body)
+ self.assert_(msg.body == body)
def test_invalid_channel(self):
channel = self.client.channel(200)
Modified: incubator/qpid/trunk/qpid/python/tests/example.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/example.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/example.py (original)
+++ incubator/qpid/trunk/qpid/python/tests/example.py Fri Mar 16 15:54:11 2007
@@ -68,18 +68,18 @@
# has fields corresponding to the reply method fields, plus a content
# field that is filled if the reply includes content. In this case the
# interesting field is the consumer_tag.
- reply = channel.basic_consume(queue="test-queue")
+ channel.message_consume(queue="test-queue", destination="consumer_tag")
# We can use the Client.queue(...) method to access the queue
# corresponding to our consumer_tag.
- queue = self.client.queue(reply.consumer_tag)
+ queue = self.client.queue("consumer_tag")
# Now lets publish a message and see if our consumer gets it. To do
# this we need to import the Content class.
body = "Hello World!"
- channel.basic_publish(exchange="test",
- routing_key="key",
- content=Content(body))
+ channel.message_transfer(destination="test",
+ routing_key="key",
+ body = body)
# Now we'll wait for the message to arrive. We can use the timeout
# argument in case the server hangs. By default queue.get() will wait
@@ -87,8 +87,8 @@
msg = queue.get(timeout=10)
# And check that we got the right response with assertEqual
- self.assertEqual(body, msg.content.body)
+ self.assertEqual(body, msg.body)
# Now acknowledge the message.
- channel.basic_ack(msg.delivery_tag, True)
+ msg.ok()
Modified: incubator/qpid/trunk/qpid/python/tests/exchange.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/exchange.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/exchange.py (original)
+++ incubator/qpid/trunk/qpid/python/tests/exchange.py Fri Mar 16 15:54:11 2007
@@ -61,10 +61,10 @@
self.assertPublishGet(q, ex, "a.x.b.x")
self.assertPublishGet(q, ex, "a.x.x.b.x")
# Shouldn't match
- self.channel.basic_publish(exchange=ex, routing_key="a.b")
- self.channel.basic_publish(exchange=ex, routing_key="a.b.x.y")
- self.channel.basic_publish(exchange=ex, routing_key="x.a.b.x")
- self.channel.basic_publish(exchange=ex, routing_key="a.b")
+ self.channel.message_transfer(destination=ex, routing_key="a.b")
+ self.channel.message_transfer(destination=ex, routing_key="a.b.x.y")
+ self.channel.message_transfer(destination=ex, routing_key="x.a.b.x")
+ self.channel.message_transfer(destination=ex, routing_key="a.b")
self.assert_(q.empty())
def verifyHeadersExchange(self, ex):
@@ -73,8 +73,8 @@
self.channel.queue_bind(queue="q", exchange=ex, arguments={
"x-match":"all", "name":"fred" , "age":3} )
q = self.consume("q")
headers = {"name":"fred", "age":3}
- self.assertPublishGet(q, exchange=ex, properties={'headers':headers})
- self.channel.basic_publish(exchange=ex) # No headers, won't deliver
+ self.assertPublishGet(q, exchange=ex, properties=headers)
+ self.channel.message_transfer(destination=ex, body="") # No headers,
won't deliver
self.assertEmpty(q);
@@ -272,10 +272,10 @@
self.q = self.consume("q")
def myAssertPublishGet(self, headers):
- self.assertPublishGet(self.q, exchange="amq.match",
properties={'headers':headers})
+ self.assertPublishGet(self.q, exchange="amq.match", properties=headers)
def myBasicPublish(self, headers):
- self.channel.basic_publish(exchange="amq.match",
content=Content("foobar", properties={'headers':headers}))
+ self.channel.message_transfer(destination="amq.match", body="foobar",
application_headers=headers)
def testMatchAll(self):
self.channel.queue_bind(queue="q", exchange="amq.match", arguments={
'x-match':'all', "name":"fred", "age":3})
Modified: incubator/qpid/trunk/qpid/python/tests/queue.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/queue.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/queue.py (original)
+++ incubator/qpid/trunk/qpid/python/tests/queue.py Fri Mar 16 15:54:11 2007
@@ -33,9 +33,9 @@
channel.exchange_declare(exchange="test-exchange", type="direct")
channel.queue_declare(queue="test-queue", exclusive=True)
channel.queue_bind(queue="test-queue", exchange="test-exchange",
routing_key="key")
- channel.basic_publish(exchange="test-exchange", routing_key="key",
content=Content("one"))
- channel.basic_publish(exchange="test-exchange", routing_key="key",
content=Content("two"))
- channel.basic_publish(exchange="test-exchange", routing_key="key",
content=Content("three"))
+ channel.message_transfer(destination="test-exchange",
routing_key="key", body="one")
+ channel.message_transfer(destination="test-exchange",
routing_key="key", body="two")
+ channel.message_transfer(destination="test-exchange",
routing_key="key", body="three")
#check that the queue now reports 3 messages:
reply = channel.queue_declare(queue="test-queue")
@@ -48,9 +48,9 @@
self.assertEqual(0, reply.message_count)
#send a further message and consume it, ensuring that the other
messages are really gone
- channel.basic_publish(exchange="test-exchange", routing_key="key",
content=Content("four"))
- reply = channel.basic_consume(queue="test-queue", no_ack=True)
- queue = self.client.queue(reply.consumer_tag)
+ channel.message_transfer(destination="test-exchange",
routing_key="key", content=Content("four"))
+ channel.message_consume(queue="test-queue", destination="tag",
no_ack=True)
+ queue = self.client.queue("tag")
msg = queue.get(timeout=1)
self.assertEqual("four", msg.content.body)
@@ -153,15 +153,15 @@
def test_delete_simple(self):
"""
- Test basic queue deletion
+ Test core queue deletion behaviour
"""
channel = self.channel
#straight-forward case:
channel.queue_declare(queue="delete-me")
- channel.basic_publish(routing_key="delete-me", content=Content("a"))
- channel.basic_publish(routing_key="delete-me", content=Content("b"))
- channel.basic_publish(routing_key="delete-me", content=Content("c"))
+ channel.message_transfer(routing_key="delete-me", body="a")
+ channel.message_transfer(routing_key="delete-me", body="b")
+ channel.message_transfer(routing_key="delete-me", body="c")
reply = channel.queue_delete(queue="delete-me")
self.assertEqual(3, reply.message_count)
#check that it has gone be declaring passively
@@ -191,7 +191,7 @@
#create a queue and add a message to it (use default binding):
channel.queue_declare(queue="delete-me-2")
channel.queue_declare(queue="delete-me-2", passive="True")
- channel.basic_publish(routing_key="delete-me-2",
content=Content("message"))
+ channel.message_transfer(routing_key="delete-me-2", body="message")
#try to delete, but only if empty:
try:
@@ -205,11 +205,11 @@
channel.channel_open()
#empty queue:
- reply = channel.basic_consume(queue="delete-me-2", no_ack=True)
- queue = self.client.queue(reply.consumer_tag)
+ channel.message_consume(destination="consumer_tag",
queue="delete-me-2", no_ack=True)
+ queue = self.client.queue("consumer_tag")
msg = queue.get(timeout=1)
self.assertEqual("message", msg.content.body)
- channel.basic_cancel(consumer_tag=reply.consumer_tag)
+ channel.message_cancel(destination="consumer_tag")
#retry deletion on empty queue:
channel.queue_delete(queue="delete-me-2", if_empty="True")
@@ -230,7 +230,7 @@
#create a queue and register a consumer:
channel.queue_declare(queue="delete-me-3")
channel.queue_declare(queue="delete-me-3", passive="True")
- reply = channel.basic_consume(queue="delete-me-3", no_ack=True)
+ channel.message_consume(destination="consumer_tag",
queue="delete-me-3", no_ack=True)
#need new channel now:
channel2 = self.client.channel(2)
@@ -243,7 +243,7 @@
self.assertChannelException(406, e.args[0])
- channel.basic_cancel(consumer_tag=reply.consumer_tag)
+ channel.message_cancel(destination="consumer_tag")
channel.queue_delete(queue="delete-me-3", if_unused="True")
#check that it has gone by declaring passively:
try:
Modified: incubator/qpid/trunk/qpid/python/tests/testlib.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/testlib.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/testlib.py (original)
+++ incubator/qpid/trunk/qpid/python/tests/testlib.py Fri Mar 16 15:54:11 2007
@@ -49,7 +49,7 @@
def testAssertEmptyFail(self):
self.queue_declare(queue="full")
q = self.consume("full")
- self.channel.basic_publish(routing_key="full")
+ self.channel.message_transfer(routing_key="full", body="")
try:
self.assertEmpty(q);
self.fail("assertEmpty did not assert on non-empty queue")
Modified: incubator/qpid/trunk/qpid/python/tests/tx.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/tx.py?view=diff&rev=519171&r1=519170&r2=519171
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/tx.py (original)
+++ incubator/qpid/trunk/qpid/python/tests/tx.py Fri Mar 16 15:54:11 2007
@@ -37,22 +37,24 @@
#check results
for i in range(1, 5):
msg = queue_c.get(timeout=1)
- self.assertEqual("TxMessage %d" % i, msg.content.body)
+ self.assertEqual("TxMessage %d" % i, msg.body)
+ msg.ok()
msg = queue_b.get(timeout=1)
- self.assertEqual("TxMessage 6", msg.content.body)
+ self.assertEqual("TxMessage 6", msg.body)
+ msg.ok()
msg = queue_a.get(timeout=1)
- self.assertEqual("TxMessage 7", msg.content.body)
+ self.assertEqual("TxMessage 7", msg.body)
+ msg.ok()
for q in [queue_a, queue_b, queue_c]:
try:
extra = q.get(timeout=1)
- self.fail("Got unexpected message: " + extra.content.body)
+ self.fail("Got unexpected message: " + extra.body)
except Empty: None
#cleanup
- channel.basic_ack(delivery_tag=0, multiple=True)
channel.tx_commit()
def test_auto_rollback(self):
@@ -65,7 +67,7 @@
for q in [queue_a, queue_b, queue_c]:
try:
extra = q.get(timeout=1)
- self.fail("Got unexpected message: " + extra.content.body)
+ self.fail("Got unexpected message: " + extra.body)
except Empty: None
channel.tx_rollback()
@@ -73,22 +75,24 @@
#check results
for i in range(1, 5):
msg = queue_a.get(timeout=1)
- self.assertEqual("Message %d" % i, msg.content.body)
+ self.assertEqual("Message %d" % i, msg.body)
+ msg.ok()
msg = queue_b.get(timeout=1)
- self.assertEqual("Message 6", msg.content.body)
+ self.assertEqual("Message 6", msg.body)
+ msg.ok()
msg = queue_c.get(timeout=1)
- self.assertEqual("Message 7", msg.content.body)
+ self.assertEqual("Message 7", msg.body)
+ msg.ok()
for q in [queue_a, queue_b, queue_c]:
try:
extra = q.get(timeout=1)
- self.fail("Got unexpected message: " + extra.content.body)
+ self.fail("Got unexpected message: " + extra.body)
except Empty: None
#cleanup
- channel.basic_ack(delivery_tag=0, multiple=True)
channel.tx_commit()
def test_rollback(self):
@@ -101,7 +105,7 @@
for q in [queue_a, queue_b, queue_c]:
try:
extra = q.get(timeout=1)
- self.fail("Got unexpected message: " + extra.content.body)
+ self.fail("Got unexpected message: " + extra.body)
except Empty: None
channel.tx_rollback()
@@ -109,22 +113,24 @@
#check results
for i in range(1, 5):
msg = queue_a.get(timeout=1)
- self.assertEqual("Message %d" % i, msg.content.body)
+ self.assertEqual("Message %d" % i, msg.body)
+ msg.ok()
msg = queue_b.get(timeout=1)
- self.assertEqual("Message 6", msg.content.body)
+ self.assertEqual("Message 6", msg.body)
+ msg.ok()
msg = queue_c.get(timeout=1)
- self.assertEqual("Message 7", msg.content.body)
+ self.assertEqual("Message 7", msg.body)
+ msg.ok()
for q in [queue_a, queue_b, queue_c]:
try:
extra = q.get(timeout=1)
- self.fail("Got unexpected message: " + extra.content.body)
+ self.fail("Got unexpected message: " + extra.body)
except Empty: None
#cleanup
- channel.basic_ack(delivery_tag=0, multiple=True)
channel.tx_commit()
def perform_txn_work(self, channel, name_a, name_b, name_c):
@@ -144,66 +150,38 @@
channel.queue_bind(queue=name_c, exchange="amq.topic",
routing_key=topic)
for i in range(1, 5):
- channel.basic_publish(routing_key=name_a, content=Content("Message
%d" % i))
+ channel.message_transfer(routing_key=name_a, body="Message %d" % i)
- channel.basic_publish(routing_key=key, exchange="amq.direct",
content=Content("Message 6"))
- channel.basic_publish(routing_key=topic, exchange="amq.topic",
content=Content("Message 7"))
+ channel.message_transfer(routing_key=key, destination="amq.direct",
body="Message 6")
+ channel.message_transfer(routing_key=topic, destination="amq.topic",
body="Message 7")
channel.tx_select()
#consume and ack messages
- sub_a = channel.basic_consume(queue=name_a, no_ack=False)
- queue_a = self.client.queue(sub_a.consumer_tag)
+ channel.message_consume(queue=name_a, destination="sub_a",
no_ack=False)
+ queue_a = self.client.queue("sub_a")
for i in range(1, 5):
msg = queue_a.get(timeout=1)
- self.assertEqual("Message %d" % i, msg.content.body)
- channel.basic_ack(delivery_tag=msg.delivery_tag, multiple=True)
+ self.assertEqual("Message %d" % i, msg.body)
+ msg.ok()
- sub_b = channel.basic_consume(queue=name_b, no_ack=False)
- queue_b = self.client.queue(sub_b.consumer_tag)
+ channel.message_consume(queue=name_b, destination="sub_b",
no_ack=False)
+ queue_b = self.client.queue("sub_b")
msg = queue_b.get(timeout=1)
- self.assertEqual("Message 6", msg.content.body)
- channel.basic_ack(delivery_tag=msg.delivery_tag)
+ self.assertEqual("Message 6", msg.body)
+ msg.ok()
- sub_c = channel.basic_consume(queue=name_c, no_ack=False)
- queue_c = self.client.queue(sub_c.consumer_tag)
+ sub_c = channel.message_consume(queue=name_c, destination="sub_c",
no_ack=False)
+ queue_c = self.client.queue("sub_c")
msg = queue_c.get(timeout=1)
- self.assertEqual("Message 7", msg.content.body)
- channel.basic_ack(delivery_tag=msg.delivery_tag)
+ self.assertEqual("Message 7", msg.body)
+ msg.ok()
#publish messages
for i in range(1, 5):
- channel.basic_publish(routing_key=topic, exchange="amq.topic",
content=Content("TxMessage %d" % i))
+ channel.message_transfer(routing_key=topic,
destination="amq.topic", body="TxMessage %d" % i)
- channel.basic_publish(routing_key=key, exchange="amq.direct",
content=Content("TxMessage 6"))
- channel.basic_publish(routing_key=name_a, content=Content("TxMessage
7"))
+ channel.message_transfer(routing_key=key, destination="amq.direct",
body="TxMessage 6")
+ channel.message_transfer(routing_key=name_a, body="TxMessage 7")
return queue_a, queue_b, queue_c
-
- def test_commit_overlapping_acks(self):
- """
- Test that logically 'overlapping' acks do not cause errors on commit
- """
- channel = self.channel
- channel.queue_declare(queue="commit-overlapping", exclusive=True)
- for i in range(1, 10):
- channel.basic_publish(routing_key="commit-overlapping",
content=Content("Message %d" % i))
-
-
- channel.tx_select()
-
- sub = channel.basic_consume(queue="commit-overlapping", no_ack=False)
- queue = self.client.queue(sub.consumer_tag)
- for i in range(1, 10):
- msg = queue.get(timeout=1)
- self.assertEqual("Message %d" % i, msg.content.body)
- if i in [3, 6, 10]:
- channel.basic_ack(delivery_tag=msg.delivery_tag)
-
- channel.tx_commit()
-
- #check all have been acked:
- try:
- extra = queue.get(timeout=1)
- self.fail("Got unexpected message: " + extra.content.body)
- except Empty: None
