You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2007/03/19 20:39:56 UTC
svn commit: r520061 - in /incubator/qpid/trunk/qpid: ./
python/qpid/client.py python/qpid/codec.py python/qpid/peer.py
python/qpid/reference.py python/qpid/testlib.py python/tests/message.py
Author: aconway
Date: Mon Mar 19 12:39:55 2007
New Revision: 520061
URL: http://svn.apache.org/viewvc?view=rev&rev=520061
Log:
Merged revisions 504590 via svnmerge from
https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9
........
r504590 | gsim | 2007-02-07 10:36:01 -0500 (Wed, 07 Feb 2007) | 6 lines
Added support for receiving and sending of references
Added asynchronous mode to channels (responses can be tracked via a future, rather than blocking on each request)
Added ability to override server suggested connection tune params
Added two tests for reference functionality (more to follow)
........
Added:
incubator/qpid/trunk/qpid/python/qpid/reference.py
- copied unchanged from r504590, incubator/qpid/branches/qpid.0-9/python/qpid/reference.py
Modified:
incubator/qpid/trunk/qpid/ (props changed)
incubator/qpid/trunk/qpid/python/qpid/client.py
incubator/qpid/trunk/qpid/python/qpid/codec.py
incubator/qpid/trunk/qpid/python/qpid/peer.py
incubator/qpid/trunk/qpid/python/qpid/testlib.py
incubator/qpid/trunk/qpid/python/tests/message.py
Propchange: incubator/qpid/trunk/qpid/
------------------------------------------------------------------------------
--- svnmerge-integrated (original)
+++ svnmerge-integrated Mon Mar 19 12:39:55 2007
@@ -1 +1 @@
-/incubator/qpid/branches/M2:0-519912,519933 /incubator/qpid/branches/qpid.0-9:1-492620,496593,497277,500305,501022,501025,501082,501143,501586,502176-502208,502210-502766,502768-503671,503673-503858
+/incubator/qpid/branches/M2:0-519912,519933 /incubator/qpid/branches/qpid.0-9:1-492620,496593,497277,500305,501022,501025,501082,501143,501586,502176-502208,502210-502766,502768-503671,503673-503858,504590
Modified: incubator/qpid/trunk/qpid/python/qpid/client.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/client.py?view=diff&rev=520061&r1=520060&r2=520061
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/client.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/client.py Mon Mar 19 12:39:55 2007
@@ -28,6 +28,7 @@
from connection import Connection, Frame, connect
from spec import load
from queue import Queue
+from reference import ReferenceId, References
class Client:
@@ -69,13 +70,14 @@
self.lock.release()
return q
- def start(self, response, mechanism="AMQPLAIN", locale="en_US"):
+ def start(self, response, mechanism="AMQPLAIN", locale="en_US", tune_params=None):
self.mechanism = mechanism
self.response = response
self.locale = locale
+ self.tune_params = tune_params
self.conn = Connection(connect(self.host, self.port), self.spec)
- self.peer = Peer(self.conn, ClientDelegate(self))
+ self.peer = Peer(self.conn, ClientDelegate(self), self.opened)
self.conn.init()
self.peer.start()
@@ -85,6 +87,9 @@
def channel(self, id):
return self.peer.channel(id)
+ def opened(self, ch):
+ ch.references = References()
+
class ClientDelegate(Delegate):
def __init__(self, client):
@@ -97,8 +102,28 @@
locale=self.client.locale)
def connection_tune(self, ch, msg):
- msg.tune_ok(*msg.frame.args)
+ if self.client.tune_params:
+ #todo: just override the params, i.e. don't require them
+ # all to be included in tune_params
+ msg.tune_ok(**self.client.tune_params)
+ else:
+ msg.tune_ok(*msg.frame.args)
self.client.started.set()
+
+ def message_transfer(self, ch, msg):
+ if isinstance(msg.body, ReferenceId):
+ self.client.queue(msg.destination).put(ch.references.get(msg.body.id))
+ else:
+ self.client.queue(msg.destination).put(msg)
+
+ def message_open(self, ch, msg):
+ ch.references.open(msg.reference)
+
+ def message_close(self, ch, msg):
+ ch.references.close(msg.reference)
+
+ def message_append(self, ch, msg):
+ ch.references.get(msg.reference).append(msg.bytes)
def basic_deliver(self, ch, msg):
self.client.queue(msg.consumer_tag).put(msg)
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=520061&r1=520060&r2=520061
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/codec.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/codec.py Mon Mar 19 12:39:55 2007
@@ -26,6 +26,7 @@
from cStringIO import StringIO
from struct import *
+from reference import ReferenceId
class EOF(Exception):
pass
@@ -195,14 +196,24 @@
return self.decode_longlong()
def encode_content(self, s):
- # XXX
- self.encode_octet(0)
- self.encode_longstr(s)
-
- def decode_content(self):
- # XXX
- self.decode_octet()
- return self.decode_longstr()
+ # content can be passed as a string in which case it is assumed to
+ # be inline data, or as an instance of ReferenceId indicating it is
+ # a reference id
+ if isinstance(s, ReferenceId):
+ self.encode_octet(1)
+ self.encode_longstr(s.id)
+ else:
+ self.encode_octet(0)
+ self.encode_longstr(s)
+
+ def decode_content(self):
+ # return a string for inline data and a ReferenceId instance for
+ # references
+ type = self.decode_octet()
+ if type == 0:
+ return self.decode_longstr()
+ else:
+ return ReferenceId(self.decode_longstr())
def test(type, value):
if isinstance(value, (list, tuple)):
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=520061&r1=520060&r2=520061
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/peer.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/peer.py Mon Mar 19 12:39:55 2007
@@ -50,13 +50,14 @@
class Peer:
- def __init__(self, conn, delegate):
+ def __init__(self, conn, delegate, channel_callback=None):
self.conn = conn
self.delegate = delegate
self.outgoing = Queue(0)
self.work = Queue(0)
self.channels = {}
self.lock = thread.allocate_lock()
+ self.channel_callback = channel_callback #notified when channels are created
def channel(self, id):
self.lock.acquire()
@@ -66,6 +67,8 @@
except KeyError:
ch = Channel(id, self.outgoing, self.conn.spec)
self.channels[id] = ch
+ if self.channel_callback:
+ self.channel_callback(ch)
finally:
self.lock.release()
return ch
@@ -177,6 +180,7 @@
# XXX: better switch
self.reliable = False
+ self.synchronous = True
def close(self, reason):
if self.closed:
@@ -238,6 +242,12 @@
content = kwargs.pop("content", None)
frame = Method(type, type.arguments(*args, **kwargs))
if self.reliable:
+ if not self.synchronous:
+ future = Future()
+ self.request(frame, future.put_response, content)
+ if not frame.method.responses: return None
+ else: return future
+
self.request(frame, self.queue_response, content)
if not frame.method.responses:
return None
@@ -304,3 +314,18 @@
buf.write(content)
read += len(content)
return Content(buf.getvalue(), children, header.properties.copy())
+
+class Future:
+ def __init__(self):
+ self.completed = threading.Event()
+
+ def put_response(self, channel, response):
+ self.response = response
+ self.completed.set()
+
+ def get_response(self, timeout=None):
+ self.completed.wait(timeout)
+ return self.response
+
+ def is_complete(self):
+ return self.completed.isSet()
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=520061&r1=520060&r2=520061
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/testlib.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/testlib.py Mon Mar 19 12:39:55 2007
@@ -145,7 +145,7 @@
print "======================================="
return result.wasSuccessful()
- def connect(self, host=None, port=None, spec=None, user=None, password=None):
+ def connect(self, host=None, port=None, spec=None, user=None, password=None, tune_params=None):
"""Connect to the broker, returns a qpid.client.Client"""
host = host or self.host
port = port or self.port
@@ -153,7 +153,7 @@
user = user or self.user
password = password or self.password
client = qpid.client.Client(host, port, spec)
- client.start({"LOGIN": user, "PASSWORD": password})
+ client.start({"LOGIN": user, "PASSWORD": password}, tune_params=tune_params)
return client
Modified: incubator/qpid/trunk/qpid/python/tests/message.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/message.py?view=diff&rev=520061&r1=520060&r2=520061
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/message.py (original)
+++ incubator/qpid/trunk/qpid/python/tests/message.py Mon Mar 19 12:39:55 2007
@@ -20,6 +20,7 @@
from qpid.queue import Empty
from qpid.content import Content
from qpid.testlib import testrunner, TestBase
+from qpid.reference import Reference, ReferenceId
class MessageTests(TestBase):
"""Tests for 'methods' on the amqp message 'class'"""
@@ -413,3 +414,66 @@
reply = channel.message_get(no_ack=True)
self.assertEqual(reply.method.klass.name, "message")
self.assertEqual(reply.method.name, "get-empty")
+
+ def test_reference_simple(self):
+ """
+ Test basic ability to handle references
+ """
+ channel = self.channel
+ channel.queue_declare(queue="ref_queue", exclusive=True)
+ channel.message_consume(queue="ref_queue", destination="c1")
+ queue = self.client.queue("c1")
+
+ refId = "myref"
+ channel.message_open(reference=refId)
+ channel.message_append(reference=refId, bytes="abcd")
+ channel.synchronous = False
+ ack = channel.message_transfer(routing_key="ref_queue", body=ReferenceId(refId))
+ channel.synchronous = True
+
+ channel.message_append(reference=refId, bytes="efgh")
+ channel.message_append(reference=refId, bytes="ijkl")
+ channel.message_close(reference=refId)
+
+ #first, wait for the ok for the transfer
+ ack.get_response(timeout=1)
+
+ msg = queue.get(timeout=1)
+ if isinstance(msg, Reference):
+ #should we force broker to deliver as reference by frame
+ #size limit? or test that separately? for compliance,
+ #allowing either seems best for now...
+ data = msg.get_complete()
+ else:
+ data = msg.body
+ self.assertEquals("abcdefghijkl", data)
+
+
+ def test_reference_large(self):
+ """
+ Test basic ability to handle references whose content exceeds max frame size
+ """
+ channel = self.channel
+ self.queue_declare(queue="ref_queue")
+
+ #generate a big data string (> max frame size of consumer):
+ data = "0123456789"
+ for i in range(0, 10):
+ data += data
+ #send it inline
+ channel.synchronous = False
+ ack = channel.message_transfer(routing_key="ref_queue", body=data)
+ channel.synchronous = True
+ #first, wait for the ok for the transfer
+ ack.get_response(timeout=1)
+
+ #create a new connection for consumer, with specific max frame size (< data)
+ other = self.connect(tune_params={"channel_max":10, "frame_max":5120, "heartbeat":0})
+ ch2 = other.channel(1)
+ ch2.channel_open()
+ ch2.message_consume(queue="ref_queue", destination="c1")
+ queue = other.queue("c1")
+
+ msg = queue.get(timeout=1)
+ self.assertTrue(isinstance(msg, Reference))
+ self.assertEquals(data, msg.get_complete())