qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject [2/3] qpid-dispatch git commit: NO-JIRA: Remove proton_future monkey-patch, ok since proton 0.10.
Date Tue, 14 Jun 2016 15:58:02 GMT
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/1c3c1f5b/python/qpid_dispatch_internal/proton_future/__init__.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/proton_future/__init__.py b/python/qpid_dispatch_internal/proton_future/__init__.py
deleted file mode 100644
index 8d19fa3..0000000
--- a/python/qpid_dispatch_internal/proton_future/__init__.py
+++ /dev/null
@@ -1,3908 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#   http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-#
-
-"""
-The proton module defines a suite of APIs that implement the AMQP 1.0
-protocol.
-
-The proton APIs consist of the following classes:
-
- - L{Messenger} -- A messaging endpoint.
- - L{Message}   -- A class for creating and/or accessing AMQP message content.
- - L{Data}      -- A class for creating and/or accessing arbitrary AMQP encoded
-                  data.
-
-"""
-
-from cproton import *
-
-import weakref, re, socket
-try:
-  import uuid
-except ImportError:
-  """
-  No 'native' UUID support.  Provide a very basic UUID type that is a compatible subset of the uuid type provided by more modern python releases.
-  """
-  import struct
-  class uuid:
-    class UUID:
-      def __init__(self, hex=None, bytes=None):
-        if [hex, bytes].count(None) != 1:
-          raise TypeError("need one of hex or bytes")
-        if bytes is not None:
-          self.bytes = bytes
-        elif hex is not None:
-          fields=hex.split("-")
-          fields[4:5] = [fields[4][:4], fields[4][4:]]
-          self.bytes = struct.pack("!LHHHHL", *[int(x,16) for x in fields])
-
-      def __cmp__(self, other):
-        if isinstance(other, uuid.UUID):
-          return cmp(self.bytes, other.bytes)
-        else:
-          return -1
-
-      def __str__(self):
-        return "%08x-%04x-%04x-%04x-%04x%08x" % struct.unpack("!LHHHHL", self.bytes)
-
-      def __repr__(self):
-        return "UUID(%r)" % str(self)
-
-      def __hash__(self):
-        return self.bytes.__hash__()
-
-  import os, random, socket, time
-  rand = random.Random()
-  rand.seed((os.getpid(), time.time(), socket.gethostname()))
-  def random_uuid():
-    bytes = [rand.randint(0, 255) for i in xrange(16)]
-
-    # From RFC4122, the version bits are set to 0100
-    bytes[7] &= 0x0F
-    bytes[7] |= 0x40
-
-    # From RFC4122, the top two bits of byte 8 get set to 01
-    bytes[8] &= 0x3F
-    bytes[8] |= 0x80
-    return "".join(map(chr, bytes))
-
-  def uuid4():
-    return uuid.UUID(bytes=random_uuid())
-
-def generate_uuid():
-  return uuid.uuid4()
-
-try:
-  bytes()
-except NameError:
-  bytes = str
-
-VERSION_MAJOR = PN_VERSION_MAJOR
-VERSION_MINOR = PN_VERSION_MINOR
-API_LANGUAGE = "C"
-IMPLEMENTATION_LANGUAGE = "C"
-
-class Constant(object):
-
-  def __init__(self, name):
-    self.name = name
-
-  def __repr__(self):
-    return self.name
-
-class ProtonException(Exception):
-  """
-  The root of the proton exception hierarchy. All proton exception
-  classes derive from this exception.
-  """
-  pass
-
-class Timeout(ProtonException):
-  """
-  A timeout exception indicates that a blocking operation has timed
-  out.
-  """
-  pass
-
-class Interrupt(ProtonException):
-  """
-  An interrupt exception indicaes that a blocking operation was interrupted.
-  """
-  pass
-
-class MessengerException(ProtonException):
-  """
-  The root of the messenger exception hierarchy. All exceptions
-  generated by the messenger class derive from this exception.
-  """
-  pass
-
-class MessageException(ProtonException):
-  """
-  The MessageException class is the root of the message exception
-  hierarhcy. All exceptions generated by the Message class derive from
-  this exception.
-  """
-  pass
-
-EXCEPTIONS = {
-  PN_TIMEOUT: Timeout,
-  PN_INTR: Interrupt
-  }
-
-PENDING = Constant("PENDING")
-ACCEPTED = Constant("ACCEPTED")
-REJECTED = Constant("REJECTED")
-RELEASED = Constant("RELEASED")
-ABORTED = Constant("ABORTED")
-SETTLED = Constant("SETTLED")
-
-STATUSES = {
-  PN_STATUS_ABORTED: ABORTED,
-  PN_STATUS_ACCEPTED: ACCEPTED,
-  PN_STATUS_REJECTED: REJECTED,
-  PN_STATUS_RELEASED: RELEASED,
-  PN_STATUS_PENDING: PENDING,
-  PN_STATUS_SETTLED: SETTLED,
-  PN_STATUS_UNKNOWN: None
-  }
-
-AUTOMATIC = Constant("AUTOMATIC")
-MANUAL = Constant("MANUAL")
-
-class Messenger(object):
-  """
-  The L{Messenger} class defines a high level interface for sending
-  and receiving L{Messages<Message>}. Every L{Messenger} contains a
-  single logical queue of incoming messages and a single logical queue
-  of outgoing messages. These messages in these queues may be destined
-  for, or originate from, a variety of addresses.
-
-  The messenger interface is single-threaded.  All methods
-  except one (L{interrupt}) are intended to be used from within
-  the messenger thread.
-
-
-  Address Syntax
-  ==============
-
-  An address has the following form::
-
-    [ amqp[s]:// ] [user[:password]@] domain [/[name]]
-
-  Where domain can be one of::
-
-    host | host:port | ip | ip:port | name
-
-  The following are valid examples of addresses:
-
-   - example.org
-   - example.org:1234
-   - amqp://example.org
-   - amqps://example.org
-   - example.org/incoming
-   - amqps://example.org/outgoing
-   - amqps://fred:trustno1@example.org
-   - 127.0.0.1:1234
-   - amqps://127.0.0.1:1234
-
-  Sending & Receiving Messages
-  ============================
-
-  The L{Messenger} class works in conjuction with the L{Message} class. The
-  L{Message} class is a mutable holder of message content.
-
-  The L{put} method copies its L{Message} to the outgoing queue, and may
-  send queued messages if it can do so without blocking.  The L{send}
-  method blocks until it has sent the requested number of messages,
-  or until a timeout interrupts the attempt.
-
-
-    >>> message = Message()
-    >>> for i in range(3):
-    ...   message.address = "amqp://host/queue"
-    ...   message.subject = "Hello World %i" % i
-    ...   messenger.put(message)
-    >>> messenger.send()
-
-  Similarly, the L{recv} method receives messages into the incoming
-  queue, and may block as it attempts to receive the requested number
-  of messages,  or until timeout is reached. It may receive fewer
-  than the requested number.  The L{get} method pops the
-  eldest L{Message} off the incoming queue and copies it into the L{Message}
-  object that you supply.  It will not block.
-
-
-    >>> message = Message()
-    >>> messenger.recv(10):
-    >>> while messenger.incoming > 0:
-    ...   messenger.get(message)
-    ...   print message.subject
-    Hello World 0
-    Hello World 1
-    Hello World 2
-
-  The blocking flag allows you to turn off blocking behavior entirely,
-  in which case L{send} and L{recv} will do whatever they can without
-  blocking, and then return.  You can then look at the number
-  of incoming and outgoing messages to see how much outstanding work
-  still remains.
-  """
-
-  def __init__(self, name=None):
-    """
-    Construct a new L{Messenger} with the given name. The name has
-    global scope. If a NULL name is supplied, a UUID based name will
-    be chosen.
-
-    @type name: string
-    @param name: the name of the messenger or None
-
-    """
-    self._mng = pn_messenger(name)
-    self._selectables = {}
-
-  def __del__(self):
-    """
-    Destroy the L{Messenger}.  This will close all connections that
-    are managed by the L{Messenger}.  Call the L{stop} method before
-    destroying the L{Messenger}.
-    """
-    if hasattr(self, "_mng"):
-      pn_messenger_free(self._mng)
-      del self._mng
-
-  def _check(self, err):
-    if err < 0:
-      if (err == PN_INPROGRESS):
-        return
-      exc = EXCEPTIONS.get(err, MessengerException)
-      raise exc("[%s]: %s" % (err, pn_error_text(pn_messenger_error(self._mng))))
-    else:
-      return err
-
-  @property
-  def name(self):
-    """
-    The name of the L{Messenger}.
-    """
-    return pn_messenger_name(self._mng)
-
-  def _get_certificate(self):
-    return pn_messenger_get_certificate(self._mng)
-
-  def _set_certificate(self, value):
-    self._check(pn_messenger_set_certificate(self._mng, value))
-
-  certificate = property(_get_certificate, _set_certificate,
-                         doc="""
-Path to a certificate file for the L{Messenger}. This certificate is
-used when the L{Messenger} accepts or establishes SSL/TLS connections.
-This property must be specified for the L{Messenger} to accept
-incoming SSL/TLS connections and to establish client authenticated
-outgoing SSL/TLS connection. Non client authenticated outgoing SSL/TLS
-connections do not require this property.
-""")
-
-  def _get_private_key(self):
-    return pn_messenger_get_private_key(self._mng)
-
-  def _set_private_key(self, value):
-    self._check(pn_messenger_set_private_key(self._mng, value))
-
-  private_key = property(_get_private_key, _set_private_key,
-                         doc="""
-Path to a private key file for the L{Messenger's<Messenger>}
-certificate. This property must be specified for the L{Messenger} to
-accept incoming SSL/TLS connections and to establish client
-authenticated outgoing SSL/TLS connection. Non client authenticated
-SSL/TLS connections do not require this property.
-""")
-
-  def _get_password(self):
-    return pn_messenger_get_password(self._mng)
-
-  def _set_password(self, value):
-    self._check(pn_messenger_set_password(self._mng, value))
-
-  password = property(_get_password, _set_password,
-                      doc="""
-This property contains the password for the L{Messenger.private_key}
-file, or None if the file is not encrypted.
-""")
-
-  def _get_trusted_certificates(self):
-    return pn_messenger_get_trusted_certificates(self._mng)
-
-  def _set_trusted_certificates(self, value):
-    self._check(pn_messenger_set_trusted_certificates(self._mng, value))
-
-  trusted_certificates = property(_get_trusted_certificates,
-                                  _set_trusted_certificates,
-                                  doc="""
-A path to a database of trusted certificates for use in verifying the
-peer on an SSL/TLS connection. If this property is None, then the peer
-will not be verified.
-""")
-
-  def _get_timeout(self):
-    t = pn_messenger_get_timeout(self._mng)
-    if t == -1:
-      return None
-    else:
-      return millis2secs(t)
-
-  def _set_timeout(self, value):
-    if value is None:
-      t = -1
-    else:
-      t = secs2millis(value)
-    self._check(pn_messenger_set_timeout(self._mng, t))
-
-  timeout = property(_get_timeout, _set_timeout,
-                     doc="""
-The timeout property contains the default timeout for blocking
-operations performed by the L{Messenger}.
-""")
-
-  def _is_blocking(self):
-    return pn_messenger_is_blocking(self._mng)
-
-  def _set_blocking(self, b):
-    self._check(pn_messenger_set_blocking(self._mng, b))
-
-  blocking = property(_is_blocking, _set_blocking,
-                      doc="""
-Enable or disable blocking behavior during L{Message} sending
-and receiving.  This affects every blocking call, with the
-exception of L{work}.  Currently, the affected calls are
-L{send}, L{recv}, and L{stop}.
-""")
-
-  def _is_passive(self):
-    return pn_messenger_is_passive(self._mng)
-
-  def _set_passive(self, b):
-    self._check(pn_messenger_set_passive(self._mng, b))
-
-  passive = property(_is_passive, _set_passive,
-                      doc="""
-When passive is set to true, Messenger will not attempt to perform I/O
-internally. In this mode it is necessary to use the selectables API to
-drive any I/O needed to perform requested actions. In this mode
-Messenger will never block.
-""")
-
-  def _get_incoming_window(self):
-    return pn_messenger_get_incoming_window(self._mng)
-
-  def _set_incoming_window(self, window):
-    self._check(pn_messenger_set_incoming_window(self._mng, window))
-
-  incoming_window = property(_get_incoming_window, _set_incoming_window,
-                             doc="""
-The incoming tracking window for the messenger. The messenger will
-track the remote status of this many incoming deliveries after they
-have been accepted or rejected. Defaults to zero.
-
-L{Messages<Message>} enter this window only when you take them into your application
-using L{get}.  If your incoming window size is I{n}, and you get I{n}+1 L{messages<Message>}
-without explicitly accepting or rejecting the oldest message, then the
-message that passes beyond the edge of the incoming window will be assigned
-the default disposition of its link.
-""")
-
-  def _get_outgoing_window(self):
-    return pn_messenger_get_outgoing_window(self._mng)
-
-  def _set_outgoing_window(self, window):
-    self._check(pn_messenger_set_outgoing_window(self._mng, window))
-
-  outgoing_window = property(_get_outgoing_window, _set_outgoing_window,
-                             doc="""
-The outgoing tracking window for the messenger. The messenger will
-track the remote status of this many outgoing deliveries after calling
-send. Defaults to zero.
-
-A L{Message} enters this window when you call the put() method with the
-message.  If your outgoing window size is I{n}, and you call L{put} I{n}+1
-times, status information will no longer be available for the
-first message.
-""")
-
-  def start(self):
-    """
-    Currently a no-op placeholder.
-    For future compatibility, do not L{send} or L{recv} messages
-    before starting the L{Messenger}.
-    """
-    self._check(pn_messenger_start(self._mng))
-
-  def stop(self):
-    """
-    Transitions the L{Messenger} to an inactive state. An inactive
-    L{Messenger} will not send or receive messages from its internal
-    queues. A L{Messenger} should be stopped before being discarded to
-    ensure a clean shutdown handshake occurs on any internally managed
-    connections.
-    """
-    self._check(pn_messenger_stop(self._mng))
-
-  @property
-  def stopped(self):
-    """
-    Returns true iff a L{Messenger} is in the stopped state.
-    This function does not block.
-    """
-    return pn_messenger_stopped(self._mng)
-
-  def subscribe(self, source):
-    """
-    Subscribes the L{Messenger} to messages originating from the
-    specified source. The source is an address as specified in the
-    L{Messenger} introduction with the following addition. If the
-    domain portion of the address begins with the '~' character, the
-    L{Messenger} will interpret the domain as host/port, bind to it,
-    and listen for incoming messages. For example "~0.0.0.0",
-    "amqp://~0.0.0.0", and "amqps://~0.0.0.0" will all bind to any
-    local interface and listen for incoming messages with the last
-    variant only permitting incoming SSL connections.
-
-    @type source: string
-    @param source: the source of messages to subscribe to
-    """
-    sub_impl = pn_messenger_subscribe(self._mng, source)
-    if not sub_impl:
-      self._check(pn_error_code(pn_messenger_error(self._mng)))
-      raise MessengerException("Cannot subscribe to %s"%source)
-    return Subscription(sub_impl)
-
-  def put(self, message):
-    """
-    Places the content contained in the message onto the outgoing
-    queue of the L{Messenger}. This method will never block, however
-    it will send any unblocked L{Messages<Message>} in the outgoing
-    queue immediately and leave any blocked L{Messages<Message>}
-    remaining in the outgoing queue. The L{send} call may be used to
-    block until the outgoing queue is empty. The L{outgoing} property
-    may be used to check the depth of the outgoing queue.
-
-    When the content in a given L{Message} object is copied to the outgoing
-    message queue, you may then modify or discard the L{Message} object
-    without having any impact on the content in the outgoing queue.
-
-    This method returns an outgoing tracker for the L{Message}.  The tracker
-    can be used to determine the delivery status of the L{Message}.
-
-    @type message: Message
-    @param message: the message to place in the outgoing queue
-    @return: a tracker
-    """
-    message._pre_encode()
-    self._check(pn_messenger_put(self._mng, message._msg))
-    return pn_messenger_outgoing_tracker(self._mng)
-
-  def status(self, tracker):
-    """
-    Gets the last known remote state of the delivery associated with
-    the given tracker.
-
-    @type tracker: tracker
-    @param tracker: the tracker whose status is to be retrieved
-
-    @return: one of None, PENDING, REJECTED, or ACCEPTED
-    """
-    disp = pn_messenger_status(self._mng, tracker);
-    return STATUSES.get(disp, disp)
-
-  def buffered(self, tracker):
-    """
-    Checks if the delivery associated with the given tracker is still
-    waiting to be sent.
-
-    @type tracker: tracker
-    @param tracker: the tracker whose status is to be retrieved
-
-    @return: true if delivery is still buffered
-    """
-    return pn_messenger_buffered(self._mng, tracker);
-
-  def settle(self, tracker=None):
-    """
-    Frees a L{Messenger} from tracking the status associated with a given
-    tracker. If you don't supply a tracker, all outgoing L{messages<Message>} up
-    to the most recent will be settled.
-    """
-    if tracker is None:
-      tracker = pn_messenger_outgoing_tracker(self._mng)
-      flags = PN_CUMULATIVE
-    else:
-      flags = 0
-    self._check(pn_messenger_settle(self._mng, tracker, flags))
-
-  def send(self, n=-1):
-    """
-    This call will block until the indicated number of L{messages<Message>}
-    have been sent, or until the operation times out.  If n is -1 this call will
-    block until all outgoing L{messages<Message>} have been sent. If n is 0 then
-    this call will send whatever it can without blocking.
-    """
-    self._check(pn_messenger_send(self._mng, n))
-
-  def recv(self, n=None):
-    """
-    Receives up to I{n} L{messages<Message>} into the incoming queue.  If no value
-    for I{n} is supplied, this call will receive as many L{messages<Message>} as it
-    can buffer internally.  If the L{Messenger} is in blocking mode, this
-    call will block until at least one L{Message} is available in the
-    incoming queue.
-    """
-    if n is None:
-      n = -1
-    self._check(pn_messenger_recv(self._mng, n))
-
-  def work(self, timeout=None):
-    """
-    Sends or receives any outstanding L{messages<Message>} queued for a L{Messenger}.
-    This will block for the indicated timeout.
-    This method may also do I/O work other than sending and receiving
-    L{messages<Message>}.  For example, closing connections after messenger.L{stop}()
-    has been called.
-    """
-    if timeout is None:
-      t = -1
-    else:
-      t = secs2millis(timeout)
-    err = pn_messenger_work(self._mng, t)
-    if (err == PN_TIMEOUT):
-      return False
-    else:
-      self._check(err)
-      return True
-
-  @property
-  def receiving(self):
-    return pn_messenger_receiving(self._mng)
-
-  def interrupt(self):
-    """
-    The L{Messenger} interface is single-threaded.
-    This is the only L{Messenger} function intended to be called
-    from outside of the L{Messenger} thread.
-    Call this from a non-messenger thread to interrupt
-    a L{Messenger} that is blocking.
-    This will cause any in-progress blocking call to throw
-    the L{Interrupt} exception.  If there is no currently blocking
-    call, then the next blocking call will be affected, even if it
-    is within the same thread that interrupt was called from.
-    """
-    self._check(pn_messenger_interrupt(self._mng))
-
-  def get(self, message=None):
-    """
-    Moves the message from the head of the incoming message queue into
-    the supplied message object. Any content in the message will be
-    overwritten.
-
-    A tracker for the incoming L{Message} is returned.  The tracker can
-    later be used to communicate your acceptance or rejection of the
-    L{Message}.
-
-    If None is passed in for the L{Message} object, the L{Message}
-    popped from the head of the queue is discarded.
-
-    @type message: Message
-    @param message: the destination message object
-    @return: a tracker
-    """
-    if message is None:
-      impl = None
-    else:
-      impl = message._msg
-    self._check(pn_messenger_get(self._mng, impl))
-    if message is not None:
-      message._post_decode()
-    return pn_messenger_incoming_tracker(self._mng)
-
-  def accept(self, tracker=None):
-    """
-    Signal the sender that you have acted on the L{Message}
-    pointed to by the tracker.  If no tracker is supplied,
-    then all messages that have been returned by the L{get}
-    method are accepted, except those that have already been
-    auto-settled by passing beyond your incoming window size.
-
-    @type tracker: tracker
-    @param tracker: a tracker as returned by get
-    """
-    if tracker is None:
-      tracker = pn_messenger_incoming_tracker(self._mng)
-      flags = PN_CUMULATIVE
-    else:
-      flags = 0
-    self._check(pn_messenger_accept(self._mng, tracker, flags))
-
-  def reject(self, tracker=None):
-    """
-    Rejects the L{Message} indicated by the tracker.  If no tracker
-    is supplied, all messages that have been returned by the L{get}
-    method are rejected, except those that have already been auto-settled
-    by passing beyond your outgoing window size.
-
-    @type tracker: tracker
-    @param tracker: a tracker as returned by get
-    """
-    if tracker is None:
-      tracker = pn_messenger_incoming_tracker(self._mng)
-      flags = PN_CUMULATIVE
-    else:
-      flags = 0
-    self._check(pn_messenger_reject(self._mng, tracker, flags))
-
-  @property
-  def outgoing(self):
-    """
-    The outgoing queue depth.
-    """
-    return pn_messenger_outgoing(self._mng)
-
-  @property
-  def incoming(self):
-    """
-    The incoming queue depth.
-    """
-    return pn_messenger_incoming(self._mng)
-
-  def route(self, pattern, address):
-    """
-          Adds a routing rule to a L{Messenger's<Messenger>} internal routing table.
-
-          The route procedure may be used to influence how a L{Messenger} will
-          internally treat a given address or class of addresses. Every call
-          to the route procedure will result in L{Messenger} appending a routing
-          rule to its internal routing table.
-
-          Whenever a L{Message} is presented to a L{Messenger} for delivery, it
-          will match the address of this message against the set of routing
-          rules in order. The first rule to match will be triggered, and
-          instead of routing based on the address presented in the message,
-          the L{Messenger} will route based on the address supplied in the rule.
-
-          The pattern matching syntax supports two types of matches, a '%'
-          will match any character except a '/', and a '*' will match any
-          character including a '/'.
-
-          A routing address is specified as a normal AMQP address, however it
-          may additionally use substitution variables from the pattern match
-          that triggered the rule.
-
-          Any message sent to "foo" will be routed to "amqp://foo.com":
-
-             >>> messenger.route("foo", "amqp://foo.com");
-
-          Any message sent to "foobar" will be routed to
-          "amqp://foo.com/bar":
-
-             >>> messenger.route("foobar", "amqp://foo.com/bar");
-
-          Any message sent to bar/<path> will be routed to the corresponding
-          path within the amqp://bar.com domain:
-
-             >>> messenger.route("bar/*", "amqp://bar.com/$1");
-
-          Route all L{messages<Message>} over TLS:
-
-             >>> messenger.route("amqp:*", "amqps:$1")
-
-          Supply credentials for foo.com:
-
-             >>> messenger.route("amqp://foo.com/*", "amqp://user:password@foo.com/$1");
-
-          Supply credentials for all domains:
-
-             >>> messenger.route("amqp://*", "amqp://user:password@$1");
-
-          Route all addresses through a single proxy while preserving the
-          original destination:
-
-             >>> messenger.route("amqp://%/*", "amqp://user:password@proxy/$1/$2");
-
-          Route any address through a single broker:
-
-             >>> messenger.route("*", "amqp://user:password@broker/$1");
-    """
-    self._check(pn_messenger_route(self._mng, pattern, address))
-
-  def rewrite(self, pattern, address):
-    """
-    Similar to route(), except that the destination of
-    the L{Message} is determined before the message address is rewritten.
-
-    The outgoing address is only rewritten after routing has been
-    finalized.  If a message has an outgoing address of
-    "amqp://0.0.0.0:5678", and a rewriting rule that changes its
-    outgoing address to "foo", it will still arrive at the peer that
-    is listening on "amqp://0.0.0.0:5678", but when it arrives there,
-    the receiver will see its outgoing address as "foo".
-
-    The default rewrite rule removes username and password from addresses
-    before they are transmitted.
-    """
-    self._check(pn_messenger_rewrite(self._mng, pattern, address))
-
-  def selectable(self):
-    impl = pn_messenger_selectable(self._mng)
-    if impl:
-      fd = pn_selectable_fd(impl)
-      sel = self._selectables.get(fd, None)
-      if sel is None:
-        sel = Selectable(self, impl)
-        self._selectables[fd] = sel
-      return sel
-    else:
-      return None
-
-  @property
-  def deadline(self):
-    tstamp = pn_messenger_deadline(self._mng)
-    if tstamp:
-      return millis2secs(tstamp)
-    else:
-      return None
-
-class Message(object):
-  """The L{Message} class is a mutable holder of message content.
-
-  @ivar instructions: delivery instructions for the message
-  @type instructions: dict
-  @ivar annotations: infrastructure defined message annotations
-  @type annotations: dict
-  @ivar properties: application defined message properties
-  @type properties: dict
-  @ivar body: message body
-  @type body: bytes | unicode | dict | list | int | long | float | UUID
-  """
-
-  DATA = PN_DATA
-  TEXT = PN_TEXT
-  AMQP = PN_AMQP
-  JSON = PN_JSON
-
-  DEFAULT_PRIORITY = PN_DEFAULT_PRIORITY
-
-  def __init__(self, **kwargs):
-    """
-    @param kwargs: Message property name/value pairs to initialise the Message
-    """
-    self._msg = pn_message()
-    self._id = Data(pn_message_id(self._msg))
-    self._correlation_id = Data(pn_message_correlation_id(self._msg))
-    self.instructions = None
-    self.annotations = None
-    self.properties = None
-    self.body = None
-    for k,v in kwargs.iteritems():
-      getattr(self, k)          # Raise exception if it's not a valid attribute.
-      setattr(self, k, v)
-
-  def __del__(self):
-    if hasattr(self, "_msg"):
-      pn_message_free(self._msg)
-      del self._msg
-
-  def _check(self, err):
-    if err < 0:
-      exc = EXCEPTIONS.get(err, MessageException)
-      raise exc("[%s]: %s" % (err, pn_error_text(pn_message_error(self._msg))))
-    else:
-      return err
-
-  def _pre_encode(self):
-    inst = Data(pn_message_instructions(self._msg))
-    ann = Data(pn_message_annotations(self._msg))
-    props = Data(pn_message_properties(self._msg))
-    body = Data(pn_message_body(self._msg))
-
-    inst.clear()
-    if self.instructions is not None:
-      inst.put_object(self.instructions)
-    ann.clear()
-    if self.annotations is not None:
-      ann.put_object(self.annotations)
-    props.clear()
-    if self.properties is not None:
-      props.put_object(self.properties)
-    body.clear()
-    if self.body is not None:
-      body.put_object(self.body)
-
-  def _post_decode(self):
-    inst = Data(pn_message_instructions(self._msg))
-    ann = Data(pn_message_annotations(self._msg))
-    props = Data(pn_message_properties(self._msg))
-    body = Data(pn_message_body(self._msg))
-
-    if inst.next():
-      self.instructions = inst.get_object()
-    else:
-      self.instructions = None
-    if ann.next():
-      self.annotations = ann.get_object()
-    else:
-      self.annotations = None
-    if props.next():
-      self.properties = props.get_object()
-    else:
-      self.properties = None
-    if body.next():
-      self.body = body.get_object()
-    else:
-      self.body = None
-
-  def clear(self):
-    """
-    Clears the contents of the L{Message}. All fields will be reset to
-    their default values.
-    """
-    pn_message_clear(self._msg)
-    self.instructions = None
-    self.annotations = None
-    self.properties = None
-    self.body = None
-
-  def _is_inferred(self):
-    return pn_message_is_inferred(self._msg)
-
-  def _set_inferred(self, value):
-    self._check(pn_message_set_inferred(self._msg, bool(value)))
-
-  inferred = property(_is_inferred, _set_inferred, doc="""
-The inferred flag for a message indicates how the message content
-is encoded into AMQP sections. If inferred is true then binary and
-list values in the body of the message will be encoded as AMQP DATA
-and AMQP SEQUENCE sections, respectively. If inferred is false,
-then all values in the body of the message will be encoded as AMQP
-VALUE sections regardless of their type.
-""")
-
-  def _is_durable(self):
-    return pn_message_is_durable(self._msg)
-
-  def _set_durable(self, value):
-    self._check(pn_message_set_durable(self._msg, bool(value)))
-
-  durable = property(_is_durable, _set_durable,
-                     doc="""
-The durable property indicates that the message should be held durably
-by any intermediaries taking responsibility for the message.
-""")
-
-  def _get_priority(self):
-    return pn_message_get_priority(self._msg)
-
-  def _set_priority(self, value):
-    self._check(pn_message_set_priority(self._msg, value))
-
-  priority = property(_get_priority, _set_priority,
-                      doc="""
-The priority of the message.
-""")
-
-  def _get_ttl(self):
-    return millis2secs(pn_message_get_ttl(self._msg))
-
-  def _set_ttl(self, value):
-    self._check(pn_message_set_ttl(self._msg, secs2millis(value)))
-
-  ttl = property(_get_ttl, _set_ttl,
-                 doc="""
-The time to live of the message measured in seconds. Expired messages
-may be dropped.
-""")
-
-  def _is_first_acquirer(self):
-    return pn_message_is_first_acquirer(self._msg)
-
-  def _set_first_acquirer(self, value):
-    self._check(pn_message_set_first_acquirer(self._msg, bool(value)))
-
-  first_acquirer = property(_is_first_acquirer, _set_first_acquirer,
-                            doc="""
-True iff the recipient is the first to acquire the message.
-""")
-
-  def _get_delivery_count(self):
-    return pn_message_get_delivery_count(self._msg)
-
-  def _set_delivery_count(self, value):
-    self._check(pn_message_set_delivery_count(self._msg, value))
-
-  delivery_count = property(_get_delivery_count, _set_delivery_count,
-                            doc="""
-The number of delivery attempts made for this message.
-""")
-
-
-  def _get_id(self):
-    return self._id.get_object()
-  def _set_id(self, value):
-    if type(value) in (int, long):
-      value = ulong(value)
-    self._id.rewind()
-    self._id.put_object(value)
-  id = property(_get_id, _set_id,
-                doc="""
-The id of the message.
-""")
-
-  def _get_user_id(self):
-    return pn_message_get_user_id(self._msg)
-
-  def _set_user_id(self, value):
-    self._check(pn_message_set_user_id(self._msg, value))
-
-  user_id = property(_get_user_id, _set_user_id,
-                     doc="""
-The user id of the message creator.
-""")
-
-  def _get_address(self):
-    return pn_message_get_address(self._msg)
-
-  def _set_address(self, value):
-    self._check(pn_message_set_address(self._msg, value))
-
-  address = property(_get_address, _set_address,
-                     doc="""
-The address of the message.
-""")
-
-  def _get_subject(self):
-    return pn_message_get_subject(self._msg)
-
-  def _set_subject(self, value):
-    self._check(pn_message_set_subject(self._msg, value))
-
-  subject = property(_get_subject, _set_subject,
-                     doc="""
-The subject of the message.
-""")
-
-  def _get_reply_to(self):
-    return pn_message_get_reply_to(self._msg)
-
-  def _set_reply_to(self, value):
-    self._check(pn_message_set_reply_to(self._msg, value))
-
-  reply_to = property(_get_reply_to, _set_reply_to,
-                      doc="""
-The reply-to address for the message.
-""")
-
-  def _get_correlation_id(self):
-    return self._correlation_id.get_object()
-  def _set_correlation_id(self, value):
-    if type(value) in (int, long):
-      value = ulong(value)
-    self._correlation_id.rewind()
-    self._correlation_id.put_object(value)
-
-  correlation_id = property(_get_correlation_id, _set_correlation_id,
-                            doc="""
-The correlation-id for the message.
-""")
-
-  def _get_content_type(self):
-    return pn_message_get_content_type(self._msg)
-
-  def _set_content_type(self, value):
-    self._check(pn_message_set_content_type(self._msg, value))
-
-  content_type = property(_get_content_type, _set_content_type,
-                          doc="""
-The content-type of the message.
-""")
-
-  def _get_content_encoding(self):
-    return pn_message_get_content_encoding(self._msg)
-
-  def _set_content_encoding(self, value):
-    self._check(pn_message_set_content_encoding(self._msg, value))
-
-  content_encoding = property(_get_content_encoding, _set_content_encoding,
-                              doc="""
-The content-encoding of the message.
-""")
-
-  def _get_expiry_time(self):
-    return millis2secs(pn_message_get_expiry_time(self._msg))
-
-  def _set_expiry_time(self, value):
-    self._check(pn_message_set_expiry_time(self._msg, secs2millis(value)))
-
-  expiry_time = property(_get_expiry_time, _set_expiry_time,
-                         doc="""
-The expiry time of the message.
-""")
-
-  def _get_creation_time(self):
-    return millis2secs(pn_message_get_creation_time(self._msg))
-
-  def _set_creation_time(self, value):
-    self._check(pn_message_set_creation_time(self._msg, secs2millis(value)))
-
-  creation_time = property(_get_creation_time, _set_creation_time,
-                           doc="""
-The creation time of the message.
-""")
-
-  def _get_group_id(self):
-    return pn_message_get_group_id(self._msg)
-
-  def _set_group_id(self, value):
-    self._check(pn_message_set_group_id(self._msg, value))
-
-  group_id = property(_get_group_id, _set_group_id,
-                      doc="""
-The group id of the message.
-""")
-
-  def _get_group_sequence(self):
-    return pn_message_get_group_sequence(self._msg)
-
-  def _set_group_sequence(self, value):
-    self._check(pn_message_set_group_sequence(self._msg, value))
-
-  group_sequence = property(_get_group_sequence, _set_group_sequence,
-                            doc="""
-The sequence of the message within its group.
-""")
-
-  def _get_reply_to_group_id(self):
-    return pn_message_get_reply_to_group_id(self._msg)
-
-  def _set_reply_to_group_id(self, value):
-    self._check(pn_message_set_reply_to_group_id(self._msg, value))
-
-  reply_to_group_id = property(_get_reply_to_group_id, _set_reply_to_group_id,
-                               doc="""
-The group-id for any replies.
-""")
-
-  # XXX
-  def _get_format(self):
-    return pn_message_get_format(self._msg)
-
-  def _set_format(self, value):
-    self._check(pn_message_set_format(self._msg, value))
-
-  format = property(_get_format, _set_format,
-                    doc="""
-The format of the message.
-""")
-
-  def encode(self):
-    self._pre_encode()
-    sz = 16
-    while True:
-      err, data = pn_message_encode(self._msg, sz)
-      if err == PN_OVERFLOW:
-        sz *= 2
-        continue
-      else:
-        self._check(err)
-        return data
-
-  def decode(self, data):
-    self._check(pn_message_decode(self._msg, data, len(data)))
-    self._post_decode()
-
-  def load(self, data):
-    self._check(pn_message_load(self._msg, data))
-
-  def save(self):
-    sz = 16
-    while True:
-      err, data = pn_message_save(self._msg, sz)
-      if err == PN_OVERFLOW:
-        sz *= 2
-        continue
-      else:
-        self._check(err)
-        return data
-
-  def __repr2__(self):
-    props = []
-    for attr in ("inferred", "address", "reply_to", "durable", "ttl",
-                 "priority", "first_acquirer", "delivery_count", "id",
-                 "correlation_id", "user_id", "group_id", "group_sequence",
-                 "reply_to_group_id", "instructions", "annotations",
-                 "properties", "body"):
-      value = getattr(self, attr)
-      if value: props.append("%s=%r" % (attr, value))
-    return "Message(%s)" % ", ".join(props)
-
-  def __repr__(self):
-    tmp = pn_string(None)
-    err = pn_inspect(self._msg, tmp)
-    result = pn_string_get(tmp)
-    pn_free(tmp)
-    self._check(err)
-    return result
-
-class Subscription(object):
-
-  def __init__(self, impl):
-    self._impl = impl
-
-  @property
-  def address(self):
-    return pn_subscription_address(self._impl)
-
-class Selectable(object):
-
-  def __init__(self, messenger, impl):
-    self.messenger = messenger
-    self._impl = impl
-
-  def fileno(self):
-    if not self._impl: raise ValueError("selectable freed")
-    return pn_selectable_fd(self._impl)
-
-  @property
-  def capacity(self):
-    if not self._impl: raise ValueError("selectable freed")
-    return pn_selectable_capacity(self._impl)
-
-  @property
-  def pending(self):
-    if not self._impl: raise ValueError("selectable freed")
-    return pn_selectable_pending(self._impl)
-
-  @property
-  def deadline(self):
-    if not self._impl: raise ValueError("selectable freed")
-    tstamp = pn_selectable_deadline(self._impl)
-    if tstamp:
-      return millis2secs(tstamp)
-    else:
-      return None
-
-  def readable(self):
-    if not self._impl: raise ValueError("selectable freed")
-    pn_selectable_readable(self._impl)
-
-  def writable(self):
-    if not self._impl: raise ValueError("selectable freed")
-    pn_selectable_writable(self._impl)
-
-  def expired(self):
-    if not self._impl: raise ValueError("selectable freed")
-    pn_selectable_expired(self._impl)
-
-  def _is_registered(self):
-    if not self._impl: raise ValueError("selectable freed")
-    return pn_selectable_is_registered(self._impl)
-
-  def _set_registered(self, registered):
-    if not self._impl: raise ValueError("selectable freed")
-    pn_selectable_set_registered(self._impl, registered)
-
-  registered = property(_is_registered, _set_registered,
-                    doc="""
-The registered property may be get/set by an I/O polling system to
-indicate whether the fd has been registered or not.
-""")
-
-  @property
-  def is_terminal(self):
-    if not self._impl: return True
-    return pn_selectable_is_terminal(self._impl)
-
-  def free(self):
-    if self._impl:
-      del self.messenger._selectables[self.fileno()]
-      pn_selectable_free(self._impl)
-      self._impl = None
-
-  def __del__(self):
-    self.free()
-
-class DataException(ProtonException):
-  """
-  The DataException class is the root of the Data exception hierarchy.
-  All exceptions raised by the Data class extend this exception.
-  """
-  pass
-
-class UnmappedType:
-
-  def __init__(self, msg):
-    self.msg = msg
-
-  def __repr__(self):
-    return "UnmappedType(%s)" % self.msg
-
-class ulong(long):
-
-  def __repr__(self):
-    return "ulong(%s)" % long.__repr__(self)
-
-class timestamp(long):
-
-  def __repr__(self):
-    return "timestamp(%s)" % long.__repr__(self)
-
-class symbol(unicode):
-
-  def __repr__(self):
-    return "symbol(%s)" % unicode.__repr__(self)
-
-class char(unicode):
-
-  def __repr__(self):
-    return "char(%s)" % unicode.__repr__(self)
-
-class Described(object):
-
-  def __init__(self, descriptor, value):
-    self.descriptor = descriptor
-    self.value = value
-
-  def __repr__(self):
-    return "Described(%r, %r)" % (self.descriptor, self.value)
-
-  def __eq__(self, o):
-    if isinstance(o, Described):
-      return self.descriptor == o.descriptor and self.value == o.value
-    else:
-      return False
-
-UNDESCRIBED = Constant("UNDESCRIBED")
-
-class Array(object):
-
-  def __init__(self, descriptor, type, *elements):
-    self.descriptor = descriptor
-    self.type = type
-    self.elements = elements
-
-  def __repr__(self):
-    if self.elements:
-      els = ", %s"  % (", ".join(map(repr, self.elements)))
-    else:
-      els = ""
-    return "Array(%r, %r%s)" % (self.descriptor, self.type, els)
-
-  def __eq__(self, o):
-    if isinstance(o, Array):
-      return self.descriptor == o.descriptor and \
-          self.type == o.type and self.elements == o.elements
-    else:
-      return False
-
-class Data:
-  """
-  The L{Data} class provides an interface for decoding, extracting,
-  creating, and encoding arbitrary AMQP data. A L{Data} object
-  contains a tree of AMQP values. Leaf nodes in this tree correspond
-  to scalars in the AMQP type system such as L{ints<INT>} or
-  L{strings<STRING>}. Non-leaf nodes in this tree correspond to
-  compound values in the AMQP type system such as L{lists<LIST>},
-  L{maps<MAP>}, L{arrays<ARRAY>}, or L{described values<DESCRIBED>}.
-  The root node of the tree is the L{Data} object itself and can have
-  an arbitrary number of children.
-
-  A L{Data} object maintains the notion of the current sibling node
-  and a current parent node. Siblings are ordered within their parent.
-  Values are accessed and/or added by using the L{next}, L{prev},
-  L{enter}, and L{exit} methods to navigate to the desired location in
-  the tree and using the supplied variety of put_*/get_* methods to
-  access or add a value of the desired type.
-
-  The put_* methods will always add a value I{after} the current node
-  in the tree. If the current node has a next sibling the put_* method
-  will overwrite the value on this node. If there is no current node
-  or the current node has no next sibling then one will be added. The
-  put_* methods always set the added/modified node to the current
-  node. The get_* methods read the value of the current node and do
-  not change which node is current.
-
-  The following types of scalar values are supported:
-
-   - L{NULL}
-   - L{BOOL}
-   - L{UBYTE}
-   - L{USHORT}
-   - L{SHORT}
-   - L{UINT}
-   - L{INT}
-   - L{ULONG}
-   - L{LONG}
-   - L{FLOAT}
-   - L{DOUBLE}
-   - L{BINARY}
-   - L{STRING}
-   - L{SYMBOL}
-
-  The following types of compound values are supported:
-
-   - L{DESCRIBED}
-   - L{ARRAY}
-   - L{LIST}
-   - L{MAP}
-  """
-
-  NULL = PN_NULL; "A null value."
-  BOOL = PN_BOOL; "A boolean value."
-  UBYTE = PN_UBYTE; "An unsigned byte value."
-  BYTE = PN_BYTE; "A signed byte value."
-  USHORT = PN_USHORT; "An unsigned short value."
-  SHORT = PN_SHORT; "A short value."
-  UINT = PN_UINT; "An unsigned int value."
-  INT = PN_INT; "A signed int value."
-  CHAR = PN_CHAR; "A character value."
-  ULONG = PN_ULONG; "An unsigned long value."
-  LONG = PN_LONG; "A signed long value."
-  TIMESTAMP = PN_TIMESTAMP; "A timestamp value."
-  FLOAT = PN_FLOAT; "A float value."
-  DOUBLE = PN_DOUBLE; "A double value."
-  DECIMAL32 = PN_DECIMAL32; "A DECIMAL32 value."
-  DECIMAL64 = PN_DECIMAL64; "A DECIMAL64 value."
-  DECIMAL128 = PN_DECIMAL128; "A DECIMAL128 value."
-  UUID = PN_UUID; "A UUID value."
-  BINARY = PN_BINARY; "A binary string."
-  STRING = PN_STRING; "A unicode string."
-  SYMBOL = PN_SYMBOL; "A symbolic string."
-  DESCRIBED = PN_DESCRIBED; "A described value."
-  ARRAY = PN_ARRAY; "An array value."
-  LIST = PN_LIST; "A list value."
-  MAP = PN_MAP; "A map value."
-
-  type_names = {
-    NULL: "null",
-    BOOL: "bool",
-    BYTE: "byte",
-    UBYTE: "ubyte",
-    SHORT: "short",
-    USHORT: "ushort",
-    INT: "int",
-    UINT: "uint",
-    CHAR: "char",
-    LONG: "long",
-    ULONG: "ulong",
-    TIMESTAMP: "timestamp",
-    FLOAT: "float",
-    DOUBLE: "double",
-    DECIMAL32: "decimal32",
-    DECIMAL64: "decimal64",
-    DECIMAL128: "decimal128",
-    UUID: "uuid",
-    BINARY: "binary",
-    STRING: "string",
-    SYMBOL: "symbol",
-    DESCRIBED: "described",
-    ARRAY: "array",
-    LIST: "list",
-    MAP: "map"
-    }
-
-  @classmethod
-  def type_name(type): return Data.type_names[type]
-
-  def __init__(self, capacity=16):
-    if type(capacity) in (int, long):
-      self._data = pn_data(capacity)
-      self._free = True
-    else:
-      self._data = capacity
-      self._free = False
-
-  def __del__(self):
-    if self._free and hasattr(self, "_data"):
-      pn_data_free(self._data)
-      del self._data
-
-  def _check(self, err):
-    if err < 0:
-      exc = EXCEPTIONS.get(err, DataException)
-      raise exc("[%s]: %s" % (err, pn_error_text(pn_data_error(self._data))))
-    else:
-      return err
-
-  def clear(self):
-    """
-    Clears the data object.
-    """
-    pn_data_clear(self._data)
-
-  def rewind(self):
-    """
-    Clears current node and sets the parent to the root node.  Clearing the
-    current node sets it _before_ the first node, calling next() will advance to
-    the first node.
-    """
-    assert self._data is not None
-    pn_data_rewind(self._data)
-
-  def next(self):
-    """
-    Advances the current node to its next sibling and returns its
-    type. If there is no next sibling the current node remains
-    unchanged and None is returned.
-    """
-    found = pn_data_next(self._data)
-    if found:
-      return self.type()
-    else:
-      return None
-
-  def prev(self):
-    """
-    Advances the current node to its previous sibling and returns its
-    type. If there is no previous sibling the current node remains
-    unchanged and None is returned.
-    """
-    found = pn_data_prev(self._data)
-    if found:
-      return self.type()
-    else:
-      return None
-
-  def enter(self):
-    """
-    Sets the parent node to the current node and clears the current node.
-    Clearing the current node sets it _before_ the first child,
-    call next() advances to the first child.
-    """
-    return pn_data_enter(self._data)
-
-  def exit(self):
-    """
-    Sets the current node to the parent node and the parent node to
-    its own parent.
-    """
-    return pn_data_exit(self._data)
-
-  def lookup(self, name):
-    return pn_data_lookup(self._data, name)
-
-  def narrow(self):
-    pn_data_narrow(self._data)
-
-  def widen(self):
-    pn_data_widen(self._data)
-
-  def type(self):
-    """
-    Returns the type of the current node.
-    """
-    dtype = pn_data_type(self._data)
-    if dtype == -1:
-      return None
-    else:
-      return dtype
-
-  def encode(self):
-    """
-    Returns a representation of the data encoded in AMQP format.
-    """
-    size = 1024
-    while True:
-      cd, enc = pn_data_encode(self._data, size)
-      if cd == PN_OVERFLOW:
-        size *= 2
-      elif cd >= 0:
-        return enc
-      else:
-        self._check(cd)
-
-  def decode(self, encoded):
-    """
-    Decodes the first value from supplied AMQP data and returns the
-    number of bytes consumed.
-
-    @type encoded: binary
-    @param encoded: AMQP encoded binary data
-    """
-    return self._check(pn_data_decode(self._data, encoded))
-
-  def put_list(self):
-    """
-    Puts a list value. Elements may be filled by entering the list
-    node and putting element values.
-
-      >>> data = Data()
-      >>> data.put_list()
-      >>> data.enter()
-      >>> data.put_int(1)
-      >>> data.put_int(2)
-      >>> data.put_int(3)
-      >>> data.exit()
-    """
-    self._check(pn_data_put_list(self._data))
-
-  def put_map(self):
-    """
-    Puts a map value. Elements may be filled by entering the map node
-    and putting alternating key value pairs.
-
-      >>> data = Data()
-      >>> data.put_map()
-      >>> data.enter()
-      >>> data.put_string("key")
-      >>> data.put_string("value")
-      >>> data.exit()
-    """
-    self._check(pn_data_put_map(self._data))
-
-  def put_array(self, described, element_type):
-    """
-    Puts an array value. Elements may be filled by entering the array
-    node and putting the element values. The values must all be of the
-    specified array element type. If an array is described then the
-    first child value of the array is the descriptor and may be of any
-    type.
-
-      >>> data = Data()
-      >>>
-      >>> data.put_array(False, Data.INT)
-      >>> data.enter()
-      >>> data.put_int(1)
-      >>> data.put_int(2)
-      >>> data.put_int(3)
-      >>> data.exit()
-      >>>
-      >>> data.put_array(True, Data.DOUBLE)
-      >>> data.enter()
-      >>> data.put_symbol("array-descriptor")
-      >>> data.put_double(1.1)
-      >>> data.put_double(1.2)
-      >>> data.put_double(1.3)
-      >>> data.exit()
-
-    @type described: bool
-    @param described: specifies whether the array is described
-    @type element_type: int
-    @param element_type: the type of the array elements
-    """
-    self._check(pn_data_put_array(self._data, described, element_type))
-
-  def put_described(self):
-    """
-    Puts a described value. A described node has two children, the
-    descriptor and the value. These are specified by entering the node
-    and putting the desired values.
-
-      >>> data = Data()
-      >>> data.put_described()
-      >>> data.enter()
-      >>> data.put_symbol("value-descriptor")
-      >>> data.put_string("the value")
-      >>> data.exit()
-    """
-    self._check(pn_data_put_described(self._data))
-
-  def put_null(self):
-    """
-    Puts a null value.
-    """
-    self._check(pn_data_put_null(self._data))
-
-  def put_bool(self, b):
-    """
-    Puts a boolean value.
-
-    @param b: a boolean value
-    """
-    self._check(pn_data_put_bool(self._data, b))
-
-  def put_ubyte(self, ub):
-    """
-    Puts an unsigned byte value.
-
-    @param ub: an integral value
-    """
-    self._check(pn_data_put_ubyte(self._data, ub))
-
-  def put_byte(self, b):
-    """
-    Puts a signed byte value.
-
-    @param b: an integral value
-    """
-    self._check(pn_data_put_byte(self._data, b))
-
-  def put_ushort(self, us):
-    """
-    Puts an unsigned short value.
-
-    @param us: an integral value.
-    """
-    self._check(pn_data_put_ushort(self._data, us))
-
-  def put_short(self, s):
-    """
-    Puts a signed short value.
-
-    @param s: an integral value
-    """
-    self._check(pn_data_put_short(self._data, s))
-
-  def put_uint(self, ui):
-    """
-    Puts an unsigned int value.
-
-    @param ui: an integral value
-    """
-    self._check(pn_data_put_uint(self._data, ui))
-
-  def put_int(self, i):
-    """
-    Puts a signed int value.
-
-    @param i: an integral value
-    """
-    self._check(pn_data_put_int(self._data, i))
-
-  def put_char(self, c):
-    """
-    Puts a char value.
-
-    @param c: a single character
-    """
-    self._check(pn_data_put_char(self._data, ord(c)))
-
-  def put_ulong(self, ul):
-    """
-    Puts an unsigned long value.
-
-    @param ul: an integral value
-    """
-    self._check(pn_data_put_ulong(self._data, ul))
-
-  def put_long(self, l):
-    """
-    Puts a signed long value.
-
-    @param l: an integral value
-    """
-    self._check(pn_data_put_long(self._data, l))
-
-  def put_timestamp(self, t):
-    """
-    Puts a timestamp value.
-
-    @param t: an integral value
-    """
-    self._check(pn_data_put_timestamp(self._data, t))
-
-  def put_float(self, f):
-    """
-    Puts a float value.
-
-    @param f: a floating point value
-    """
-    self._check(pn_data_put_float(self._data, f))
-
-  def put_double(self, d):
-    """
-    Puts a double value.
-
-    @param d: a floating point value.
-    """
-    self._check(pn_data_put_double(self._data, d))
-
-  def put_decimal32(self, d):
-    """
-    Puts a decimal32 value.
-
-    @param d: a decimal32 value
-    """
-    self._check(pn_data_put_decimal32(self._data, d))
-
-  def put_decimal64(self, d):
-    """
-    Puts a decimal64 value.
-
-    @param d: a decimal64 value
-    """
-    self._check(pn_data_put_decimal64(self._data, d))
-
-  def put_decimal128(self, d):
-    """
-    Puts a decimal128 value.
-
-    @param d: a decimal128 value
-    """
-    self._check(pn_data_put_decimal128(self._data, d))
-
-  def put_uuid(self, u):
-    """
-    Puts a UUID value.
-
-    @param u: a uuid value
-    """
-    self._check(pn_data_put_uuid(self._data, u.bytes))
-
-  def put_binary(self, b):
-    """
-    Puts a binary value.
-
-    @type b: binary
-    @param b: a binary value
-    """
-    self._check(pn_data_put_binary(self._data, b))
-
-  def put_string(self, s):
-    """
-    Puts a unicode value.
-
-    @type s: unicode
-    @param s: a unicode value
-    """
-    self._check(pn_data_put_string(self._data, s.encode("utf8")))
-
-  def put_symbol(self, s):
-    """
-    Puts a symbolic value.
-
-    @type s: string
-    @param s: the symbol name
-    """
-    self._check(pn_data_put_symbol(self._data, s))
-
-  def get_list(self):
-    """
-    If the current node is a list, return the number of elements,
-    otherwise return zero. List elements can be accessed by entering
-    the list.
-
-      >>> count = data.get_list()
-      >>> data.enter()
-      >>> for i in range(count):
-      ...   type = data.next()
-      ...   if type == Data.STRING:
-      ...     print data.get_string()
-      ...   elif type == ...:
-      ...     ...
-      >>> data.exit()
-    """
-    return pn_data_get_list(self._data)
-
-  def get_map(self):
-    """
-    If the current node is a map, return the number of child elements,
-    otherwise return zero. Key value pairs can be accessed by entering
-    the map.
-
-      >>> count = data.get_map()
-      >>> data.enter()
-      >>> for i in range(count/2):
-      ...   type = data.next()
-      ...   if type == Data.STRING:
-      ...     print data.get_string()
-      ...   elif type == ...:
-      ...     ...
-      >>> data.exit()
-    """
-    return pn_data_get_map(self._data)
-
-  def get_array(self):
-    """
-    If the current node is an array, return a tuple of the element
-    count, a boolean indicating whether the array is described, and
-    the type of each element, otherwise return (0, False, None). Array
-    data can be accessed by entering the array.
-
-      >>> # read an array of strings with a symbolic descriptor
-      >>> count, described, type = data.get_array()
-      >>> data.enter()
-      >>> data.next()
-      >>> print "Descriptor:", data.get_symbol()
-      >>> for i in range(count):
-      ...    data.next()
-      ...    print "Element:", data.get_string()
-      >>> data.exit()
-    """
-    count = pn_data_get_array(self._data)
-    described = pn_data_is_array_described(self._data)
-    type = pn_data_get_array_type(self._data)
-    if type == -1:
-      type = None
-    return count, described, type
-
-  def is_described(self):
-    """
-    Checks if the current node is a described value. The descriptor
-    and value may be accessed by entering the described value.
-
-      >>> # read a symbolically described string
-      >>> assert data.is_described() # will error if the current node is not described
-      >>> data.enter()
-      >>> print data.get_symbol()
-      >>> print data.get_string()
-      >>> data.exit()
-    """
-    return pn_data_is_described(self._data)
-
-  def is_null(self):
-    """
-    Checks if the current node is a null.
-    """
-    return pn_data_is_null(self._data)
-
-  def get_bool(self):
-    """
-    If the current node is a boolean, returns its value, returns False
-    otherwise.
-    """
-    return pn_data_get_bool(self._data)
-
-  def get_ubyte(self):
-    """
-    If the current node is an unsigned byte, returns its value,
-    returns 0 otherwise.
-    """
-    return pn_data_get_ubyte(self._data)
-
-  def get_byte(self):
-    """
-    If the current node is a signed byte, returns its value, returns 0
-    otherwise.
-    """
-    return pn_data_get_byte(self._data)
-
-  def get_ushort(self):
-    """
-    If the current node is an unsigned short, returns its value,
-    returns 0 otherwise.
-    """
-    return pn_data_get_ushort(self._data)
-
-  def get_short(self):
-    """
-    If the current node is a signed short, returns its value, returns
-    0 otherwise.
-    """
-    return pn_data_get_short(self._data)
-
-  def get_uint(self):
-    """
-    If the current node is an unsigned int, returns its value, returns
-    0 otherwise.
-    """
-    return pn_data_get_uint(self._data)
-
-  def get_int(self):
-    """
-    If the current node is a signed int, returns its value, returns 0
-    otherwise.
-    """
-    return pn_data_get_int(self._data)
-
-  def get_char(self):
-    """
-    If the current node is a char, returns its value, returns 0
-    otherwise.
-    """
-    return char(unichr(pn_data_get_char(self._data)))
-
-  def get_ulong(self):
-    """
-    If the current node is an unsigned long, returns its value,
-    returns 0 otherwise.
-    """
-    return ulong(pn_data_get_ulong(self._data))
-
-  def get_long(self):
-    """
-    If the current node is an signed long, returns its value, returns
-    0 otherwise.
-    """
-    return pn_data_get_long(self._data)
-
-  def get_timestamp(self):
-    """
-    If the current node is a timestamp, returns its value, returns 0
-    otherwise.
-    """
-    return timestamp(pn_data_get_timestamp(self._data))
-
-  def get_float(self):
-    """
-    If the current node is a float, returns its value, raises 0
-    otherwise.
-    """
-    return pn_data_get_float(self._data)
-
-  def get_double(self):
-    """
-    If the current node is a double, returns its value, returns 0
-    otherwise.
-    """
-    return pn_data_get_double(self._data)
-
-  # XXX: need to convert
-  def get_decimal32(self):
-    """
-    If the current node is a decimal32, returns its value, returns 0
-    otherwise.
-    """
-    return pn_data_get_decimal32(self._data)
-
-  # XXX: need to convert
-  def get_decimal64(self):
-    """
-    If the current node is a decimal64, returns its value, returns 0
-    otherwise.
-    """
-    return pn_data_get_decimal64(self._data)
-
-  # XXX: need to convert
-  def get_decimal128(self):
-    """
-    If the current node is a decimal128, returns its value, returns 0
-    otherwise.
-    """
-    return pn_data_get_decimal128(self._data)
-
-  def get_uuid(self):
-    """
-    If the current node is a UUID, returns its value, returns None
-    otherwise.
-    """
-    if pn_data_type(self._data) == Data.UUID:
-      return uuid.UUID(bytes=pn_data_get_uuid(self._data))
-    else:
-      return None
-
-  def get_binary(self):
-    """
-    If the current node is binary, returns its value, returns ""
-    otherwise.
-    """
-    return pn_data_get_binary(self._data)
-
-  def get_string(self):
-    """
-    If the current node is a string, returns its value, returns ""
-    otherwise.
-    """
-    return pn_data_get_string(self._data).decode("utf8")
-
-  def get_symbol(self):
-    """
-    If the current node is a symbol, returns its value, returns ""
-    otherwise.
-    """
-    return symbol(pn_data_get_symbol(self._data))
-
-  def copy(self, src):
-    self._check(pn_data_copy(self._data, src._data))
-
-  def format(self):
-    sz = 16
-    while True:
-      err, result = pn_data_format(self._data, sz)
-      if err == PN_OVERFLOW:
-        sz *= 2
-        continue
-      else:
-        self._check(err)
-        return result
-
-  def dump(self):
-    pn_data_dump(self._data)
-
-  def put_dict(self, d):
-    self.put_map()
-    self.enter()
-    try:
-      for k, v in d.items():
-        self.put_object(k)
-        self.put_object(v)
-    finally:
-      self.exit()
-
-  def get_dict(self):
-    if self.enter():
-      try:
-        result = {}
-        while self.next():
-          k = self.get_object()
-          if self.next():
-            v = self.get_object()
-          else:
-            v = None
-          result[k] = v
-      finally:
-        self.exit()
-      return result
-
-  def put_sequence(self, s):
-    self.put_list()
-    self.enter()
-    try:
-      for o in s:
-        self.put_object(o)
-    finally:
-      self.exit()
-
-  def get_sequence(self):
-    if self.enter():
-      try:
-        result = []
-        while self.next():
-          result.append(self.get_object())
-      finally:
-        self.exit()
-      return result
-
-  def get_py_described(self):
-    if self.enter():
-      try:
-        self.next()
-        descriptor = self.get_object()
-        self.next()
-        value = self.get_object()
-      finally:
-        self.exit()
-      return Described(descriptor, value)
-
-  def put_py_described(self, d):
-    self.put_described()
-    self.enter()
-    try:
-      self.put_object(d.descriptor)
-      self.put_object(d.value)
-    finally:
-      self.exit()
-
-  def get_py_array(self):
-    """
-    If the current node is an array, return an Array object
-    representing the array and its contents. Otherwise return None.
-    This is a convenience wrapper around get_array, enter, etc.
-    """
-
-    count, described, type = self.get_array()
-    if type is None: return None
-    if self.enter():
-      try:
-        if described:
-          self.next()
-          descriptor = self.get_object()
-        else:
-          descriptor = UNDESCRIBED
-        elements = []
-        while self.next():
-          elements.append(self.get_object())
-      finally:
-        self.exit()
-      return Array(descriptor, type, *elements)
-
-  def put_py_array(self, a):
-    described = a.descriptor != UNDESCRIBED
-    self.put_array(described, a.type)
-    self.enter()
-    try:
-      if described:
-        self.put_object(a.descriptor)
-      for e in a.elements:
-        self.put_object(e)
-    finally:
-      self.exit()
-
-  put_mappings = {
-    None.__class__: lambda s, _: s.put_null(),
-    bool: put_bool,
-    dict: put_dict,
-    list: put_sequence,
-    tuple: put_sequence,
-    unicode: put_string,
-    bytes: put_binary,
-    symbol: put_symbol,
-    int: put_long,
-    char: put_char,
-    long: put_long,
-    ulong: put_ulong,
-    timestamp: put_timestamp,
-    float: put_double,
-    uuid.UUID: put_uuid,
-    Described: put_py_described,
-    Array: put_py_array
-    }
-  get_mappings = {
-    NULL: lambda s: None,
-    BOOL: get_bool,
-    BYTE: get_byte,
-    UBYTE: get_ubyte,
-    SHORT: get_short,
-    USHORT: get_ushort,
-    INT: get_int,
-    UINT: get_uint,
-    CHAR: get_char,
-    LONG: get_long,
-    ULONG: get_ulong,
-    TIMESTAMP: get_timestamp,
-    FLOAT: get_float,
-    DOUBLE: get_double,
-    DECIMAL32: get_decimal32,
-    DECIMAL64: get_decimal64,
-    DECIMAL128: get_decimal128,
-    UUID: get_uuid,
-    BINARY: get_binary,
-    STRING: get_string,
-    SYMBOL: get_symbol,
-    DESCRIBED: get_py_described,
-    ARRAY: get_py_array,
-    LIST: get_sequence,
-    MAP: get_dict
-    }
-
-
-  def put_object(self, obj):
-    putter = self.put_mappings[obj.__class__]
-    putter(self, obj)
-
-  def get_object(self):
-    type = self.type()
-    if type is None: return None
-    getter = self.get_mappings.get(type)
-    if getter:
-      return getter(self)
-    else:
-      return UnmappedType(str(type))
-
-class ConnectionException(ProtonException):
-  pass
-
-class Endpoint(object):
-
-  LOCAL_UNINIT = PN_LOCAL_UNINIT
-  REMOTE_UNINIT = PN_REMOTE_UNINIT
-  LOCAL_ACTIVE = PN_LOCAL_ACTIVE
-  REMOTE_ACTIVE = PN_REMOTE_ACTIVE
-  LOCAL_CLOSED = PN_LOCAL_CLOSED
-  REMOTE_CLOSED  = PN_REMOTE_CLOSED
-
-  def __init__(self):
-    self.condition = None
-    self._release_invoked = False
-
-  def _release(self):
-    """Release the underlying C Engine resource."""
-    if not self._release_invoked:
-      for c in self._children:
-        c._release()
-      self._free_resource()
-      self.connection._releasing(self)
-      self._release_invoked = True
-
-  def _update_cond(self):
-    obj2cond(self.condition, self._get_cond_impl())
-
-  @property
-  def remote_condition(self):
-    return cond2obj(self._get_remote_cond_impl())
-
-  # the following must be provided by subclasses
-  def _get_cond_impl(self):
-      assert False, "Subclass must override this!"
-
-  def _get_remote_cond_impl(self):
-      assert False, "Subclass must override this!"
-
-class Condition:
-
-  def __init__(self, name, description=None, info=None):
-    self.name = name
-    self.description = description
-    self.info = info
-
-  def __repr__(self):
-    return "Condition(%s)" % ", ".join([repr(x) for x in
-                                        (self.name, self.description, self.info)
-                                        if x])
-
-  def __eq__(self, o):
-    if not isinstance(o, Condition): return False
-    return self.name == o.name and \
-        self.description == o.description and \
-        self.info == o.info
-
-def obj2cond(obj, cond):
-  pn_condition_clear(cond)
-  if obj:
-    pn_condition_set_name(cond, str(obj.name))
-    pn_condition_set_description(cond, obj.description)
-    info = Data(pn_condition_info(cond))
-    if obj.info:
-      info.put_object(obj.info)
-
-def cond2obj(cond):
-  if pn_condition_is_set(cond):
-    return Condition(pn_condition_get_name(cond),
-                     pn_condition_get_description(cond),
-                     dat2obj(pn_condition_info(cond)))
-  else:
-    return None
-
-def dat2obj(dimpl):
-  if dimpl:
-    d = Data(dimpl)
-    d.rewind()
-    d.next()
-    obj = d.get_object()
-    d.rewind()
-    return obj
-
-def obj2dat(obj, dimpl):
-  if obj is not None:
-    d = Data(dimpl)
-    d.put_object(obj)
-
-def secs2millis(secs):
-  return long(secs*1000)
-
-def millis2secs(millis):
-  return float(millis)/1000.0
-
-class Connection(Endpoint):
-
-  @staticmethod
-  def _wrap_connection(c_conn):
-    """Maintain only a single instance of this class for each Connection
-    object that exists in the the C Engine.  This is done by storing a (weak)
-    reference to the python instance in the context field of the C object.
-    """
-    if not c_conn: return None
-    py_conn = pn_void2py(pn_connection_get_context(c_conn))
-    if py_conn: return py_conn
-    wrapper = Connection(_conn=c_conn)
-    return wrapper
-
-  def __init__(self, _conn=None):
-    Endpoint.__init__(self)
-    if _conn:
-      self._conn = _conn
-    else:
-      self._conn = pn_connection()
-    pn_connection_set_context(self._conn, pn_py2void(self))
-    self.offered_capabilities = None
-    self.desired_capabilities = None
-    self.properties = None
-    self._sessions = set()
-
-  def __del__(self):
-    if hasattr(self, "_conn") and self._conn:
-      self._release()
-
-  def free(self):
-    self._release()
-
-  @property
-  def _children(self):
-    return self._sessions
-
-  @property
-  def connection(self):
-    return self
-
-  def _free_resource(self):
-    pn_connection_free(self._conn)
-
-  def _released(self):
-    self._conn = None
-
-  def _releasing(self, child):
-    coll = getattr(self, "_collector", None)
-    if coll: coll = coll()
-    if coll:
-      coll._contexts.add(child)
-    else:
-      child._released()
-
-  def _check(self, err):
-    if err < 0:
-      exc = EXCEPTIONS.get(err, ConnectionException)
-      raise exc("[%s]: %s" % (err, pn_connection_error(self._conn)))
-    else:
-      return err
-
-  def _get_cond_impl(self):
-    return pn_connection_condition(self._conn)
-
-  def _get_remote_cond_impl(self):
-    return pn_connection_remote_condition(self._conn)
-
-  def collect(self, collector):
-    if collector is None:
-      pn_connection_collect(self._conn, None)
-    else:
-      pn_connection_collect(self._conn, collector._impl)
-    self._collector = weakref.ref(collector)
-
-  def _get_container(self):
-    return pn_connection_get_container(self._conn)
-  def _set_container(self, name):
-    return pn_connection_set_container(self._conn, name)
-
-  container = property(_get_container, _set_container)
-
-  def _get_hostname(self):
-    return pn_connection_get_hostname(self._conn)
-  def _set_hostname(self, name):
-    return pn_connection_set_hostname(self._conn, name)
-
-  hostname = property(_get_hostname, _set_hostname)
-
-  @property
-  def remote_container(self):
-    return pn_connection_remote_container(self._conn)
-
-  @property
-  def remote_hostname(self):
-    return pn_connection_remote_hostname(self._conn)
-
-  @property
-  def remote_offered_capabilities(self):
-    return dat2obj(pn_connection_remote_offered_capabilities(self._conn))
-
-  @property
-  def remote_desired_capabilities(self):
-    return dat2obj(pn_connection_remote_desired_capabilities(self._conn))
-
-  @property
-  def remote_properties(self):
-    return dat2obj(pn_connection_remote_properties(self._conn))
-
-  def open(self):
-    obj2dat(self.offered_capabilities,
-            pn_connection_offered_capabilities(self._conn))
-    obj2dat(self.desired_capabilities,
-            pn_connection_desired_capabilities(self._conn))
-    obj2dat(self.properties, pn_connection_properties(self._conn))
-    pn_connection_open(self._conn)
-
-  def close(self):
-    self._update_cond()
-    pn_connection_close(self._conn)
-
-  @property
-  def state(self):
-    return pn_connection_state(self._conn)
-
-  def session(self):
-    return Session._wrap_session(pn_session(self._conn))
-
-  def session_head(self, mask):
-    return Session._wrap_session(pn_session_head(self._conn, mask))
-
-  def link_head(self, mask):
-    return Link._wrap_link(pn_link_head(self._conn, mask))
-
-  @property
-  def work_head(self):
-    return Delivery._wrap_delivery(pn_work_head(self._conn))
-
-  @property
-  def error(self):
-    return pn_error_code(pn_connection_error(self._conn))
-
-class SessionException(ProtonException):
-  pass
-
-class Session(Endpoint):
-
-  @staticmethod
-  def _wrap_session(c_ssn):
-    """Maintain only a single instance of this class for each Session object that
-    exists in the C Engine.
-    """
-    if c_ssn is None: return None
-    py_ssn = pn_void2py(pn_session_get_context(c_ssn))
-    if py_ssn: return py_ssn
-    wrapper = Session(c_ssn)
-    return wrapper
-
-  def __init__(self, ssn):
-    Endpoint.__init__(self)
-    self._ssn = ssn
-    pn_session_set_context(self._ssn, pn_py2void(self))
-    self._links = set()
-    self.connection._sessions.add(self)
-
-  @property
-  def _children(self):
-    return self._links
-
-  def _free_resource(self):
-    pn_session_free(self._ssn)
-
-  def _released(self):
-    self._ssn = None
-
-  def free(self):
-    """Release the Session, freeing its resources.
-
-    Call this when you no longer need the session.  This will allow the
-    session's resources to be reclaimed.  Once called, you should no longer
-    reference the session.
-
-    """
-    self.connection._sessions.remove(self)
-    self._release()
-
-  def _get_cond_impl(self):
-    return pn_session_condition(self._ssn)
-
-  def _get_remote_cond_impl(self):
-    return pn_session_remote_condition(self._ssn)
-
-  def _get_incoming_capacity(self):
-    return pn_session_get_incoming_capacity(self._ssn)
-
-  def _set_incoming_capacity(self, capacity):
-    pn_session_set_incoming_capacity(self._ssn, capacity)
-
-  incoming_capacity = property(_get_incoming_capacity, _set_incoming_capacity)
-
-  @property
-  def outgoing_bytes(self):
-    return pn_session_outgoing_bytes(self._ssn)
-
-  @property
-  def incoming_bytes(self):
-    return pn_session_incoming_bytes(self._ssn)
-
-  def open(self):
-    pn_session_open(self._ssn)
-
-  def close(self):
-    self._update_cond()
-    pn_session_close(self._ssn)
-
-  def next(self, mask):
-    return Session._wrap_session(pn_session_next(self._ssn, mask))
-
-  @property
-  def state(self):
-    return pn_session_state(self._ssn)
-
-  @property
-  def connection(self):
-    return Connection._wrap_connection(pn_session_connection(self._ssn))
-
-  def sender(self, name):
-    return Link._wrap_link(pn_sender(self._ssn, name))
-
-  def receiver(self, name):
-    return Link._wrap_link(pn_receiver(self._ssn, name))
-
-class LinkException(ProtonException):
-  pass
-
-class Link(Endpoint):
-
-  SND_UNSETTLED = PN_SND_UNSETTLED
-  SND_SETTLED = PN_SND_SETTLED
-  SND_MIXED = PN_SND_MIXED
-
-  RCV_FIRST = PN_RCV_FIRST
-  RCV_SECOND = PN_RCV_SECOND
-
-  @staticmethod
-  def _wrap_link(c_link):
-    """Maintain only a single instance of this class for each Session object that
-    exists in the C Engine.
-    """
-    if c_link is None: return None
-    py_link = pn_void2py(pn_link_get_context(c_link))
-    if py_link: return py_link
-    if pn_link_is_sender(c_link):
-      wrapper = Sender(c_link)
-    else:
-      wrapper = Receiver(c_link)
-    return wrapper
-
-  def __init__(self, c_link):
-    Endpoint.__init__(self)
-    self._link = c_link
-    pn_link_set_context(self._link, pn_py2void(self))
-    self._deliveries = set()
-    self.session._links.add(self)
-
-  @property
-  def _children(self):
-    return self._deliveries
-
-  def _free_resource(self):
-    pn_link_free(self._link)
-
-  def _released(self):
-    self._link = None
-
-  def free(self):
-    """Release the Link, freeing its resources"""
-    self.session._links.remove(self)
-    self._release()
-
-  def _check(self, err):
-    if err < 0:
-      exc = EXCEPTIONS.get(err, LinkException)
-      raise exc("[%s]: %s" % (err, pn_link_error(self._link)))
-    else:
-      return err
-
-  def _get_cond_impl(self):
-    return pn_link_condition(self._link)
-
-  def _get_remote_cond_impl(self):
-    return pn_link_remote_condition(self._link)
-
-  def open(self):
-    pn_link_open(self._link)
-
-  def close(self):
-    self._update_cond()
-    pn_link_close(self._link)
-
-  @property
-  def state(self):
-    return pn_link_state(self._link)
-
-  @property
-  def source(self):
-    return Terminus(pn_link_source(self._link))
-
-  @property
-  def target(self):
-    return Terminus(pn_link_target(self._link))
-
-  @property
-  def remote_source(self):
-    return Terminus(pn_link_remote_source(self._link))
-  @property
-  def remote_target(self):
-    return Terminus(pn_link_remote_target(self._link))
-
-  @property
-  def session(self):
-    return Session._wrap_session(pn_link_session(self._link))
-
-  @property
-  def connection(self):
-    return self.session.connection
-
-  def delivery(self, tag):
-    return Delivery._wrap_delivery(pn_delivery(self._link, tag))
-
-  @property
-  def current(self):
-    return Delivery._wrap_delivery(pn_link_current(self._link))
-
-  def advance(self):
-    return pn_link_advance(self._link)
-
-  @property
-  def unsettled(self):
-    return pn_link_unsettled(self._link)
-
-  @property
-  def credit(self):
-    return pn_link_credit(self._link)
-
-  @property
-  def available(self):
-    return pn_link_available(self._link)
-
-  @property
-  def queued(self):
-    return pn_link_queued(self._link)
-
-  def next(self, mask):
-    return Link._wrap_link(pn_link_next(self._link, mask))
-
-  @property
-  def name(self):
-      return pn_link_name(self._link)
-
-  @property
-  def is_sender(self):
-    return pn_link_is_sender(self._link)
-
-  @property
-  def is_receiver(self):
-    return pn_link_is_receiver(self._link)
-
-  @property
-  def remote_snd_settle_mode(self):
-    return pn_link_remote_snd_settle_mode(self._link)
-
-  @property
-  def remote_rcv_settle_mode(self):
-    return pn_link_remote_rcv_settle_mode(self._link)
-
-  def _get_snd_settle_mode(self):
-    return pn_link_snd_settle_mode(self._link)
-  def _set_snd_settle_mode(self, mode):
-    pn_link_set_snd_settle_mode(self._link, mode)
-  snd_settle_mode = property(_get_snd_settle_mode, _set_snd_settle_mode)
-
-  def _get_rcv_settle_mode(self):
-    return pn_link_rcv_settle_mode(self._link)
-  def _set_rcv_settle_mode(self, mode):
-    pn_link_set_rcv_settle_mode(self._link, mode)
-  rcv_settle_mode = property(_get_rcv_settle_mode, _set_rcv_settle_mode)
-
-  def drained(self):
-    return pn_link_drained(self._link)
-
-  def detach(self):
-    return pn_link_detach(self._link)
-
-class Terminus(object):
-
-  UNSPECIFIED = PN_UNSPECIFIED
-  SOURCE = PN_SOURCE
-  TARGET = PN_TARGET
-  COORDINATOR = PN_COORDINATOR
-
-  NONDURABLE = PN_NONDURABLE
-  CONFIGURATION = PN_CONFIGURATION
-  DELIVERIES = PN_DELIVERIES
-
-  DIST_MODE_UNSPECIFIED = PN_DIST_MODE_UNSPECIFIED
-  DIST_MODE_COPY = PN_DIST_MODE_COPY
-  DIST_MODE_MOVE = PN_DIST_MODE_MOVE
-
-  def __init__(self, impl):
-    self._impl = impl
-
-  def _check(self, err):
-    if err < 0:
-      exc = EXCEPTIONS.get(err, LinkException)
-      raise exc("[%s]" % err)
-    else:
-      return err
-
-  def _get_type(self):
-    return pn_terminus_get_type(self._impl)
-  def _set_type(self, type):
-    self._check(pn_terminus_set_type(self._impl, type))
-  type = property(_get_type, _set_type)
-
-  def _get_address(self):
-    return pn_terminus_get_address(self._impl)
-  def _set_address(self, address):
-    self._check(pn_terminus_set_address(self._impl, address))
-  address = property(_get_address, _set_address)
-
-  def _get_durability(self):
-    return pn_terminus_get_durability(self._impl)
-  def _set_durability(self, seconds):
-    self._check(pn_terminus_set_durability(self._impl, seconds))
-  durability = property(_get_durability, _set_durability)
-
-  def _get_expiry_policy(self):
-    return pn_terminus_get_expiry_policy(self._impl)
-  def _set_expiry_policy(self, seconds):
-    self._check(pn_terminus_set_expiry_policy(self._impl, seconds))
-  expiry_policy = property(_get_expiry_policy, _set_expiry_policy)
-
-  def _get_timeout(self):
-    return pn_terminus_get_timeout(self._impl)
-  def _set_timeout(self, seconds):
-    self._check(pn_terminus_set_timeout(self._impl, seconds))
-  timeout = property(_get_timeout, _set_timeout)
-
-  def _is_dynamic(self):
-    return pn_terminus_is_dynamic(self._impl)
-  def _set_dynamic(self, dynamic):
-    self._check(pn_terminus_set_dynamic(self._impl, dynamic))
-  dynamic = property(_is_dynamic, _set_dynamic)
-
-  def _get_distribution_mode(self):
-    return pn_terminus_get_distribution_mode(self._impl)
-  def _set_distribution_mode(self, mode):
-    self._check(pn_terminus_set_distribution_mode(self._impl, mode))
-  distribution_mode = property(_get_distribution_mode, _set_distribution_mode)
-
-  @property
-  def properties(self):
-    return Data(pn_terminus_properties(self._impl))
-
-  @property
-  def capabilities(self):
-    return Data(pn_terminus_capabilities(self._impl))
-
-  @property
-  def outcomes(self):
-    return Data(pn_terminus_outcomes(self._impl))
-
-  @property
-  def filter(self):
-    return Data(pn_terminus_filter(self._impl))
-
-  def copy(self, src):
-    self._check(pn_terminus_copy(self._impl, src._impl))
-
-class Sender(Link):
-
-  def __init__(self, c_link):
-    super(Sender, self).__init__(c_link)
-
-  def offered(self, n):
-    pn_link_offered(self._link, n)
-
-  def send(self, bytes):
-    return self._check(pn_link_send(self._link, bytes))
-
-class Receiver(Link):
-
-  def __init__(self, c_link):
-    super(Receiver, self).__init__(c_link)
-
-  def flow(self, n):
-    pn_link_flow(self._link, n)
-
-  def recv(self, limit):
-    n, bytes = pn_link_recv(self._link, limit)
-    if n == PN_EOS:
-      return None
-    else:
-      self._check(n)
-      return bytes
-
-  def drain(self, n):
-    pn_link_drain(self._link, n)
-
-  def draining(self):
-    return pn_link_draining(self._link)
-
-class NamedInt(int):
-
-  values = {}
-
-  def __new__(cls, i, name):
-    ni = super(NamedInt, cls).__new__(cls, i)
-    cls.values[i] = ni
-    return ni
-
-  def __init__(self, i, name):
-    self.name = name
-
-  def __repr__(self):
-    return self.name
-
-  def __str__(self):
-    return self.name
-
-  @classmethod
-  def get(cls, i):
-    return cls.values.get(i, i)
-
-class DispositionType(NamedInt):
-  values = {}
-
-class Disposition(object):
-
-  RECEIVED = DispositionType(PN_RECEIVED, "RECEIVED")
-  ACCEPTED = DispositionType(PN_ACCEPTED, "ACCEPTED")
-  REJECTED = DispositionType(PN_REJECTED, "REJECTED")
-  RELEASED = DispositionType(PN_RELEASED, "RELEASED")
-  MODIFIED = DispositionType(PN_MODIFIED, "MODIFIED")
-
-  def __init__(self, impl, local):
-    self._impl = impl
-    self.local = local
-    self._data = None
-    self._condition = None
-    self._annotations = None
-
-  @property
-  def type(self):
-    return DispositionType.get(pn_disposition_type(self._impl))
-
-  def _get_section_number(self):
-    return pn_disposition_get_section_number(self._impl)
-  def _set_section_number(self, n):
-    pn_disposition_set_section_number(self._impl, n)
-  section_number = property(_get_section_number, _set_section_number)
-
-  def _get_section_offset(self):
-    return pn_disposition_get_section_offset(self._impl)
-  def _set_section_offset(self, n):
-    pn_disposition_set_section_offset(self._impl, n)
-  section_offset = property(_get_section_offset, _set_section_offset)
-
-  def _get_failed(self):
-    return pn_disposition_is_failed(self._impl)
-  def _set_failed(self, b):
-    pn_disposition_set_failed(self._impl, b)
-  failed = property(_get_failed, _set_failed)
-
-  def _get_undeliverable(self):
-    return pn_disposition_is_undeliverable(self._impl)
-  def _set_undeliverable(self, b):
-    pn_disposition_set_undeliverable(self._impl, b)
-  undeliverable = property(_get_undeliverable, _set_undeliverable)
-
-  def _get_data(self):
-    if self.local:
-      return self._data
-    else:
-      return dat2obj(pn_disposition_data(self._impl))
-  def _set_data(self, obj):
-    if self.local:
-      self._data = obj
-    else:
-      raise AttributeError("data attribute is read-only")
-  data = property(_get_data, _set_data)
-
-  def _get_annotations(self):
-    if self.local:
-      return self._annotations
-    else:
-      return dat2obj(pn_disposition_annotations(self._impl))
-  def _set_annotations(self, obj):
-    if self.local:
-      self._annotations = obj
-    else:
-      raise AttributeError("annotations attribute is read-only")
-  annotations = property(_get_annotations, _set_annotations)
-
-  def _get_condition(self):
-    if self.local:
-      return self._condition
-    else:
-      return cond2obj(pn_disposition_condition(self._impl))
-  def _set_condition(self, obj):
-    if self.local:
-      self._condition = obj
-    else:
-      raise AttributeError("condition attribute is read-only")
-  condition = property(_get_condition, _set_condition)
-
-class Delivery(object):
-
-  RECEIVED = Disposition.RECEIVED
-  ACCEPTED = Disposition.ACCEPTED
-  REJECTED = Disposition.REJECTED
-  RELEASED = Disposition.RELEASED
-  MODIFIED = Disposition.MODIFIED
-
-  @staticmethod
-  def _wrap_delivery(c_dlv):
-    """Maintain only a single instance of this class for each Delivery object that
-    exists in the C Engine.
-    """
-    if not c_dlv: return None
-    py_dlv = pn_void2py(pn_delivery_get_context(c_dlv))
-    if py_dlv: return py_dlv
-    wrapper = Delivery(c_dlv)
-    return wrapper
-
-  def __init__(self, dlv):
-    self._dlv = dlv
-    pn_delivery_set_context(self._dlv, pn_py2void(self))
-    self.local = Disposition(pn_delivery_local(self._dlv), True)
-    self.remote = Disposition(pn_delivery_remote(self._dlv), False)
-    self.link._deliveries.add(self)
-
-  def __del__(self):
-    self._release()
-
-  def _release(self):
-    """Release the underlying C Engine resource."""
-    if self._dlv:
-      pn_delivery_set_context(self._dlv, pn_py2void(None))
-      pn_delivery_settle(self._dlv)
-      self._dlv = None
-
-  @property
-  def released(self):
-    return self._dlv is None
-
-  @property
-  def tag(self):
-    return pn_delivery_tag(self._dlv)
-
-  @property
-  def writable(self):
-    return pn_delivery_writable(self._dlv)
-
-  @property
-  def readable(self):
-    return pn_delivery_readable(self._dlv)
-
-  @property
-  def updated(self):
-    return pn_delivery_updated(self._dlv)
-
-  def update(self, state):
-    obj2dat(self.local._data, pn_disposition_data(self.local._impl))
-    obj2dat(self.local._annotations, pn_disposition_annotations(self.local._impl))
-    obj2cond(self.local._condition, pn_disposition_condition(self.local._impl))
-    pn_delivery_update(self._dlv, state)
-
-  @property
-  def pending(self):
-    return pn_delivery_pending(self._dlv)
-
-  @property
-  def partial(self):
-    return pn_delivery_partial(self._dlv)
-
-  @property
-  def local_state(self):
-    return DispositionType.get(pn_delivery_local_state(self._dlv))
-
-  @property
-  def remote_state(self):
-    return DispositionType.get(pn_delivery_remote_state(self._dlv))
-
-  @property
-  def settled(self):
-    return pn_delivery_settled(self._dlv)
-
-  def settle(self):
-    """Release the delivery"""
-    self.link._deliveries.remove(self)
-    self._release()
-
-  @property
-  def work_next(self):
-    return Delivery._wrap_delivery(pn_work_next(self._dlv))
-
-  @property
-  def link(self):
-    return Link._wrap_link(pn_delivery_link(self._dlv))
-
-class TransportException(ProtonException):
-  pass
-
-class Transport(object):
-
-  TRACE_OFF = PN_TRACE_OFF
-  TRACE_DRV = PN_TRACE_DRV
-  TRACE_FRM = PN_TRACE_FRM
-  TRACE_RAW = PN_TRACE_RAW
-
-  CLIENT = 1
-  SERVER = 2
-
-  @staticmethod
-  def _wrap_transport(c_trans):
-    if not c_trans: return None
-    wrapper = Transport(_trans=c_trans)
-    return wrapper
-
-  def __init__(self, mode=None, _trans=None):
-    if not mode and not _trans:
-      self._trans = pn_transport()
-    elif not mode:
-      self._shared_trans = True
-      self._trans = _trans
-    elif mode==Transport.CLIENT:
-      self._trans = pn_transport()
-    elif mode==Transport.SERVER:
-      self._trans = pn_transport()
-      pn_transport_set_server(self._trans)
-    else:
-      raise TransportException("Cannot initialise Transport from mode: %s" % str(mode))
-    self._sasl = None
-    self._ssl = None
-
-  def __del__(self):
-    if hasattr(self, "_trans"):
-      if not hasattr(self, "_shared_trans"):
-        pn_transport_free(self._trans)
-        if hasattr(self, "_sasl") and self._sasl:
-            # pn_transport_free deallocs the C sasl associated with the
-            # transport, so erase the reference if a SASL object was used.
-            self._sasl._sasl = None
-            self._sasl = None
-        if hasattr(self, "_ssl") and self._ssl:
-            # ditto the owned c SSL object
-            self._ssl._ssl = None
-            self._ssl = None
-      del self._trans
-
-  def _check(self, err):
-    if err < 0:
-      exc = EXCEPTIONS.get(err, TransportException)
-      raise exc("[%s]: %s" % (err, pn_error_text(pn_transport_error(self._trans))))
-    else:
-      return err
-
-  def bind(self, connection):
-    """Assign a connection to the transport"""
-    self._check(pn_transport_bind(self._trans, connection._conn))
-    # keep python connection from being garbage collected:
-    self._connection = connection
-
-  def unbind(self):
-    """Release the connection"""
-    self._check(pn_transport_unbind(self._trans))
-    self._connection = None
-
-  def trace(self, n):
-    pn_transport_trace(self._trans, n)
-
-  def tick(self, now):
-    """Process any timed events (like heartbeat generation).
-    now = seconds since epoch (float).
-    """
-    return millis2secs(pn_transport_tick(self._trans, secs2millis(now)))
-
-  def capacity(self):
-    c = pn_transport_capacity(self._trans)
-    if c >= PN_EOS:
-      return c
-    else:
-      return self._check(c)
-
-  def push(self, bytes):
-    n = self._check(pn_transport_push(self._trans, bytes))
-    if n != len(bytes):
-      raise OverflowError("unable to process all bytes")
-
-  def close_tail(self):
-    self._check(pn_transport_close_tail(self._trans))
-
-  def pending(self):
-    p = pn_transport_pending(self._trans)
-    if p >= PN_EOS:
-      return p
-    else:
-      return self._check(p)
-
-  def peek(self, size):
-    cd, out = pn_transport_peek(self._trans, size)
-    if cd == PN_EOS:
-      return None
-    else:
-      self._check(cd)
-      return out
-
-  def pop(self, size):
-    pn_transport_pop(self._trans, size)
-
-  def close_head(self):
-    self._check(pn_transport_close_head(self._trans))
-
-  @property
-  def closed(self):
-    return pn_transport_closed(self._trans)
-
-  # AMQP 

<TRUNCATED>

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message