You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2017/01/10 15:58:42 UTC
[21/55] [partial] qpid-proton-j git commit: PROTON-1385: retain
proton-j content only, the rest remains in the other repo at:
https://git-wip-us.apache.org/repos/asf/qpid-proton.git
http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/2f85988e/proton-c/bindings/python/proton/__init__.py
----------------------------------------------------------------------
diff --git a/proton-c/bindings/python/proton/__init__.py b/proton-c/bindings/python/proton/__init__.py
deleted file mode 100644
index d3f6922..0000000
--- a/proton-c/bindings/python/proton/__init__.py
+++ /dev/null
@@ -1,4298 +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 __future__ import absolute_import
-
-from cproton import *
-from .wrapper import Wrapper
-from proton import _compat
-
-import weakref, socket, sys, threading
-
-try:
- import uuid
-
- def generate_uuid():
- return uuid.uuid4()
-
-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, time
- rand = random.Random()
- rand.seed((os.getpid(), time.time(), socket.gethostname()))
- def random_uuid():
- data = [rand.randint(0, 255) for i in xrange(16)]
-
- # From RFC4122, the version bits are set to 0100
- data[6] &= 0x0F
- data[6] |= 0x40
-
- # From RFC4122, the top two bits of byte 8 get set to 01
- data[8] &= 0x3F
- data[8] |= 0x80
- return "".join(map(chr, data))
-
- def uuid4():
- return uuid.UUID(bytes=random_uuid())
-
- def generate_uuid():
- return uuid4()
-
-#
-# Hacks to provide Python2 <---> Python3 compatibility
-#
-try:
- bytes()
-except NameError:
- bytes = str
-try:
- long()
-except NameError:
- long = int
-try:
- unicode()
-except NameError:
- unicode = str
-
-
-VERSION_MAJOR = PN_VERSION_MAJOR
-VERSION_MINOR = PN_VERSION_MINOR
-VERSION_POINT = PN_VERSION_POINT
-VERSION = (VERSION_MAJOR, VERSION_MINOR, VERSION_POINT)
-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")
-MODIFIED = Constant("MODIFIED")
-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_MODIFIED: MODIFIED,
- 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, MODIFIED, 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, unicode2utf8(pattern), unicode2utf8(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, unicode2utf8(pattern), unicode2utf8(address)))
-
- def selectable(self):
- return Selectable.wrap(pn_messenger_selectable(self._mng))
-
- @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
- """
-
- DEFAULT_PRIORITY = PN_DEFAULT_PRIORITY
-
- def __init__(self, body=None, **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 = body
- for k,v in _compat.iteritems(kwargs):
- 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 _compat.INT_TYPES:
- 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 utf82unicode(pn_message_get_address(self._msg))
-
- def _set_address(self, value):
- self._check(pn_message_set_address(self._msg, unicode2utf8(value)))
-
- address = property(_get_address, _set_address,
- doc="""
-The address of the message.
-""")
-
- def _get_subject(self):
- return utf82unicode(pn_message_get_subject(self._msg))
-
- def _set_subject(self, value):
- self._check(pn_message_set_subject(self._msg, unicode2utf8(value)))
-
- subject = property(_get_subject, _set_subject,
- doc="""
-The subject of the message.
-""")
-
- def _get_reply_to(self):
- return utf82unicode(pn_message_get_reply_to(self._msg))
-
- def _set_reply_to(self, value):
- self._check(pn_message_set_reply_to(self._msg, unicode2utf8(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 _compat.INT_TYPES:
- 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 symbol(utf82unicode(pn_message_get_content_type(self._msg)))
-
- def _set_content_type(self, value):
- self._check(pn_message_set_content_type(self._msg, unicode2utf8(value)))
-
- content_type = property(_get_content_type, _set_content_type,
- doc="""
-The content-type of the message.
-""")
-
- def _get_content_encoding(self):
- return symbol(utf82unicode(pn_message_get_content_encoding(self._msg)))
-
- def _set_content_encoding(self, value):
- self._check(pn_message_set_content_encoding(self._msg, unicode2utf8(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 utf82unicode(pn_message_get_group_id(self._msg))
-
- def _set_group_id(self, value):
- self._check(pn_message_set_group_id(self._msg, unicode2utf8(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 utf82unicode(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, unicode2utf8(value)))
-
- reply_to_group_id = property(_get_reply_to_group_id, _set_reply_to_group_id,
- doc="""
-The group-id for any replies.
-""")
-
- 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))
- self._post_decode()
-
- def send(self, sender, tag=None):
- dlv = sender.delivery(tag or sender.delivery_tag())
- encoded = self.encode()
- sender.stream(encoded)
- sender.advance()
- if sender.snd_settle_mode == Link.SND_SETTLED:
- dlv.settle()
- return dlv
-
- def recv(self, link):
- """
- Receives and decodes the message content for the current delivery
- from the link. Upon success it will return the current delivery
- for the link. If there is no current delivery, or if the current
- delivery is incomplete, or if the link is not a receiver, it will
- return None.
-
- @type link: Link
- @param link: the link to receive a message from
- @return the delivery associated with the decoded message (or None)
-
- """
- if link.is_sender: return None
- dlv = link.current
- if not dlv or dlv.partial: return None
- dlv.encoded = link.recv(dlv.pending)
- link.advance()
- # the sender has already forgotten about the delivery, so we might
- # as well too
- if link.remote_snd_settle_mode == Link.SND_SETTLED:
- dlv.settle()
- self.decode(dlv.encoded)
- return dlv
-
- 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)
-
-_DEFAULT = object()
-
-class Selectable(Wrapper):
-
- @staticmethod
- def wrap(impl):
- if impl is None:
- return None
- else:
- return Selectable(impl)
-
- def __init__(self, impl):
- Wrapper.__init__(self, impl, pn_selectable_attachments)
-
- def _init(self):
- pass
-
- def fileno(self, fd = _DEFAULT):
- if fd is _DEFAULT:
- return pn_selectable_get_fd(self._impl)
- elif fd is None:
- pn_selectable_set_fd(self._impl, PN_INVALID_SOCKET)
- else:
- pn_selectable_set_fd(self._impl, fd)
-
- def _is_reading(self):
- return pn_selectable_is_reading(self._impl)
-
- def _set_reading(self, val):
- pn_selectable_set_reading(self._impl, bool(val))
-
- reading = property(_is_reading, _set_reading)
-
- def _is_writing(self):
- return pn_selectable_is_writing(self._impl)
-
- def _set_writing(self, val):
- pn_selectable_set_writing(self._impl, bool(val))
-
- writing = property(_is_writing, _set_writing)
-
- def _get_deadline(self):
- tstamp = pn_selectable_get_deadline(self._impl)
- if tstamp:
- return millis2secs(tstamp)
- else:
- return None
-
- def _set_deadline(self, deadline):
- pn_selectable_set_deadline(self._impl, secs2millis(deadline))
-
- deadline = property(_get_deadline, _set_deadline)
-
- def readable(self):
- pn_selectable_readable(self._impl)
-
- def writable(self):
- pn_selectable_writable(self._impl)
-
- def expired(self):
- pn_selectable_expired(self._impl)
-
- def _is_registered(self):
- return pn_selectable_is_registered(self._impl)
-
- def _set_registered(self, registered):
- 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):
- return pn_selectable_is_terminal(self._impl)
-
- def terminate(self):
- pn_selectable_terminate(self._impl)
-
- def release(self):
- pn_selectable_release(self._impl)
-
-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 byte(int):
-
- def __repr__(self):
- return "byte(%s)" % int.__repr__(self)
-
-class short(int):
-
- def __repr__(self):
- return "short(%s)" % int.__repr__(self)
-
-class int32(int):
-
- def __repr__(self):
- return "int32(%s)" % int.__repr__(self)
-
-class ubyte(int):
-
- def __repr__(self):
- return "ubyte(%s)" % int.__repr__(self)
-
-class ushort(int):
-
- def __repr__(self):
- return "ushort(%s)" % int.__repr__(self)
-
-class uint(long):
-
- def __repr__(self):
- return "uint(%s)" % long.__repr__(self)
-
-class float32(float):
-
- def __repr__(self):
- return "float32(%s)" % float.__repr__(self)
-
-class decimal32(int):
-
- def __repr__(self):
- return "decimal32(%s)" % int.__repr__(self)
-
-class decimal64(long):
-
- def __repr__(self):
- return "decimal64(%s)" % long.__repr__(self)
-
-class decimal128(bytes):
-
- def __repr__(self):
- return "decimal128(%s)" % bytes.__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 __iter__(self):
- return iter(self.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 _compat.INT_TYPES:
- 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 encoded_size(self):
- """
- Returns the size in bytes needed to encode the data in AMQP format.
- """
- return pn_data_encoded_size(self._data)
-
- 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, bytes(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.encode('ascii')))
-
- 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()
- >>> data.next()
- >>> print data.get_symbol()
- >>> data.next()
- >>> 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 ubyte(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 byte(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 ushort(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 short(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 uint(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 int32(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(_compat.unichar(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 long(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 float32(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 decimal32(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 decimal64(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 decimal128(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).decode('ascii'))
-
- 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,
- ubyte: put_ubyte,
- ushort: put_ushort,
- uint: put_uint,
- ulong: put_ulong,
- byte: put_byte,
- short: put_short,
- int32: put_int,
- long: put_long,
- float32: put_float,
- float: put_double,
- decimal32: put_decimal32,
- decimal64: put_decimal64,
- decimal128: put_decimal128,
- char: put_char,
- timestamp: put_timestamp,
- uuid.UUID: put_uuid,
- bytes: put_binary,
- unicode: put_string,
- symbol: put_symbol,
- list: put_sequence,
- tuple: put_sequence,
- dict: put_dict,
- Described: put_py_described,
- Array: put_py_array
- }
- # for python 3.x, long is merely an alias for int, but for python 2.x
- # we need to add an explicit int since it is a different type
- if int not in put_mappings:
- put_mappings[int] = put_int
- # For python 3.x use 'memoryview', for <=2.5 use 'buffer'. Python >=2.6 has both.
- if getattr(__builtins__, 'memoryview', None):
- put_mappings[memoryview] = put_binary
- if getattr(__builtins__, 'buffer', None):
- put_mappings[buffer] = put_binary
-
- 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
-
- 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!"
-
- def _get_handler(self):
- from . import reactor
- ractor = reactor.Reactor.wrap(pn_object_reactor(self._impl))
- if ractor:
- on_error = ractor.on_error
- else:
- on_error = None
- record = self._get_attachments()
- return WrappedHandler.wrap(pn_record_get_handler(record), on_error)
-
- def _set_handler(self, handler):
- from . import reactor
- ractor = reactor.Reactor.wrap(pn_object_reactor(self._impl))
- if ractor:
- on_error = ractor.on_error
- else:
- on_error = None
- impl = _chandler(handler, on_error)
- record = self._get_attachments()
- pn_record_set_handler(record, impl)
- pn_decref(impl)
-
- handler = property(_get_handler, _set_handler)
-
- @property
- def transport(self):
- return self.connection.transport
-
-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
-
-def timeout2millis(secs):
- if secs is None: return PN_MILLIS_MAX
- return secs2millis(secs)
-
-def millis2timeout(millis):
- if millis == PN_MILLIS_MAX: return None
- return millis2secs(millis)
-
-def unicode2utf8(string):
- """Some Proton APIs expect a null terminated string. Convert python text
- types to UTF8 to avoid zero bytes introduced by other multi-byte encodings.
- This method will throw if the string cannot be converted.
- """
- if string is None:
- return None
- if _compat.IS_PY2:
- if isinstance(string, unicode):
- return string.encode('utf-8')
- elif isinstance(string, str):
- return string
- else:
- # decoding a string results in bytes
- if isinstance(string, str):
- string = string.encode('utf-8')
- # fall through
- if isinstance(string, bytes):
- return string.decode('utf-8')
- raise TypeError("Unrecognized string type: %r (%s)" % (string, type(string)))
-
-def utf82unicode(string):
- """Covert C strings returned from proton-c into python unicode"""
- if string is None:
- return None
- if isinstance(string, _compat.TEXT_TYPES):
- # already unicode
- return string
- elif isinstance(string, _compat.BINARY_TYPES):
- return string.decode('utf8')
- else:
- raise TypeError("Unrecognized string type")
-
-class Connection(Wrapper, Endpoint):
- """
- A representation of an AMQP connection
- """
-
- @staticmethod
- def wrap(impl):
- if impl is None:
- return None
- else:
- return Connection(impl)
-
- def __init__(self, impl = pn_connection):
- Wrapper.__init__(self, impl, pn_connection_attachments)
-
- def _init(self):
- Endpoint._init(self)
- self.offered_capabilities = None
- self.desired_capabilities = None
- self.properties = None
-
- def _get_attachments(self):
- return pn_connection_attachments(self._impl)
-
- @property
- def connection(self):
- return self
-
- @property
- def transport(self):
- return Transport.wrap(pn_connection_transport(self._impl))
-
- def _check(self, err):
- if err < 0:
- exc = EXCEPTIONS.get(err, ConnectionException)
- raise exc("[%s]: %s" % (err, pn_connection_error(self._impl)))
- else:
- return err
-
- def _get_cond_impl(self):
- return pn_connection_condition(self._impl)
-
- def _get_remote_cond_impl(self):
- return pn_connection_remote_condition(self._impl)
-
- def collect(self, collector):
- if collector is None:
- pn_connection_collect(self._impl, None)
- else:
- pn_connection_collect(self._impl, collector._impl)
- self._collector = weakref.ref(collector)
-
- def _get_container(self):
- return utf82unicode(pn_connection_get_container(self._impl))
- def _set_container(self, name):
- return pn_connection_set_container(self._impl, unicode2utf8(name))
-
- container = property(_get_container, _set_container)
-
- def _get_hostname(self):
- return utf82unicode(pn_connection_get_hostname(self._impl))
- def _set_hostname(self, name):
- return pn_connection_set_hostname(self._impl, unicode2utf8(name))
-
- hostname = property(_get_hostname, _set_hostname,
- doc="""
-Set the name of the host (either fully qualified or relative) to which this
-connection is connecting to. This information may be used by the remote
-peer to determine the correct back-end service to connect the client to.
-This value will be sent in the Open performative, and will be used by SSL
-and SASL layers to identify the peer.
-""")
-
- def _get_user(self):
- return utf82unicode(pn_connection_get_user(self._impl))
- def _set_user(self, name):
- return pn_connection_set_user(self._impl, unicode2utf8(name))
-
- user = property(_get_user, _set_user)
-
- def _get_password(self):
- return None
- def _set_password(self, name):
- return pn_connection_set_password(self._impl, unicode2utf8(name))
-
- password = property(_get_password, _set_password)
-
- @property
- def remote_container(self):
- """The container identifier specified by the remote peer for this connection."""
- return pn_connection_remote_container(self._impl)
-
- @property
- def remote_hostname(self):
- """The hostname specified by the remote peer for this connection."""
- return pn_connection_remote_hostname(self._impl)
-
- @property
- def remote_offered_capabilities(self):
- """The capabilities offered by the remote peer for this connection."""
- return dat2obj(pn_connection_remote_offered_capabilities(self._impl))
-
- @property
- def remote_desired_capabilities(self):
- """The capabilities desired by the remote peer for this connection."""
- return dat2obj(pn_connection_remote_desired_capabilities(self._impl))
-
- @property
- def remote_properties(self):
- """The properties specified by the remote peer for this connection."""
- return dat2obj(pn_connection_remote_properties(self._impl))
-
- def open(self):
- """
- Opens the connection.
-
- In more detail, this moves the local state of the connection to
- the ACTIVE state and triggers an open frame to be sent to the
- peer. A connection is fully active once both peers have opened it.
- """
- obj2dat(self.offered_capabilities,
- pn_connection_offered_capabilities(self._impl))
- obj2dat(self.desired_capabilities,
- pn_connection_desired_capabilities(self._impl))
- obj2dat(self.properties, pn_connection_properties(self._impl))
- pn_connection_open(self._impl)
-
- def close(self):
- """
- Closes the connection.
-
- In more detail, this moves the local state of the connection to
- the CLOSED state and triggers a close frame to be sent to the
- peer. A connection is fully closed once both peers have closed it.
- """
- self._update_cond()
- pn_connection_close(self._impl)
-
- @property
- def state(self):
- """
- The state of the connection as a bit field. The state has a local
- and a remote component. Each of these can be in one of three
- states: UNINIT, ACTIVE or CLOSED. These can be tested by masking
- against LOCAL_UNINIT, LOCAL_ACTIVE, LOCAL_CLOSED, REMOTE_UNINIT,
- REMOTE_ACTIVE and REMOTE_CLOSED.
- """
- return pn_connection_state(self._impl)
-
- def session(self):
- """
- Returns a new session on this connection.
- """
- ssn = pn_session(self._impl)
- if ssn is None:
- raise(SessionException("Session allocation failed."))
- else:
- return Session(ssn)
-
- def session_head(self, mask):
- return Session.wrap(pn_session_head(self._impl, mask))
-
- def link_head(self, mask):
- return Link.wrap(pn_link_head(self._impl, mask))
-
- @property
- def work_head(self):
- return Delivery.wrap(pn_work_head(self._impl))
-
- @property
- def error(self):
- return pn_error_code(pn_connection_error(self._impl))
-
- def free(self):
- pn_connection_release(self._impl)
-
-class SessionException(ProtonException):
- pass
-
-class Session(Wrapper, Endpoint):
-
- @staticmethod
- def wrap(impl):
- if impl is None:
- return None
- else:
- return Session(impl)
-
- def __init__(self, impl):
- Wrapper.__init__(self, impl, pn_session_attachments)
-
- def _get_attachments(self):
- return pn_session_attachments(self._impl)
-
- def _get_cond_impl(self):
- return pn_session_condition(self._impl)
-
- def _get_remote_cond_impl(self):
- return pn_session_remote_condition(self._impl)
-
- def _get_incoming_capacity(self):
- return pn_session_get_incoming_capacity(self._impl)
-
- def _set_incoming_capacity(self, capacity):
- pn_session_set_incoming_capacity(self._impl, capacity)
-
- incoming_capacity = property(_get_incoming_capacity, _set_incoming_capacity)
-
- def _get_outgoing_window(self):
- return pn_session_get_outgoing_window(self._impl)
-
- def _set_outgoing_window(self, window):
- pn_session_set_outgoing_window(self._impl, window)
-
- outgoing_window = property(_get_outgoing_window, _set_outgoing_window)
-
- @property
- def outgoing_bytes(self):
- return pn_session_outgoing_bytes(self._impl)
-
- @property
- def incoming_bytes(self):
- return pn_session_incoming_bytes(self._impl)
-
- def open(self):
- pn_session_open(self._impl)
-
- def close(self):
- self._update_cond()
- pn_session_close(self._impl)
-
- def next(self, mask):
- return Session.wrap(pn_session_next(self._impl, mask))
-
- @property
- def state(self):
- return pn_session_state(self._impl)
-
- @property
- def connection(self):
- return Connection.wrap(pn_session_connection(self._impl))
-
- def sender(self, name):
- return Sender(pn_sender(self._impl, unicode2utf8(name)))
-
- def receiver(self, name):
- return Receiver(pn_receiver(self._impl, unicode2utf8(name)))
-
- def free(self):
- pn_session_free(self._impl)
-
-class LinkException(ProtonException):
- pass
-
-class Link(Wrapper, Endpoint):
- """
- A representation of an AMQP link, of which there are two concrete
- implementations, Sender and Receiver.
- """
-
- 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(impl):
- if impl is None: return None
- if pn_link_is_sender(impl):
- return Sender(impl)
- else:
- return Receiver(impl)
-
- def __init__(self, impl):
- Wrapper.__init__(self, impl, pn_link_attachments)
-
- def _get_attachments(self):
- return pn_link_attachments(self._impl)
-
- def _check(self, err):
- if err < 0:
- exc = EXCEPTIONS.get(err, LinkException)
- raise exc("[%s]: %s" % (err, pn_error_text(pn_link_error(self._impl))))
- else:
- return err
-
- def _get_cond_impl(self):
- return pn_link_condition(self._impl)
-
- def _get_remote_cond_impl(self):
- return pn_link_remote_condition(self._impl)
-
- def open(self):
- """
- Opens the link.
-
- In more detail, this moves the local state of the link to the
- ACTIVE state and triggers an attach frame to be sent to the
- peer. A link is fully active once both peers have attached it.
- """
- pn_link_open(self._impl)
-
- def close(self):
- """
- Closes the link.
-
- In more detail, this moves the local state of the link to the
- CLOSED state and triggers an detach frame (with the closed flag
- set) to be sent to the peer. A link is fully closed once both
- peers have detached it.
- """
- self._update_cond()
- pn_link_close(self._impl)
-
- @property
- def state(self):
- """
- The state of the link as a bit field. The state has a local
- and a remote component. Each of these can be in one of three
- states: UNINIT, ACTIVE or CLOSED. These can be tested by masking
- against LOCAL_UNINIT, LOCAL_ACTIVE, LOCAL_CLOSED, REMOTE_UNINIT,
- REMOTE_ACTIVE and REMOTE_CLOSED.
- """
- return pn_link_state(self._impl)
-
- @property
- def source(self):
- """The source of the link as described by the local peer."""
- return Terminus(pn_link_source(self._impl))
-
- @property
- def target(self):
- """The target of the link as described by the local peer."""
- return Terminus(pn_link_target(self._impl))
-
- @property
- def remote_source(self):
- """The source of the link as described by the remote peer."""
- return Terminus(pn_link_remote_source(self._impl))
- @property
- def remote_target(self):
- """The target of the link as described by the remote peer."""
- return Terminus(pn_link_remote_target(self._impl))
-
- @property
- def session(self):
- return Session.wrap(pn_link_session(self._impl))
-
- @property
- def connection(self):
- """The connection on which this link was attached."""
- return self.session.connection
-
- def delivery(self, tag):
- return Delivery(pn_delivery(self._impl, tag))
-
- @property
- def current(self):
- return Delivery.wrap(pn_link_current(self._impl))
-
- def advance(self):
- return pn_link_advance(self._impl)
-
- @property
- def unsettled(self):
- return pn_link_unsettled(self._impl)
-
- @property
- def credit(self):
- """The amount of oustanding credit on this link."""
- return pn_link_credit(self._impl)
-
- @property
- def available(self):
- return pn_link_available(self._impl)
-
- @property
- def queued(self):
- return pn_link_queued(self._impl)
-
- def next(self, mask):
- return Link.wrap(pn_link_next(self._impl, mask))
-
- @property
- def name(self):
- """Returns the name of the link"""
- return utf82unicode(pn_link_name(self._impl))
-
- @property
- def is_sender(self):
- """Returns true if this link is a sender."""
- return pn_link_is_sender(self._impl)
-
- @property
- def is_receiver(self):
- """Returns true if this link is a receiver."""
- return pn_link_is_receiver(self._impl)
-
- @property
- def remote_snd_settle_mode(self):
- return pn_link_remote_snd_settle_mode(self._impl)
-
- @property
- def remote_rcv_settle_mode(self):
- return pn_link_remote_rcv_settle_mode(self._impl)
-
- def _get_snd_settle_mode(self):
- return pn_link_snd_settle_mode(self._impl)
- def _set_snd_settle_mode(self, mode):
- pn_link_set_snd_settle_mode(self._impl, 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._impl)
- def _set_rcv_settle_mode(self, mode):
- pn_link_set_rcv_settle_mode(self._impl, mode)
- rcv_settle_mode = property(_get_rcv_settle_mode, _set_rcv_settle_mode)
-
- def _get_drain(self):
- return pn_link_get_drain(self._impl)
-
- def _set_drain(self, b):
- pn_link_set_drain(self._impl, bool(b))
-
- drain_mode = property(_get_drain, _set_drain)
-
- def drained(self):
- return pn_link_drained(self._impl)
-
- @property
- def remote_max_message_size(self):
- return pn_link_remote_max_message_size(self._impl)
-
- def _get_max_message_size(self):
- return pn_link_max_message_size(self._impl)
- def _set_max_message_size(self, mode):
- pn_link_set_max_message_size(self._impl, mode)
- max_message_size = property(_get_max_message_size, _set_max_message_size)
-
- def detach(self):
- return pn_link_detach(self._impl)
-
- def free(self):
- pn_link_free(self._impl)
-
-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
-
- EXPIRE_WITH_LINK = PN_EXPIRE_WITH_LINK
- EXPIRE_WITH_SESSION = PN_EXPIRE_WITH_SESSION
- EXPIRE_WITH_CONNECTION = PN_EXPIRE_WITH_CONNECTION
- EXPIRE_NEVER = PN_EXPIRE_NEVER
-
- 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):
- """The address that identifies the source or target node"""
- return utf82unicode(pn_terminus_get_address(self._impl))
- def _set_address(self, address):
- self._check(pn_terminus_set_address(self._impl, unicode2utf8(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):
- """Indicates whether the source or target node was dynamically
- created"""
- 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):
- """Properties of a dynamic source or target."""
- return Data(pn_terminus_properties(self._impl))
-
- @property
- def capabilities(self):
- """Capabilities of the source or target."""
- return Data(pn_terminus_capabilities(self._impl))
-
- @property
- def outcomes(self):
- return Data(pn_terminus_outcomes(self._impl))
-
- @property
- def filter(self):
- """A filter on a source allows the set of messages transfered over
- the link to be restricted"""
- return Data(pn_terminus_filter(self._impl))
-
- def copy(self, src):
- self._check(pn_terminus_copy(self._impl, src._impl))
-
-class Sender(Link):
- """
- A link over which messages are sent.
- """
-
- def offered(self, n):
- pn_link_offered(self._impl, n)
-
- def stream(self, data):
- """
- Send specified data as part of the current delivery
-
- @type data: binary
- @param data: data to send
- """
- return self._check(pn_link_send(self._impl, data))
-
- def send(self, obj, tag=None):
- """
- Send specified object over this sender; the object is expected to
- have a send() method on it that takes the sender and an optional
- tag as arguments.
-
- Where the object is a Message, this will send the message over
- this link, creating a new delivery for the purpose.
- """
- if hasattr(obj, 'send'):
- return obj.send(self, tag=tag)
- else:
- # treat object as bytes
- return self.stream(obj)
-
- def delivery_tag(self):
- if not hasattr(self, 'tag_generator'):
- def simple_tags():
- count = 1
- while True:
- yield str(count)
- count += 1
- self.tag_generator = simple_tags()
- return next(self.tag_generator)
-
-class Receiver(Link):
- """
- A link over which messages are received.
- """
-
- def flow(self, n):
- """Increases the credit issued to the remote sender by the specified number of
<TRUNCATED>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org