You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2007/01/26 19:51:22 UTC
svn commit: r500305 - in /incubator/qpid/branches/qpid.0-9/python:
java_failing.txt qpid/codec.py qpid/connection.py qpid/peer.py
qpid/testlib.py tests/broker.py tests/example.py tests/exchange.py
tests/queue.py tests/testlib.py tests/tx.py
Author: gsim
Date: Fri Jan 26 10:51:21 2007
New Revision: 500305
URL: http://svn.apache.org/viewvc?view=rev&rev=500305
Log:
Updates to use message class in place of basic.
Modified:
incubator/qpid/branches/qpid.0-9/python/java_failing.txt
incubator/qpid/branches/qpid.0-9/python/qpid/codec.py
incubator/qpid/branches/qpid.0-9/python/qpid/connection.py
incubator/qpid/branches/qpid.0-9/python/qpid/peer.py
incubator/qpid/branches/qpid.0-9/python/qpid/testlib.py
incubator/qpid/branches/qpid.0-9/python/tests/broker.py
incubator/qpid/branches/qpid.0-9/python/tests/example.py
incubator/qpid/branches/qpid.0-9/python/tests/exchange.py
incubator/qpid/branches/qpid.0-9/python/tests/queue.py
incubator/qpid/branches/qpid.0-9/python/tests/testlib.py
incubator/qpid/branches/qpid.0-9/python/tests/tx.py
Modified: incubator/qpid/branches/qpid.0-9/python/java_failing.txt
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/java_failing.txt?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/java_failing.txt (original)
+++ incubator/qpid/branches/qpid.0-9/python/java_failing.txt Fri Jan 26 10:51:21 2007
@@ -1,3 +1,5 @@
+tests.basic.BasicTests.test_qos_prefetch_count
+tests.basic.BasicTests.test_ack
tests.basic.BasicTests.test_cancel
tests.basic.BasicTests.test_consume_exclusive
tests.basic.BasicTests.test_consume_no_local
@@ -6,17 +8,7 @@
tests.basic.BasicTests.test_get
tests.basic.BasicTests.test_qos_prefetch_size
tests.basic.BasicTests.test_recover_requeue
-tests.exchange.ExchangeTests
-tests.exchange.DefaultExchangeRuleTests.testDefaultExchange
-tests.exchange.HeadersExchangeTests.testMatchAll
-tests.exchange.HeadersExchangeTests.testMatchAny
-tests.exchange.RecommendedTypesRuleTests.testDirect
-tests.exchange.RecommendedTypesRuleTests.testFanout
-tests.exchange.RecommendedTypesRuleTests.testHeaders
tests.exchange.RecommendedTypesRuleTests.testTopic
-tests.exchange.RequiredInstancesRuleTests.testAmqDirect
-tests.exchange.RequiredInstancesRuleTests.testAmqFanOut
-tests.exchange.RequiredInstancesRuleTests.testAmqMatch
tests.exchange.RequiredInstancesRuleTests.testAmqTopic
tests.queue.QueueTests.test_declare_exclusive
tests.queue.QueueTests.test_declare_passive
@@ -24,6 +16,4 @@
tests.queue.QueueTests.test_delete_ifunused
tests.queue.QueueTests.test_delete_simple
tests.queue.QueueTests.test_purge
-tests.queue.QueueTests.test_bind
tests.testlib.TestBaseTest.testMessageProperties
-tests.broker.BrokerTests.test_invalid_channel
Modified: incubator/qpid/branches/qpid.0-9/python/qpid/codec.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/qpid/codec.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/qpid/codec.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/qpid/codec.py Fri Jan 26 10:51:21 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/branches/qpid.0-9/python/qpid/connection.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/qpid/connection.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/qpid/connection.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/qpid/connection.py Fri Jan 26 10:51:21 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/branches/qpid.0-9/python/qpid/peer.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/qpid/peer.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/qpid/peer.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/qpid/peer.py Fri Jan 26 10:51:21 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/branches/qpid.0-9/python/qpid/testlib.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/qpid/testlib.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/qpid/testlib.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/qpid/testlib.py Fri Jan 26 10:51:21 2007
@@ -80,7 +80,7 @@
def __init__(self):
# Defaults
self.setBroker("localhost")
- self.spec = "../specs/amqp.0-8.xml"
+ self.spec = "../specs/amqp.0-9.xml"
self.verbose = 1
self.ignore = []
@@ -195,8 +195,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"""
@@ -210,12 +213,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/branches/qpid.0-9/python/tests/broker.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/tests/broker.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/tests/broker.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/tests/broker.py Fri Jan 26 10:51:21 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,21 +81,23 @@
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):
- other = self.connect()
- channel = other.channel(200)
+ channel = self.client.channel(200)
try:
channel.queue_declare(exclusive=True)
self.fail("Expected error on queue_declare for invalid channel")
except Closed, e:
self.assertConnectionException(504, e.args[0])
+ def test_closed_channel(self):
channel = self.client.channel(200)
channel.channel_open()
channel.channel_close()
Modified: incubator/qpid/branches/qpid.0-9/python/tests/example.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/tests/example.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/tests/example.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/tests/example.py Fri Jan 26 10:51:21 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/branches/qpid.0-9/python/tests/exchange.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/tests/exchange.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/tests/exchange.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/tests/exchange.py Fri Jan 26 10:51:21 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/branches/qpid.0-9/python/tests/queue.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/tests/queue.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/tests/queue.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/tests/queue.py Fri Jan 26 10:51:21 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)
@@ -152,16 +152,15 @@
def test_delete_simple(self):
"""
- Test basic queue deletion
+ Test core queue deletion behaviour
"""
- channel = self.client.channel(1)
- channel.channel_open()
+ 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
@@ -186,13 +185,12 @@
"""
Test that if_empty field of queue_delete is honoured
"""
- channel = self.client.channel(1)
- channel.channel_open()
+ channel = self.channel
#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:
@@ -206,11 +204,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")
@@ -226,13 +224,12 @@
"""
Test that if_unused field of queue_delete is honoured
"""
- channel = self.client.channel(1)
- channel.channel_open()
+ channel = self.channel
#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)
@@ -245,7 +242,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/branches/qpid.0-9/python/tests/testlib.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/tests/testlib.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/tests/testlib.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/tests/testlib.py Fri Jan 26 10:51:21 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/branches/qpid.0-9/python/tests/tx.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/python/tests/tx.py?view=diff&rev=500305&r1=500304&r2=500305
==============================================================================
--- incubator/qpid/branches/qpid.0-9/python/tests/tx.py (original)
+++ incubator/qpid/branches/qpid.0-9/python/tests/tx.py Fri Jan 26 10:51:21 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