You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2014/12/12 15:27:25 UTC
[5/5] qpid-proton git commit: updated python bindings to use a
simpler and more robust wrapping strategy
updated python bindings to use a simpler and more robust wrapping strategy
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/2ada5d2b
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/2ada5d2b
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/2ada5d2b
Branch: refs/heads/master
Commit: 2ada5d2bc4a1ee74d5ee57e84a6ecc0b418e7dc2
Parents: 35bc4d9
Author: Rafael Schloming <rh...@alum.mit.edu>
Authored: Fri Dec 12 08:15:41 2014 -0500
Committer: Rafael Schloming <rh...@alum.mit.edu>
Committed: Fri Dec 12 09:17:59 2014 -0500
----------------------------------------------------------------------
proton-c/bindings/python/proton/__init__.py | 543 +++++++------------
proton-c/bindings/python/proton/wrapper.py | 88 +++
proton-c/include/proton/connection.h | 11 +
proton-c/include/proton/link.h | 12 +
proton-c/include/proton/session.h | 12 +
proton-c/src/engine/engine-internal.h | 12 +-
proton-c/src/engine/engine.c | 280 +++++++---
proton-c/src/transport/transport.c | 35 +-
.../apache/qpid/proton/engine/Connection.java | 2 +
.../qpid/proton/engine/impl/ConnectionImpl.java | 2 +-
.../qpid/proton/engine/impl/DeliveryImpl.java | 2 +-
.../qpid/proton/engine/impl/LinkImpl.java | 6 +
.../qpid/proton/engine/impl/TransportImpl.java | 12 +-
proton-j/src/main/resources/cengine.py | 45 +-
proton-j/src/main/resources/cobject.py | 17 +
tests/python/proton_tests/engine.py | 69 +--
tests/python/proton_tests/sasl.py | 14 +-
17 files changed, 672 insertions(+), 490 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/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
index a4ec1c6..ae520f6 100644
--- a/proton-c/bindings/python/proton/__init__.py
+++ b/proton-c/bindings/python/proton/__init__.py
@@ -31,6 +31,7 @@ The proton APIs consist of the following classes:
"""
from cproton import *
+from wrapper import Wrapper
import weakref, re, socket
try:
@@ -2183,19 +2184,8 @@ class Endpoint(object):
LOCAL_CLOSED = PN_LOCAL_CLOSED
REMOTE_CLOSED = PN_REMOTE_CLOSED
- def __init__(self):
+ def _init(self):
self.condition = None
- self._release_invoked = False
-
- def _release(self):
- """Release the underlying C Engine resource."""
- if not self._release_invoked:
- conn = self.connection # Releasing the children may break self.connection.
- for c in self._children:
- c._release()
- self._free_resource()
- conn._releasing(self)
- self._release_invoked = True
def _update_cond(self):
obj2cond(self.condition, self._get_cond_impl())
@@ -2266,236 +2256,193 @@ def secs2millis(secs):
def millis2secs(millis):
return float(millis)/1000.0
-class Connection(Endpoint):
+class Connection(Wrapper, 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
+ def wrap(impl):
+ if impl is None:
+ return None
else:
- self._conn = pn_connection()
- pn_connection_set_context(self._conn, pn_py2void(self))
+ 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
- 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)
- del self._conn
-
- def _releasing(self, child):
- coll = getattr(self, "_collector", None)
- if coll: coll = coll()
- if coll:
- coll._contexts.add(child)
+ @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._conn)))
+ raise exc("[%s]: %s" % (err, pn_connection_error(self._impl)))
else:
return err
def _get_cond_impl(self):
- return pn_connection_condition(self._conn)
+ return pn_connection_condition(self._impl)
def _get_remote_cond_impl(self):
- return pn_connection_remote_condition(self._conn)
+ return pn_connection_remote_condition(self._impl)
def collect(self, collector):
if collector is None:
- pn_connection_collect(self._conn, None)
+ pn_connection_collect(self._impl, None)
else:
- pn_connection_collect(self._conn, collector._impl)
+ pn_connection_collect(self._impl, collector._impl)
self._collector = weakref.ref(collector)
def _get_container(self):
- return pn_connection_get_container(self._conn)
+ return pn_connection_get_container(self._impl)
def _set_container(self, name):
- return pn_connection_set_container(self._conn, name)
+ return pn_connection_set_container(self._impl, name)
container = property(_get_container, _set_container)
def _get_hostname(self):
- return pn_connection_get_hostname(self._conn)
+ return pn_connection_get_hostname(self._impl)
def _set_hostname(self, name):
- return pn_connection_set_hostname(self._conn, name)
+ return pn_connection_set_hostname(self._impl, name)
hostname = property(_get_hostname, _set_hostname)
@property
def remote_container(self):
- return pn_connection_remote_container(self._conn)
+ return pn_connection_remote_container(self._impl)
@property
def remote_hostname(self):
- return pn_connection_remote_hostname(self._conn)
+ return pn_connection_remote_hostname(self._impl)
@property
def remote_offered_capabilities(self):
- return dat2obj(pn_connection_remote_offered_capabilities(self._conn))
+ return dat2obj(pn_connection_remote_offered_capabilities(self._impl))
@property
def remote_desired_capabilities(self):
- return dat2obj(pn_connection_remote_desired_capabilities(self._conn))
+ return dat2obj(pn_connection_remote_desired_capabilities(self._impl))
@property
def remote_properties(self):
- return dat2obj(pn_connection_remote_properties(self._conn))
+ return dat2obj(pn_connection_remote_properties(self._impl))
def open(self):
obj2dat(self.offered_capabilities,
- pn_connection_offered_capabilities(self._conn))
+ pn_connection_offered_capabilities(self._impl))
obj2dat(self.desired_capabilities,
- pn_connection_desired_capabilities(self._conn))
- obj2dat(self.properties, pn_connection_properties(self._conn))
- pn_connection_open(self._conn)
+ pn_connection_desired_capabilities(self._impl))
+ obj2dat(self.properties, pn_connection_properties(self._impl))
+ pn_connection_open(self._impl)
def close(self):
self._update_cond()
- pn_connection_close(self._conn)
+ pn_connection_close(self._impl)
@property
def state(self):
- return pn_connection_state(self._conn)
+ return pn_connection_state(self._impl)
+
+ def _pn_session(self):
+ return pn_session(self._impl)
def session(self):
- return Session._wrap_session(pn_session(self._conn))
+ return Session(self._pn_session)
def session_head(self, mask):
- return Session._wrap_session(pn_session_head(self._conn, mask))
+ return Session.wrap(pn_session_head(self._impl, mask))
def link_head(self, mask):
- return Link._wrap_link(pn_link_head(self._conn, mask))
+ return Link.wrap(pn_link_head(self._impl, mask))
@property
def work_head(self):
- return Delivery._wrap_delivery(pn_work_head(self._conn))
+ return Delivery.wrap(pn_work_head(self._impl))
@property
def error(self):
- return pn_error_code(pn_connection_error(self._conn))
+ return pn_error_code(pn_connection_error(self._impl))
+
+ def free(self):
+ pn_connection_release(self._impl)
class SessionException(ProtonException):
pass
-class Session(Endpoint):
+class Session(Wrapper, 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)
- del self._ssn
-
- 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.
+ def wrap(impl):
+ if impl is None:
+ return None
+ else:
+ return Session(impl)
- """
- self.connection._sessions.remove(self)
- self._release()
+ def __init__(self, impl):
+ Wrapper.__init__(self, impl, pn_session_attachments)
def _get_cond_impl(self):
- return pn_session_condition(self._ssn)
+ return pn_session_condition(self._impl)
def _get_remote_cond_impl(self):
- return pn_session_remote_condition(self._ssn)
+ return pn_session_remote_condition(self._impl)
def _get_incoming_capacity(self):
- return pn_session_get_incoming_capacity(self._ssn)
+ return pn_session_get_incoming_capacity(self._impl)
def _set_incoming_capacity(self, capacity):
- pn_session_set_incoming_capacity(self._ssn, capacity)
+ pn_session_set_incoming_capacity(self._impl, capacity)
incoming_capacity = property(_get_incoming_capacity, _set_incoming_capacity)
@property
def outgoing_bytes(self):
- return pn_session_outgoing_bytes(self._ssn)
+ return pn_session_outgoing_bytes(self._impl)
@property
def incoming_bytes(self):
- return pn_session_incoming_bytes(self._ssn)
+ return pn_session_incoming_bytes(self._impl)
def open(self):
- pn_session_open(self._ssn)
+ pn_session_open(self._impl)
def close(self):
self._update_cond()
- pn_session_close(self._ssn)
+ pn_session_close(self._impl)
def next(self, mask):
- return Session._wrap_session(pn_session_next(self._ssn, mask))
+ return Session.wrap(pn_session_next(self._impl, mask))
@property
def state(self):
- return pn_session_state(self._ssn)
+ return pn_session_state(self._impl)
@property
def connection(self):
- return Connection._wrap_connection(pn_session_connection(self._ssn))
+ return Connection.wrap(pn_session_connection(self._impl))
def sender(self, name):
- return Link._wrap_link(pn_sender(self._ssn, name))
+ return Sender(lambda: pn_sender(self._impl, name))
def receiver(self, name):
- return Link._wrap_link(pn_receiver(self._ssn, name))
+ return Receiver(lambda: pn_receiver(self._impl, name))
+
+ def free(self):
+ pn_session_release(self._impl)
class LinkException(ProtonException):
pass
-class Link(Endpoint):
+class Link(Wrapper, Endpoint):
SND_UNSETTLED = PN_SND_UNSETTLED
SND_SETTLED = PN_SND_SETTLED
@@ -2505,152 +2452,132 @@ class Link(Endpoint):
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)
+ def wrap(impl):
+ if impl is None: return None
+ if pn_link_is_sender(impl):
+ return Sender(impl)
else:
- wrapper = Receiver(c_link)
- return wrapper
+ return Receiver(impl)
- 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):
- if self._link: pn_link_free(self._link)
- del self._link
-
- def free(self):
- """Release the Link, freeing its resources"""
- self.session._links.remove(self)
- self._release()
+ def __init__(self, impl):
+ Wrapper.__init__(self, impl, pn_link_attachments)
def _check(self, err):
if err < 0:
exc = EXCEPTIONS.get(err, LinkException)
- raise exc("[%s]: %s" % (err, pn_link_error(self._link)))
+ raise exc("[%s]: %s" % (err, pn_link_error(self._impl)))
else:
return err
def _get_cond_impl(self):
- return pn_link_condition(self._link)
+ return pn_link_condition(self._impl)
def _get_remote_cond_impl(self):
- return pn_link_remote_condition(self._link)
+ return pn_link_remote_condition(self._impl)
def open(self):
- pn_link_open(self._link)
+ pn_link_open(self._impl)
def close(self):
self._update_cond()
- pn_link_close(self._link)
+ pn_link_close(self._impl)
@property
def state(self):
- return pn_link_state(self._link)
+ return pn_link_state(self._impl)
@property
def source(self):
- return Terminus(pn_link_source(self._link))
+ return Terminus(pn_link_source(self._impl))
@property
def target(self):
- return Terminus(pn_link_target(self._link))
+ return Terminus(pn_link_target(self._impl))
@property
def remote_source(self):
- return Terminus(pn_link_remote_source(self._link))
+ return Terminus(pn_link_remote_source(self._impl))
@property
def remote_target(self):
- return Terminus(pn_link_remote_target(self._link))
+ return Terminus(pn_link_remote_target(self._impl))
@property
def session(self):
- return Session._wrap_session(pn_link_session(self._link))
+ return Session.wrap(pn_link_session(self._impl))
@property
def connection(self):
return self.session.connection
def delivery(self, tag):
- return Delivery._wrap_delivery(pn_delivery(self._link, tag))
+ return Delivery(lambda: pn_delivery(self._impl, tag))
@property
def current(self):
- return Delivery._wrap_delivery(pn_link_current(self._link))
+ return Delivery.wrap(pn_link_current(self._impl))
def advance(self):
- return pn_link_advance(self._link)
+ return pn_link_advance(self._impl)
@property
def unsettled(self):
- return pn_link_unsettled(self._link)
+ return pn_link_unsettled(self._impl)
@property
def credit(self):
- return pn_link_credit(self._link)
+ return pn_link_credit(self._impl)
@property
def available(self):
- return pn_link_available(self._link)
+ return pn_link_available(self._impl)
@property
def queued(self):
- return pn_link_queued(self._link)
+ return pn_link_queued(self._impl)
def next(self, mask):
- return Link._wrap_link(pn_link_next(self._link, mask))
+ return Link.wrap(pn_link_next(self._impl, mask))
@property
def name(self):
- return pn_link_name(self._link)
+ return pn_link_name(self._impl)
@property
def is_sender(self):
- return pn_link_is_sender(self._link)
+ return pn_link_is_sender(self._impl)
@property
def is_receiver(self):
- return pn_link_is_receiver(self._link)
+ return pn_link_is_receiver(self._impl)
@property
def remote_snd_settle_mode(self):
- return pn_link_remote_snd_settle_mode(self._link)
+ return pn_link_remote_snd_settle_mode(self._impl)
@property
def remote_rcv_settle_mode(self):
- return pn_link_remote_rcv_settle_mode(self._link)
+ return pn_link_remote_rcv_settle_mode(self._impl)
def _get_snd_settle_mode(self):
- return pn_link_snd_settle_mode(self._link)
+ return pn_link_snd_settle_mode(self._impl)
def _set_snd_settle_mode(self, mode):
- pn_link_set_snd_settle_mode(self._link, 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._link)
+ return pn_link_rcv_settle_mode(self._impl)
def _set_rcv_settle_mode(self, mode):
- pn_link_set_rcv_settle_mode(self._link, mode)
+ pn_link_set_rcv_settle_mode(self._impl, mode)
rcv_settle_mode = property(_get_rcv_settle_mode, _set_rcv_settle_mode)
def drained(self):
- return pn_link_drained(self._link)
+ return pn_link_drained(self._impl)
def detach(self):
- return pn_link_detach(self._link)
+ return pn_link_detach(self._impl)
+
+ def free(self):
+ pn_link_release(self._impl)
class Terminus(object):
@@ -2740,25 +2667,19 @@ class Terminus(object):
class Sender(Link):
- def __init__(self, c_link):
- super(Sender, self).__init__(c_link)
-
def offered(self, n):
- pn_link_offered(self._link, n)
+ pn_link_offered(self._impl, n)
def send(self, bytes):
- return self._check(pn_link_send(self._link, bytes))
+ return self._check(pn_link_send(self._impl, 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)
+ pn_link_flow(self._impl, n)
def recv(self, limit):
- n, bytes = pn_link_recv(self._link, limit)
+ n, bytes = pn_link_recv(self._impl, limit)
if n == PN_EOS:
return None
else:
@@ -2766,10 +2687,10 @@ class Receiver(Link):
return bytes
def drain(self, n):
- pn_link_drain(self._link, n)
+ pn_link_drain(self._impl, n)
def draining(self):
- return pn_link_draining(self._link)
+ return pn_link_draining(self._impl)
class NamedInt(int):
@@ -2875,7 +2796,7 @@ class Disposition(object):
raise AttributeError("condition attribute is read-only")
condition = property(_get_condition, _set_condition)
-class Delivery(object):
+class Delivery(Wrapper):
RECEIVED = Disposition.RECEIVED
ACCEPTED = Disposition.ACCEPTED
@@ -2884,113 +2805,87 @@ class Delivery(object):
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 wrap(impl):
+ if impl is None:
+ return None
+ else:
+ return Delivery(impl)
- 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
+ def __init__(self, impl):
+ Wrapper.__init__(self, impl, pn_delivery_attachments)
- @property
- def released(self):
- return self._dlv is None
+ def _init(self):
+ self.local = Disposition(pn_delivery_local(self._impl), True)
+ self.remote = Disposition(pn_delivery_remote(self._impl), False)
@property
def tag(self):
- return pn_delivery_tag(self._dlv)
+ return pn_delivery_tag(self._impl)
@property
def writable(self):
- return pn_delivery_writable(self._dlv)
+ return pn_delivery_writable(self._impl)
@property
def readable(self):
- return pn_delivery_readable(self._dlv)
+ return pn_delivery_readable(self._impl)
@property
def updated(self):
- return pn_delivery_updated(self._dlv)
+ return pn_delivery_updated(self._impl)
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)
+ pn_delivery_update(self._impl, state)
@property
def pending(self):
- return pn_delivery_pending(self._dlv)
+ return pn_delivery_pending(self._impl)
@property
def partial(self):
- return pn_delivery_partial(self._dlv)
+ return pn_delivery_partial(self._impl)
@property
def local_state(self):
- return DispositionType.get(pn_delivery_local_state(self._dlv))
+ return DispositionType.get(pn_delivery_local_state(self._impl))
@property
def remote_state(self):
- return DispositionType.get(pn_delivery_remote_state(self._dlv))
+ return DispositionType.get(pn_delivery_remote_state(self._impl))
@property
def settled(self):
- return pn_delivery_settled(self._dlv)
+ return pn_delivery_settled(self._impl)
def settle(self):
- """Release the delivery"""
- self.link._deliveries.remove(self)
- self._release()
+ # XXX: we have to incref here because settle is overloaded to
+ # decref also, remove this when C semantics are fixed
+ pn_incref(self._impl)
+ pn_delivery_settle(self._impl)
@property
def work_next(self):
- return Delivery._wrap_delivery(pn_work_next(self._dlv))
+ return Delivery.wrap(pn_work_next(self._impl))
@property
def link(self):
- if not self.released:
- return Link._wrap_link(pn_delivery_link(self._dlv))
- else:
- return None
+ return Link.wrap(pn_delivery_link(self._impl))
@property
def session(self):
- if self.link:
- return self.link.session
- else:
- return None
+ return self.link.session
@property
def connection(self):
- if self.session:
- return self.session.connection
- else:
- return None
+ return self.session.connection
class TransportException(ProtonException):
pass
-class Transport(object):
+class Transport(Wrapper):
TRACE_OFF = PN_TRACE_OFF
TRACE_DRV = PN_TRACE_DRV
@@ -3001,93 +2896,73 @@ class Transport(object):
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)
+ def wrap(impl):
+ if impl is None:
+ return None
+ else:
+ return Transport(_impl=impl)
+
+ def __init__(self, mode=None, _impl = pn_transport):
+ Wrapper.__init__(self, _impl, pn_transport_attachments)
+ if mode == Transport.SERVER:
+ pn_transport_set_server(self._impl)
+ elif mode is None or mode==Transport.CLIENT:
+ pass
else:
raise TransportException("Cannot initialise Transport from mode: %s" % str(mode))
+
+ def _init(self):
self._sasl = None
self._ssl = None
- def __del__(self):
- if hasattr(self, "_trans"):
- if not hasattr(self, "_shared_trans"):
- pn_transport_free(self._trans)
- del 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
-
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))))
+ raise exc("[%s]: %s" % (err, pn_error_text(pn_transport_error(self._impl))))
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
+ self._check(pn_transport_bind(self._impl, connection._impl))
def unbind(self):
"""Release the connection"""
- self._check(pn_transport_unbind(self._trans))
- self._connection = None
+ self._check(pn_transport_unbind(self._impl))
def trace(self, n):
- pn_transport_trace(self._trans, n)
+ pn_transport_trace(self._impl, 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)))
+ return millis2secs(pn_transport_tick(self._impl, secs2millis(now)))
def capacity(self):
- c = pn_transport_capacity(self._trans)
+ c = pn_transport_capacity(self._impl)
if c >= PN_EOS:
return c
else:
return self._check(c)
def push(self, bytes):
- n = self._check(pn_transport_push(self._trans, bytes))
+ n = self._check(pn_transport_push(self._impl, bytes))
if n != len(bytes):
raise OverflowError("unable to process all bytes")
def close_tail(self):
- self._check(pn_transport_close_tail(self._trans))
+ self._check(pn_transport_close_tail(self._impl))
def pending(self):
- p = pn_transport_pending(self._trans)
+ p = pn_transport_pending(self._impl)
if p >= PN_EOS:
return p
else:
return self._check(p)
def peek(self, size):
- cd, out = pn_transport_peek(self._trans, size)
+ cd, out = pn_transport_peek(self._impl, size)
if cd == PN_EOS:
return None
else:
@@ -3095,21 +2970,21 @@ class Transport(object):
return out
def pop(self, size):
- pn_transport_pop(self._trans, size)
+ pn_transport_pop(self._impl, size)
def close_head(self):
- self._check(pn_transport_close_head(self._trans))
+ self._check(pn_transport_close_head(self._impl))
@property
def closed(self):
- return pn_transport_closed(self._trans)
+ return pn_transport_closed(self._impl)
# AMQP 1.0 max-frame-size
def _get_max_frame_size(self):
- return pn_transport_get_max_frame(self._trans)
+ return pn_transport_get_max_frame(self._impl)
def _set_max_frame_size(self, value):
- pn_transport_set_max_frame(self._trans, value)
+ pn_transport_set_max_frame(self._impl, value)
max_frame_size = property(_get_max_frame_size, _set_max_frame_size,
doc="""
@@ -3118,13 +2993,13 @@ Sets the maximum size for received frames (in bytes).
@property
def remote_max_frame_size(self):
- return pn_transport_get_remote_max_frame(self._trans)
+ return pn_transport_get_remote_max_frame(self._impl)
def _get_channel_max(self):
- return pn_transport_get_channel_max(self._trans)
+ return pn_transport_get_channel_max(self._impl)
def _set_channel_max(self, value):
- pn_transport_set_channel_max(self._trans, value)
+ pn_transport_set_channel_max(self._impl, value)
channel_max = property(_get_channel_max, _set_channel_max,
doc="""
@@ -3133,14 +3008,14 @@ Sets the maximum channel that may be used on the transport.
@property
def remote_channel_max(self):
- return pn_transport_remote_channel_max(self._trans)
+ return pn_transport_remote_channel_max(self._impl)
# AMQP 1.0 idle-time-out
def _get_idle_timeout(self):
- return millis2secs(pn_transport_get_idle_timeout(self._trans))
+ return millis2secs(pn_transport_get_idle_timeout(self._impl))
def _set_idle_timeout(self, sec):
- pn_transport_set_idle_timeout(self._trans, secs2millis(sec))
+ pn_transport_set_idle_timeout(self._impl, secs2millis(sec))
idle_timeout = property(_get_idle_timeout, _set_idle_timeout,
doc="""
@@ -3149,21 +3024,18 @@ The idle timeout of the connection (float, in seconds).
@property
def remote_idle_timeout(self):
- return millis2secs(pn_transport_get_remote_idle_timeout(self._trans))
+ return millis2secs(pn_transport_get_remote_idle_timeout(self._impl))
@property
def frames_output(self):
- return pn_transport_get_frames_output(self._trans)
+ return pn_transport_get_frames_output(self._impl)
@property
def frames_input(self):
- return pn_transport_get_frames_input(self._trans)
+ return pn_transport_get_frames_input(self._impl)
def sasl(self):
- # SASL factory (singleton for this transport)
- if not self._sasl:
- self._sasl = SASL(self)
- return self._sasl
+ return SASL(self)
def ssl(self, domain=None, session_details=None):
# SSL factory (singleton for this transport)
@@ -3173,28 +3045,24 @@ The idle timeout of the connection (float, in seconds).
@property
def condition(self):
- return cond2obj(pn_transport_condition(self._trans))
+ return cond2obj(pn_transport_condition(self._impl))
@property
def connection(self):
- return Connection._wrap_connection(pn_transport_connection(self._trans))
+ return Connection.wrap(pn_transport_connection(self._impl))
class SASLException(TransportException):
pass
-class SASL(object):
+class SASL(Wrapper):
OK = PN_SASL_OK
AUTH = PN_SASL_AUTH
SKIPPED = PN_SASL_SKIPPED
- def __new__(cls, transport):
- """Enforce a singleton SASL object per Transport"""
- if not transport._sasl:
- obj = super(SASL, cls).__new__(cls)
- obj._sasl = pn_sasl(transport._trans)
- transport._sasl = obj
- return transport._sasl
+ def __init__(self, transport):
+ Wrapper.__init__(self, transport._impl, pn_transport_attachments)
+ self._sasl = pn_sasl(transport._impl)
def _check(self, err):
if err < 0:
@@ -3328,7 +3196,7 @@ class SSL(object):
session_id = None
if session_details:
session_id = session_details.get_session_id()
- obj._ssl = pn_ssl( transport._trans )
+ obj._ssl = pn_ssl( transport._impl )
if obj._ssl is None:
raise SSLUnavailable()
pn_ssl_init( obj._ssl, domain._domain, session_id )
@@ -3381,18 +3249,17 @@ class SSLSessionDetails(object):
wrappers = {
"pn_void": lambda x: pn_void2py(x),
"pn_pyref": lambda x: pn_void2py(x),
- "pn_connection": lambda x: Connection._wrap_connection(pn_cast_pn_connection(x)),
- "pn_session": lambda x: Session._wrap_session(pn_cast_pn_session(x)),
- "pn_link": lambda x: Link._wrap_link(pn_cast_pn_link(x)),
- "pn_delivery": lambda x: Delivery._wrap_delivery(pn_cast_pn_delivery(x)),
- "pn_transport": lambda x: Transport._wrap_transport(pn_cast_pn_transport(x))
+ "pn_connection": lambda x: Connection.wrap(pn_cast_pn_connection(x)),
+ "pn_session": lambda x: Session.wrap(pn_cast_pn_session(x)),
+ "pn_link": lambda x: Link.wrap(pn_cast_pn_link(x)),
+ "pn_delivery": lambda x: Delivery.wrap(pn_cast_pn_delivery(x)),
+ "pn_transport": lambda x: Transport.wrap(pn_cast_pn_transport(x))
}
class Collector:
def __init__(self):
self._impl = pn_collector()
- self._contexts = set()
def put(self, obj, etype):
pn_collector_put(self._impl, PN_PYREF, pn_py2void(obj), etype.number)
@@ -3411,8 +3278,6 @@ class Collector:
def pop(self):
ev = self.peek()
- if ev is not None:
- ev._popped(self)
pn_collector_pop(self._impl)
def __del__(self):
@@ -3491,10 +3356,8 @@ class Event(EventBase):
def __init__(self, clazz, context, type):
super(Event, self).__init__(clazz, context, type)
- def _popped(self, collector):
- if self.type in (Event.LINK_FINAL, Event.SESSION_FINAL,
- Event.CONNECTION_FINAL):
- collector._contexts.remove(self.context)
+ def dispatch(self, handler):
+ return dispatch(handler, self.type.method, self)
@property
def connection(self):
@@ -3625,7 +3488,7 @@ class Connector(object):
@property
def transport(self):
trans = pn_connector_transport(self._cxtr)
- return Transport._wrap_transport(trans)
+ return Transport.wrap(trans)
def close(self):
return pn_connector_close(self._cxtr)
@@ -3639,7 +3502,7 @@ class Connector(object):
def _set_connection(self, conn):
if conn:
- pn_connector_set_connection(self._cxtr, conn._conn)
+ pn_connector_set_connection(self._cxtr, conn._impl)
else:
pn_connector_set_connection(self._cxtr, None)
self._connection = conn
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-c/bindings/python/proton/wrapper.py
----------------------------------------------------------------------
diff --git a/proton-c/bindings/python/proton/wrapper.py b/proton-c/bindings/python/proton/wrapper.py
new file mode 100644
index 0000000..7f3f421
--- /dev/null
+++ b/proton-c/bindings/python/proton/wrapper.py
@@ -0,0 +1,88 @@
+#
+# 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.
+#
+from cproton import *
+
+class Wrapper(object):
+
+ def __init__(self, impl_or_constructor, get_context):
+ init = False
+ if callable(impl_or_constructor):
+ # we are constructing a new object
+ impl = impl_or_constructor()
+ init = True
+ else:
+ # we are wrapping an existing object
+ impl = impl_or_constructor
+ pn_incref(impl)
+
+ record = get_context(impl)
+ attrs = pn_void2py(pn_record_get(record, PYCTX))
+ if attrs is None:
+ attrs = {}
+ pn_record_def(record, PYCTX, PN_PYREF)
+ pn_record_set(record, PYCTX, pn_py2void(attrs))
+ init = True
+ self.__dict__["_impl"] = impl
+ self.__dict__["_attrs"] = attrs
+ if init: self._init()
+
+ def __getattr__(self, name):
+ attrs = self.__dict__["_attrs"]
+ if name in attrs:
+ return attrs[name]
+ else:
+ raise AttributeError(name)
+
+ def __setattr__(self, name, value):
+ if hasattr(self.__class__, name):
+ object.__setattr__(self, name, value)
+ else:
+ attrs = self.__dict__["_attrs"]
+ attrs[name] = value
+
+ def __delattr__(self, name):
+ attrs = self.__dict__["_attrs"]
+ if attrs:
+ del attrs[name]
+
+ def __hash__(self):
+ return hash(addressof(self._impl))
+
+ def __cmp__(self, other):
+ if isinstance(other, Wrapper):
+ return cmp(addressof(self._impl), addressof(other._impl))
+ else:
+ return -1
+
+ def __del__(self):
+ pn_decref(self._impl)
+
+ def __repr__(self):
+ return '<%s.%s 0x%x ~ 0x%x>' % (self.__class__.__module__,
+ self.__class__.__name__,
+ id(self), addressof(self._impl))
+
+
+if pn_py2void(Wrapper) is Wrapper:
+ PYCTX = Wrapper
+ import java.lang.System
+ addressof = java.lang.System.identityHashCode
+else:
+ PYCTX = int(pn_py2void(Wrapper))
+ addressof = int
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-c/include/proton/connection.h
----------------------------------------------------------------------
diff --git a/proton-c/include/proton/connection.h b/proton-c/include/proton/connection.h
index b7e525c..fcaddf7 100644
--- a/proton-c/include/proton/connection.h
+++ b/proton-c/include/proton/connection.h
@@ -102,6 +102,17 @@ PN_EXTERN pn_connection_t *pn_connection(void);
PN_EXTERN void pn_connection_free(pn_connection_t *connection);
/**
+ * Release a connection object.
+ *
+ * When a connection object is released, all ::pn_session_t and
+ * ::pn_link_t, objects associated with the connection are also
+ * released and all ::pn_delivery_t objects are settled.
+ *
+ * @param[in] connection the connection object to be released
+ */
+PN_EXTERN void pn_connection_release(pn_connection_t *connection);
+
+/**
* Get additional error information associated with the connection.
*
* Whenever a connection operation fails (i.e. returns an error code),
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-c/include/proton/link.h
----------------------------------------------------------------------
diff --git a/proton-c/include/proton/link.h b/proton-c/include/proton/link.h
index 245cce5..a3bdc15 100644
--- a/proton-c/include/proton/link.h
+++ b/proton-c/include/proton/link.h
@@ -81,6 +81,18 @@ PN_EXTERN pn_link_t *pn_receiver(pn_session_t *session, const char *name);
PN_EXTERN void pn_link_free(pn_link_t *link);
/**
+ * Releasing a link object.
+ *
+ * When a link object is freed, it will no longer be retained by its
+ * session once all internal references to the link are no longer
+ * needed. Releasing a link will settle any unsettled deliveries on
+ * the link.
+ *
+ * @param[in] link the link to be released
+ */
+PN_EXTERN void pn_link_release(pn_link_t *link);
+
+/**
* @deprecated
* Get the application context that is associated with a link object.
*
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-c/include/proton/session.h
----------------------------------------------------------------------
diff --git a/proton-c/include/proton/session.h b/proton-c/include/proton/session.h
index 8b6e9fa..4e3bbb5 100644
--- a/proton-c/include/proton/session.h
+++ b/proton-c/include/proton/session.h
@@ -62,6 +62,18 @@ PN_EXTERN pn_session_t *pn_session(pn_connection_t *connection);
PN_EXTERN void pn_session_free(pn_session_t *session);
/**
+ * Release a session object.
+ *
+ * When a session is released it will no longer be retained by the
+ * connection once any internal references to the session are no
+ * longer needed. Releasing a session will release all links on that
+ * session and settle any deliveries on those links.
+ *
+ * @param[in] session a session object to release
+ */
+PN_EXTERN void pn_session_release(pn_session_t *session);
+
+/**
* @deprecated
* Get the application context that is associated with a session
* object.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-c/src/engine/engine-internal.h
----------------------------------------------------------------------
diff --git a/proton-c/src/engine/engine-internal.h b/proton-c/src/engine/engine-internal.h
index 4ee9a69..ab88ad2 100644
--- a/proton-c/src/engine/engine-internal.h
+++ b/proton-c/src/engine/engine-internal.h
@@ -49,10 +49,12 @@ struct pn_endpoint_t {
pn_endpoint_t *endpoint_prev;
pn_endpoint_t *transport_next;
pn_endpoint_t *transport_prev;
+ int refcount; // when this hits zero we generate a final event
bool modified;
bool freed;
+ bool constructed; // track whether the endpoint was explicitly
+ // constructed or not
bool referenced;
- bool posted_final;
};
typedef struct {
@@ -191,6 +193,7 @@ struct pn_connection_t {
pn_endpoint_t *transport_head; // reference counted
pn_endpoint_t *transport_tail;
pn_list_t *sessions;
+ pn_list_t *freed;
pn_transport_t *transport;
pn_delivery_t *work_head;
pn_delivery_t *work_tail;
@@ -210,6 +213,7 @@ struct pn_session_t {
pn_endpoint_t endpoint;
pn_connection_t *connection; // reference counted
pn_list_t *links;
+ pn_list_t *freed;
pn_record_t *context;
size_t incoming_capacity;
pn_sequence_t incoming_bytes;
@@ -291,6 +295,8 @@ struct pn_delivery_t {
bool work;
bool tpwork;
bool done;
+ bool constructed; // track whether the delivery was explicitly
+ // constructed or not
bool referenced;
};
@@ -315,7 +321,11 @@ void pn_clear_modified(pn_connection_t *connection, pn_endpoint_t *endpoint);
void pn_connection_bound(pn_connection_t *conn);
void pn_connection_unbound(pn_connection_t *conn);
int pn_do_error(pn_transport_t *transport, const char *condition, const char *fmt, ...);
+void pn_session_bound(pn_session_t* ssn);
void pn_session_unbound(pn_session_t* ssn);
+void pn_link_bound(pn_link_t* link);
void pn_link_unbound(pn_link_t* link);
+void pn_ep_incref(pn_endpoint_t *endpoint);
+void pn_ep_decref(pn_endpoint_t *endpoint);
#endif /* engine-internal.h */
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-c/src/engine/engine.c
----------------------------------------------------------------------
diff --git a/proton-c/src/engine/engine.c b/proton-c/src/engine/engine.c
index 1f2768a..9bf5cb6 100644
--- a/proton-c/src/engine/engine.c
+++ b/proton-c/src/engine/engine.c
@@ -105,7 +105,10 @@ void pn_connection_close(pn_connection_t *connection)
void pn_endpoint_tini(pn_endpoint_t *endpoint);
-void pn_connection_free(pn_connection_t *connection)
+static void pn_session_free2(pn_session_t *session, bool decref);
+static void pn_link_free2(pn_link_t *link, bool decref);
+
+static void pn_connection_free2(pn_connection_t *connection, bool decref)
{
assert(!connection->endpoint.freed);
// free those endpoints that haven't been freed by the application
@@ -115,11 +118,11 @@ void pn_connection_free(pn_connection_t *connection)
switch (ep->type) {
case SESSION:
// note: this will free all child links:
- pn_session_free((pn_session_t *)ep);
+ pn_session_free2((pn_session_t *)ep, decref);
break;
case SENDER:
case RECEIVER:
- pn_link_free((pn_link_t *)ep);
+ pn_link_free2((pn_link_t *)ep, decref);
break;
default:
assert(false);
@@ -129,14 +132,32 @@ void pn_connection_free(pn_connection_t *connection)
if (!connection->transport) {
// no transport available to consume transport work items,
// so manually clear them:
+ pn_ep_incref(&connection->endpoint);
pn_connection_unbound(connection);
}
- pn_decref(connection);
+ pn_ep_decref(&connection->endpoint);
+ if (decref) {
+ pn_decref(connection);
+ }
+}
+
+void pn_connection_free(pn_connection_t *connection) {
+ pn_connection_free2(connection, true);
+}
+
+void pn_connection_release(pn_connection_t *connection) {
+ pn_connection_free2(connection, false);
}
void pn_connection_bound(pn_connection_t *connection)
{
pn_collector_put(connection->collector, PN_OBJECT, connection, PN_CONNECTION_BOUND);
+ pn_ep_incref(&connection->endpoint);
+
+ size_t nsessions = pn_list_size(connection->sessions);
+ for (size_t i = 0; i < nsessions; i++) {
+ pn_session_bound((pn_session_t *) pn_list_get(connection->sessions, i));
+ }
}
// invoked when transport has been removed:
@@ -154,6 +175,7 @@ void pn_connection_unbound(pn_connection_t *connection)
pn_clear_tpwork(connection->tpwork_head);
}
}
+ pn_ep_decref(&connection->endpoint);
}
pn_record_t *pn_connection_attachments(pn_connection_t *connection)
@@ -200,11 +222,14 @@ void pn_add_session(pn_connection_t *conn, pn_session_t *ssn)
pn_list_add(conn->sessions, ssn);
ssn->connection = conn;
pn_incref(conn); // keep around until finalized
+ pn_ep_incref(&conn->endpoint);
}
void pn_remove_session(pn_connection_t *conn, pn_session_t *ssn)
{
- pn_list_remove(conn->sessions, ssn);
+ if (pn_list_remove(conn->sessions, ssn)) {
+ pn_ep_decref(&conn->endpoint);
+ }
}
pn_connection_t *pn_session_connection(pn_session_t *session)
@@ -225,18 +250,30 @@ void pn_session_close(pn_session_t *session)
pn_endpoint_close(&session->endpoint);
}
-void pn_session_free(pn_session_t *session)
+static void pn_session_free2(pn_session_t *session, bool decref)
{
assert(!session->endpoint.freed);
while(pn_list_size(session->links)) {
pn_link_t *link = (pn_link_t *)pn_list_get(session->links, 0);
- pn_link_free(link);
+ pn_link_free2(link, decref);
}
pn_remove_session(session->connection, session);
+ pn_list_add(session->connection->freed, session);
pn_endpoint_t *endpoint = (pn_endpoint_t *) session;
LL_REMOVE(pn_ep_get_connection(endpoint), endpoint, endpoint);
session->endpoint.freed = true;
- pn_decref(session);
+ pn_ep_decref(&session->endpoint);
+ if (decref && session->endpoint.constructed) {
+ pn_decref(session);
+ }
+}
+
+void pn_session_free(pn_session_t *session) {
+ pn_session_free2(session, true);
+}
+
+void pn_session_release(pn_session_t *session) {
+ pn_session_free2(session, false);
}
pn_record_t *pn_session_attachments(pn_session_t *session)
@@ -261,11 +298,14 @@ void pn_add_link(pn_session_t *ssn, pn_link_t *link)
{
pn_list_add(ssn->links, link);
link->session = ssn;
+ pn_ep_incref(&ssn->endpoint);
}
void pn_remove_link(pn_session_t *ssn, pn_link_t *link)
{
- pn_list_remove(ssn->links, link);
+ if (pn_list_remove(ssn->links, link)) {
+ pn_ep_decref(&ssn->endpoint);
+ }
}
void pn_link_open(pn_link_t *link)
@@ -298,20 +338,35 @@ void pn_terminus_free(pn_terminus_t *terminus)
pn_free(terminus->filter);
}
-void pn_link_free(pn_link_t *link)
+static void pn_link_free2(pn_link_t *link, bool decref)
{
assert(!link->endpoint.freed);
- pn_remove_link(link->session, link);
pn_endpoint_t *endpoint = (pn_endpoint_t *) link;
+ pn_remove_link(link->session, link);
+ pn_list_add(link->session->freed, link);
LL_REMOVE(pn_ep_get_connection(endpoint), endpoint, endpoint);
pn_delivery_t *delivery = link->unsettled_head;
while (delivery) {
pn_delivery_t *next = delivery->unsettled_next;
+ if (!(decref && delivery->constructed)) {
+ pn_incref(delivery);
+ }
pn_delivery_settle(delivery);
delivery = next;
}
link->endpoint.freed = true;
- pn_decref(link);
+ pn_ep_decref(&link->endpoint);
+ if (decref && link->endpoint.constructed) {
+ pn_decref(link);
+ }
+}
+
+void pn_link_free(pn_link_t *link) {
+ pn_link_free2(link, true);
+}
+
+void pn_link_release(pn_link_t *link) {
+ pn_link_free2(link, false);
}
void *pn_link_get_context(pn_link_t *link)
@@ -346,32 +401,66 @@ void pn_endpoint_init(pn_endpoint_t *endpoint, int type, pn_connection_t *conn)
endpoint->transport_prev = NULL;
endpoint->modified = false;
endpoint->freed = false;
- endpoint->posted_final = false;
+ endpoint->constructed = true;
+ endpoint->refcount = 1;
+ //fprintf(stderr, "initting 0x%lx\n", (uintptr_t) endpoint);
LL_ADD(conn, endpoint, endpoint);
}
-void pn_endpoint_tini(pn_endpoint_t *endpoint)
+void pn_ep_incref(pn_endpoint_t *endpoint)
{
- pn_error_free(endpoint->error);
- pn_condition_tini(&endpoint->remote_condition);
- pn_condition_tini(&endpoint->condition);
+ endpoint->refcount++;
+}
+
+static pn_event_type_t pn_final_type(pn_endpoint_type_t type) {
+ switch (type) {
+ case CONNECTION:
+ return PN_CONNECTION_FINAL;
+ case SESSION:
+ return PN_SESSION_FINAL;
+ case SENDER:
+ case RECEIVER:
+ return PN_LINK_FINAL;
+ default:
+ assert(false);
+ return PN_EVENT_NONE;
+ }
+}
+
+static pn_endpoint_t *pn_ep_parent(pn_endpoint_t *endpoint) {
+ switch (endpoint->type) {
+ case CONNECTION:
+ return NULL;
+ case SESSION:
+ return &((pn_session_t *) endpoint)->connection->endpoint;
+ case SENDER:
+ case RECEIVER:
+ return &((pn_link_t *) endpoint)->session->endpoint;
+ default:
+ assert(false);
+ return NULL;
+ }
}
-static bool pni_post_final(pn_endpoint_t *endpoint, pn_event_type_t type)
+void pn_ep_decref(pn_endpoint_t *endpoint)
{
- pn_connection_t *conn = pn_ep_get_connection(endpoint);
- if (!endpoint->posted_final) {
- endpoint->posted_final = true;
- pn_event_t *event = pn_collector_put(conn->collector, PN_OBJECT, endpoint,
- type);
- if (event) { return true; }
+ assert(endpoint->refcount > 0);
+ endpoint->refcount--;
+ if (endpoint->refcount == 0) {
+ pn_connection_t *conn = pn_ep_get_connection(endpoint);
+ pn_collector_put(conn->collector, PN_OBJECT, endpoint, pn_final_type(endpoint->type));
}
+}
- return false;
+void pn_endpoint_tini(pn_endpoint_t *endpoint)
+{
+ pn_error_free(endpoint->error);
+ pn_condition_tini(&endpoint->remote_condition);
+ pn_condition_tini(&endpoint->condition);
}
-static void pni_free_children(pn_list_t *children)
+static void pni_free_children(pn_list_t *children, pn_list_t *freed)
{
while (pn_list_size(children) > 0) {
pn_endpoint_t *endpoint = (pn_endpoint_t *) pn_list_get(children, 0);
@@ -379,19 +468,22 @@ static void pni_free_children(pn_list_t *children)
pn_free(endpoint);
}
+ while (pn_list_size(freed) > 0) {
+ pn_endpoint_t *endpoint = (pn_endpoint_t *) pn_list_get(freed, 0);
+ assert(!endpoint->referenced);
+ pn_free(endpoint);
+ }
+
pn_free(children);
+ pn_free(freed);
}
static void pn_connection_finalize(void *object)
{
pn_connection_t *conn = (pn_connection_t *) object;
-
pn_endpoint_t *endpoint = &conn->endpoint;
- if (pni_post_final(endpoint, PN_CONNECTION_FINAL)) {
- return;
- }
- pni_free_children(conn->sessions);
+ pni_free_children(conn->sessions, conn->freed);
pn_free(conn->context);
pn_decref(conn->collector);
@@ -421,6 +513,7 @@ pn_connection_t *pn_connection(void)
conn->transport_head = NULL;
conn->transport_tail = NULL;
conn->sessions = pn_list(PN_WEAKREF, 0);
+ conn->freed = pn_list(PN_WEAKREF, 0);
conn->transport = NULL;
conn->work_head = NULL;
conn->work_tail = NULL;
@@ -734,10 +827,64 @@ static void pn_session_incref(void *object)
}
}
-static bool pni_preserve_child(pn_endpoint_t *endpoint, pn_endpoint_t *parent)
+static bool pn_ep_bound(pn_endpoint_t *endpoint)
+{
+ pn_connection_t *conn = pn_ep_get_connection(endpoint);
+ pn_session_t *ssn;
+ pn_link_t *lnk;
+
+ if (!conn->transport) return false;
+ if (endpoint->modified) return true;
+
+ switch (endpoint->type) {
+ case CONNECTION:
+ return ((pn_connection_t *)endpoint)->transport;
+ case SESSION:
+ ssn = (pn_session_t *) endpoint;
+ return (((int16_t) ssn->state.local_channel) >= 0 || ((int16_t) ssn->state.remote_channel) >= 0);
+ case SENDER:
+ case RECEIVER:
+ lnk = (pn_link_t *) endpoint;
+ return ((int32_t) lnk->state.local_handle) >= 0 || ((int32_t) lnk->state.remote_handle) >= 0;
+ default:
+ assert(false);
+ return false;
+ }
+}
+
+static bool pni_connection_live(pn_connection_t *conn) {
+ return pn_refcount(conn) > 1;
+}
+
+static bool pni_session_live(pn_session_t *ssn) {
+ return pni_connection_live(ssn->connection) || pn_refcount(ssn) > 1;
+}
+
+static bool pni_link_live(pn_link_t *link) {
+ return pni_session_live(link->session) || pn_refcount(link) > 1;
+}
+
+static bool pni_endpoint_live(pn_endpoint_t *endpoint) {
+ switch (endpoint->type) {
+ case CONNECTION:
+ return pni_connection_live((pn_connection_t *)endpoint);
+ case SESSION:
+ return pni_session_live((pn_session_t *) endpoint);
+ case SENDER:
+ case RECEIVER:
+ return pni_link_live((pn_link_t *) endpoint);
+ default:
+ assert(false);
+ return false;
+ }
+}
+
+static bool pni_preserve_child(pn_endpoint_t *endpoint)
{
pn_connection_t *conn = pn_ep_get_connection(endpoint);
- if ((!endpoint->freed || (conn->transport && endpoint->modified)) && endpoint->referenced) {
+ pn_endpoint_t *parent = pn_ep_parent(endpoint);
+ if (pni_endpoint_live(parent) && (!endpoint->freed || (pn_ep_bound(endpoint)))
+ && endpoint->referenced) {
pn_object_incref(endpoint);
endpoint->referenced = false;
pn_decref(parent);
@@ -748,40 +895,24 @@ static bool pni_preserve_child(pn_endpoint_t *endpoint, pn_endpoint_t *parent)
}
}
-static bool pni_connection_live(pn_connection_t *conn)
-{
- return pn_refcount(conn) > 1;
-}
-
static void pn_session_finalize(void *object)
{
pn_session_t *session = (pn_session_t *) object;
pn_endpoint_t *endpoint = &session->endpoint;
- //pn_transport_t *transport = session->connection->transport;
- //if (transport) {
- /* if ((int16_t)session->state.local_channel >= 0) // END not sent */
- /* pn_hash_del(transport->local_channels, session->state.local_channel); */
- /* if ((int16_t)session->state.remote_channel >= 0) // END not received */
- /* pn_unmap_channel(transport, session); */
- /* } */
-
- if (pni_post_final(endpoint, PN_SESSION_FINAL)) {
- return;
- }
- if (pni_connection_live(session->connection) &&
- pni_preserve_child(endpoint, &session->connection->endpoint)) {
+ if (pni_preserve_child(endpoint)) {
return;
}
pn_free(session->context);
- pni_free_children(session->links);
+ pni_free_children(session->links, session->freed);
pn_endpoint_tini(endpoint);
pn_delivery_map_free(&session->state.incoming);
pn_delivery_map_free(&session->state.outgoing);
pn_free(session->state.local_handles);
pn_free(session->state.remote_handles);
pn_remove_session(session->connection, session);
+ pn_list_remove(session->connection->freed, session);
if (endpoint->referenced) {
pn_decref(session->connection);
}
@@ -808,6 +939,7 @@ pn_session_t *pn_session(pn_connection_t *conn)
pn_endpoint_init(&ssn->endpoint, SESSION, conn);
pn_add_session(conn, ssn);
ssn->links = pn_list(PN_WEAKREF, 0);
+ ssn->freed = pn_list(PN_WEAKREF, 0);
ssn->context = pn_record();
ssn->incoming_capacity = 1024*1024;
ssn->incoming_bytes = 0;
@@ -826,9 +958,21 @@ pn_session_t *pn_session(pn_connection_t *conn)
// end transport state
pn_collector_put(conn->collector, PN_OBJECT, ssn, PN_SESSION_INIT);
+ if (conn->transport) {
+ pn_session_bound(ssn);
+ }
return ssn;
}
+void pn_session_bound(pn_session_t *ssn)
+{
+ assert(ssn);
+ size_t nlinks = pn_list_size(ssn->links);
+ for (size_t i = 0; i < nlinks; i++) {
+ pn_link_bound((pn_link_t *) pn_list_get(ssn->links, i));
+ }
+}
+
void pn_session_unbound(pn_session_t* ssn)
{
assert(ssn);
@@ -901,22 +1045,12 @@ static void pn_link_incref(void *object)
}
}
-static bool pni_session_live(pn_session_t *ssn)
-{
- return pni_connection_live(ssn->connection) || pn_refcount(ssn) > 1;
-}
-
static void pn_link_finalize(void *object)
{
pn_link_t *link = (pn_link_t *) object;
pn_endpoint_t *endpoint = &link->endpoint;
- if (pni_post_final(endpoint, PN_LINK_FINAL)) {
- return;
- }
-
- if (pni_session_live(link->session) &&
- pni_preserve_child(endpoint, &link->session->endpoint)) {
+ if (pni_preserve_child(endpoint)) {
return;
}
@@ -933,6 +1067,7 @@ static void pn_link_finalize(void *object)
pn_free(link->name);
pn_endpoint_tini(endpoint);
pn_remove_link(link->session, link);
+ pn_list_remove(link->session->freed, link);
if (endpoint->referenced) {
pn_decref(link->session);
}
@@ -986,9 +1121,17 @@ pn_link_t *pn_link_new(int type, pn_session_t *session, const char *name)
// end transport state
pn_collector_put(session->connection->collector, PN_OBJECT, link, PN_LINK_INIT);
+ if (session->connection->transport) {
+ pn_link_bound(link);
+ }
return link;
}
+void pn_link_bound(pn_link_t *link)
+{
+
+}
+
void pn_link_unbound(pn_link_t* link)
{
assert(link);
@@ -1196,11 +1339,6 @@ static void pn_disposition_finalize(pn_disposition_t *ds)
pn_condition_tini(&ds->condition);
}
-static bool pni_link_live(pn_link_t *link)
-{
- return pni_session_live(link->session) || pn_refcount(link) > 1;
-}
-
static void pn_delivery_incref(void *object)
{
pn_delivery_t *delivery = (pn_delivery_t *) object;
@@ -1215,13 +1353,14 @@ static void pn_delivery_incref(void *object)
static bool pni_preserve_delivery(pn_delivery_t *delivery)
{
pn_connection_t *conn = delivery->link->session->connection;
- return !delivery->local.settled || (conn->transport && delivery->tpwork);
+ return !delivery->local.settled || (conn->transport && (delivery->state.init || delivery->tpwork));
}
static void pn_delivery_finalize(void *object)
{
pn_delivery_t *delivery = (pn_delivery_t *) object;
pn_link_t *link = delivery->link;
+ // assert(!delivery->state.init);
bool pooled = false;
bool referenced = true;
@@ -1313,7 +1452,7 @@ pn_delivery_t *pn_delivery(pn_link_t *link, pn_delivery_tag_t tag)
pn_disposition_init(&delivery->remote);
delivery->context = pn_record();
} else {
- assert(!delivery->tpwork);
+ assert(!delivery->state.init);
}
delivery->link = link;
pn_incref(delivery->link); // keep link until finalized
@@ -1334,6 +1473,7 @@ pn_delivery_t *pn_delivery(pn_link_t *link, pn_delivery_tag_t tag)
pn_buffer_clear(delivery->bytes);
delivery->done = false;
pn_record_clear(delivery->context);
+ delivery->constructed = true;
// begin delivery state
delivery->state.init = false;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-c/src/transport/transport.c
----------------------------------------------------------------------
diff --git a/proton-c/src/transport/transport.c b/proton-c/src/transport/transport.c
index 8b41984..2d379e0 100644
--- a/proton-c/src/transport/transport.c
+++ b/proton-c/src/transport/transport.c
@@ -376,6 +376,7 @@ static void pni_map_remote_channel(pn_session_t *session, uint16_t channel)
pn_transport_t *transport = session->connection->transport;
pn_hash_put(transport->remote_channels, channel, session);
session->state.remote_channel = channel;
+ pn_ep_incref(&session->endpoint);
}
void pni_transport_unbind_handles(pn_hash_t *handles, bool reset_state);
@@ -388,6 +389,9 @@ static void pni_unmap_remote_channel(pn_session_t *ssn)
pn_transport_t *transport = ssn->connection->transport;
uint16_t channel = ssn->state.remote_channel;
ssn->state.remote_channel = -2;
+ if (pn_hash_get(transport->remote_channels, channel)) {
+ pn_ep_decref(&ssn->endpoint);
+ }
// note: may free the session:
pn_hash_del(transport->remote_channels, channel);
}
@@ -429,9 +433,6 @@ void pn_transport_free(pn_transport_t *transport)
if (!transport) return;
assert(!transport->freed);
transport->freed = true;
- // once the application frees the transport, no further I/O
- // processing can be done to the connection:
- pn_transport_unbind(transport);
pn_decref(transport);
}
@@ -439,6 +440,12 @@ static void pn_transport_finalize(void *object)
{
pn_transport_t *transport = (pn_transport_t *) object;
+ // once the application frees the transport, no further I/O
+ // processing can be done to the connection:
+ pn_transport_unbind(transport);
+ // we may have posted events, so stay alive until they are processed
+ if (pn_refcount(transport) > 0) return;
+
pn_free(transport->context);
pn_ssl_free(transport);
pn_sasl_free(transport);
@@ -488,10 +495,11 @@ void pni_transport_unbind_handles(pn_hash_t *handles, bool reset_state)
{
for (pn_handle_t h = pn_hash_head(handles); h; h = pn_hash_next(handles, h)) {
uintptr_t key = pn_hash_key(handles, h);
+ pn_link_t *link = (pn_link_t *) pn_hash_value(handles, h);
if (reset_state) {
- pn_link_t *link = (pn_link_t *) pn_hash_value(handles, h);
pn_link_unbound(link);
}
+ pn_ep_decref(&link->endpoint);
pn_hash_del(handles, key);
}
}
@@ -506,6 +514,7 @@ void pni_transport_unbind_channels(pn_hash_t *channels)
pni_transport_unbind_handles(ssn->state.local_handles, true);
pni_transport_unbind_handles(ssn->state.remote_handles, true);
pn_session_unbound(ssn);
+ pn_ep_decref(&ssn->endpoint);
pn_hash_del(channels, key);
}
}
@@ -567,12 +576,16 @@ static void pni_map_remote_handle(pn_link_t *link, uint32_t handle)
{
link->state.remote_handle = handle;
pn_hash_put(link->session->state.remote_handles, handle, link);
+ pn_ep_incref(&link->endpoint);
}
static void pni_unmap_remote_handle(pn_link_t *link)
{
uintptr_t handle = link->state.remote_handle;
link->state.remote_handle = -2;
+ if (pn_hash_get(link->session->state.remote_handles, handle)) {
+ pn_ep_decref(&link->endpoint);
+ }
// may delete link:
pn_hash_del(link->session->state.remote_handles, handle);
}
@@ -763,6 +776,8 @@ int pn_do_begin(pn_transport_t *transport, uint8_t frame_type, uint16_t channel,
ssn = (pn_session_t *) pn_hash_get(transport->local_channels, remote_channel);
} else {
ssn = pn_session(transport->connection);
+ ssn->endpoint.constructed = false;
+ pn_decref(ssn);
}
ssn->state.incoming_transfer_count = next;
pni_map_remote_channel(ssn, channel);
@@ -876,6 +891,8 @@ int pn_do_attach(pn_transport_t *transport, uint8_t frame_type, uint16_t channel
} else {
link = (pn_link_t *) pn_receiver(ssn, strname);
}
+ link->endpoint.constructed = false;
+ pn_decref(link);
}
if (strheap) {
@@ -993,6 +1010,8 @@ int pn_do_transfer(pn_transport_t *transport, uint8_t frame_type, uint16_t chann
}
delivery = pn_delivery(link, pn_dtag(tag.start, tag.size));
+ delivery->constructed = false;
+ pn_decref(delivery);
pn_delivery_state_t *state = pn_delivery_map_push(incoming, delivery);
if (id_present && id != state->id) {
return pn_do_error(transport, "amqp:session:invalid-field",
@@ -1454,6 +1473,7 @@ static void pni_map_local_channel(pn_session_t *ssn)
uint16_t channel = allocate_alias(transport->local_channels);
state->local_channel = channel;
pn_hash_put(transport->local_channels, channel, ssn);
+ pn_ep_incref(&ssn->endpoint);
}
int pn_process_ssn_setup(pn_transport_t *transport, pn_endpoint_t *endpoint)
@@ -1499,6 +1519,7 @@ static void pni_map_local_handle(pn_link_t *link) {
pn_session_state_t *ssn_state = &link->session->state;
state->local_handle = allocate_alias(ssn_state->local_handles);
pn_hash_put(ssn_state->local_handles, state->local_handle, link);
+ pn_ep_incref(&link->endpoint);
}
int pn_process_link_setup(pn_transport_t *transport, pn_endpoint_t *endpoint)
@@ -1833,6 +1854,9 @@ static void pni_unmap_local_handle(pn_link_t *link) {
pn_link_state_t *state = &link->state;
uintptr_t handle = state->local_handle;
state->local_handle = -2;
+ if (pn_hash_get(link->session->state.local_handles, handle)) {
+ pn_ep_decref(&link->endpoint);
+ }
// may delete link
pn_hash_del(link->session->state.local_handles, handle);
}
@@ -1907,6 +1931,9 @@ static void pni_unmap_local_channel(pn_session_t *ssn) {
pn_session_state_t *state = &ssn->state;
uintptr_t channel = state->local_channel;
state->local_channel = -2;
+ if (pn_hash_get(transport->local_channels, channel)) {
+ pn_ep_decref(&ssn->endpoint);
+ }
// may delete session
pn_hash_del(transport->local_channels, channel);
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java
index 5fe66d5..e025356 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java
@@ -114,4 +114,6 @@ public interface Connection extends Endpoint
void collect(Collector collector);
+ Transport getTransport();
+
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java
index 9a27c1a..eb68f42 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java
@@ -459,7 +459,7 @@ public class ConnectionImpl extends EndpointImpl implements ProtonJConnection
_transport = transport;
}
- TransportImpl getTransport()
+ public TransportImpl getTransport()
{
return _transport;
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
index 7b54b36..566b84a 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
@@ -164,7 +164,7 @@ public class DeliveryImpl implements Delivery
public void free()
{
-
+ settle();
}
DeliveryImpl getLinkPrevious()
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java
index 2190ced..9966526 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java
@@ -116,6 +116,12 @@ public abstract class LinkImpl extends EndpointImpl implements Link
@Override
void doFree()
{
+ DeliveryImpl dlv = _head;
+ while (dlv != null) {
+ dlv.free();
+ dlv = dlv.next();
+ }
+
_session.getConnectionImpl().removeLinkEndpoint(_node);
_node = null;
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
index 2e8a6e7..9f215ee 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
@@ -242,12 +242,6 @@ public class TransportImpl extends EndpointImpl
@Override
public void unbind()
{
- put(Event.Type.CONNECTION_UNBOUND, _connectionEndpoint);
- _connectionEndpoint.modifyEndpoints();
-
- _connectionEndpoint.setTransport(null);
- _connectionEndpoint.decref();
-
for (TransportSession ts: _transportSessionState.values()) {
ts.unbind();
}
@@ -255,6 +249,12 @@ public class TransportImpl extends EndpointImpl
for (TransportLink tl: _transportLinkState.values()) {
tl.unbind();
}
+
+ put(Event.Type.CONNECTION_UNBOUND, _connectionEndpoint);
+
+ _connectionEndpoint.modifyEndpoints();
+ _connectionEndpoint.setTransport(null);
+ _connectionEndpoint.decref();
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-j/src/main/resources/cengine.py
----------------------------------------------------------------------
diff --git a/proton-j/src/main/resources/cengine.py b/proton-j/src/main/resources/cengine.py
index 8f25a0e..c0f6392 100644
--- a/proton-j/src/main/resources/cengine.py
+++ b/proton-j/src/main/resources/cengine.py
@@ -143,7 +143,7 @@ class endpoint_wrapper:
def __init__(self, impl):
self.impl = impl
- self.context = None
+ self.attachments = {}
self.condition = pn_condition()
self.remote_condition = pn_condition()
@@ -214,11 +214,8 @@ def pn_connection_desired_capabilities(conn):
def pn_connection_remote_desired_capabilities(conn):
return array2dat(conn.impl.getRemoteDesiredCapabilities(), PN_SYMBOL)
-def pn_connection_get_context(conn):
- return conn.context
-
-def pn_connection_set_context(conn, ctx):
- conn.context = ctx
+def pn_connection_attachments(conn):
+ return conn.attachments
def pn_connection_set_container(conn, name):
conn.impl.setContainer(name)
@@ -251,20 +248,20 @@ def pn_connection_close(conn):
conn.on_close()
conn.impl.close()
-def pn_connection_free(conn):
+def pn_connection_release(conn):
conn.impl.free()
+def pn_connection_transport(conn):
+ return wrap(conn.impl.getTransport(), pn_transport_wrapper)
+
class pn_session_wrapper(endpoint_wrapper):
pass
def pn_session(conn):
return wrap(conn.impl.session(), pn_session_wrapper)
-def pn_session_get_context(ssn):
- return ssn.context
-
-def pn_session_set_context(ssn, ctx):
- ssn.context = ctx
+def pn_session_attachments(ssn):
+ return ssn.attachments
def pn_session_state(ssn):
return endpoint_state(ssn.impl)
@@ -334,7 +331,7 @@ def pn_sender(ssn, name):
def pn_receiver(ssn, name):
return wrap(ssn.impl.receiver(name), pn_link_wrapper)
-def pn_session_free(ssn):
+def pn_session_release(ssn):
ssn.impl.free()
TERMINUS_TYPES_J2P = {
@@ -523,11 +520,8 @@ class pn_link_wrapper(endpoint_wrapper):
self.impl.setSource(self.source.encode())
self.impl.setTarget(self.target.encode())
-def pn_link_get_context(link):
- return link.context
-
-def pn_link_set_context(link, ctx):
- link.context = ctx
+def pn_link_attachments(link):
+ return link.attachments
def pn_link_source(link):
link.source.decode(link.impl.getSource())
@@ -669,7 +663,7 @@ def pn_link_advance(link):
def pn_link_current(link):
return wrap(link.impl.current(), pn_delivery_wrapper)
-def pn_link_free(link):
+def pn_link_release(link):
link.impl.free()
def pn_work_head(conn):
@@ -804,7 +798,7 @@ class pn_delivery_wrapper:
def __init__(self, impl):
self.impl = impl
- self.context = None
+ self.attachments = {}
self.local = pn_disposition()
self.remote = pn_disposition()
@@ -814,11 +808,8 @@ def pn_delivery(link, tag):
def pn_delivery_tag(dlv):
return dlv.impl.getTag().tostring()
-def pn_delivery_get_context(dlv):
- return dlv.context
-
-def pn_delivery_set_context(dlv, ctx):
- dlv.context = ctx
+def pn_delivery_attachments(dlv):
+ return dlv.attachments
def pn_delivery_partial(dlv):
return dlv.impl.isPartial()
@@ -867,12 +858,16 @@ def pn_delivery_settle(dlv):
class pn_transport_wrapper:
def __init__(self, impl):
self.impl = impl
+ self.attachments = {}
self.server = False
self.condition = pn_condition()
def pn_transport():
return wrap(Proton.transport(), pn_transport_wrapper)
+def pn_transport_attachments(trans):
+ return trans.attachments
+
def pn_transport_set_server(trans):
trans.server = True;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/proton-j/src/main/resources/cobject.py
----------------------------------------------------------------------
diff --git a/proton-j/src/main/resources/cobject.py b/proton-j/src/main/resources/cobject.py
index 29cc06f..18ad7e9 100644
--- a/proton-j/src/main/resources/cobject.py
+++ b/proton-j/src/main/resources/cobject.py
@@ -40,3 +40,20 @@ def pn_cast_pn_delivery(obj):
def pn_cast_pn_transport(obj):
return obj
+
+PN_PYREF = None
+
+def pn_record_def(record, key, clazz):
+ pass
+
+def pn_record_get(record, key):
+ return record.get(key)
+
+def pn_record_set(record, key, value):
+ record[key] = value
+
+def pn_incref(obj):
+ pass
+
+def pn_decref(obj):
+ pass
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/tests/python/proton_tests/engine.py
----------------------------------------------------------------------
diff --git a/tests/python/proton_tests/engine.py b/tests/python/proton_tests/engine.py
index 6a07aea..8205454 100644
--- a/tests/python/proton_tests/engine.py
+++ b/tests/python/proton_tests/engine.py
@@ -46,10 +46,8 @@ class Test(common.Test):
c2 = Connection()
t1 = Transport()
t1.bind(c1)
- c1._transport = t1
t2 = Transport()
t2.bind(c2)
- c2._transport = t2
self._wires.append((c1, t1, c2, t2))
mask1 = 0
@@ -69,12 +67,12 @@ class Test(common.Test):
def link(self, name, max_frame=None, idle_timeout=None):
c1, c2 = self.connection()
if max_frame:
- c1._transport.max_frame_size = max_frame[0]
- c2._transport.max_frame_size = max_frame[1]
+ c1.transport.max_frame_size = max_frame[0]
+ c2.transport.max_frame_size = max_frame[1]
if idle_timeout:
# idle_timeout in seconds expressed as float
- c1._transport.idle_timeout = idle_timeout[0]
- c2._transport.idle_timeout = idle_timeout[1]
+ c1.transport.idle_timeout = idle_timeout[0]
+ c2.transport.idle_timeout = idle_timeout[1]
c1.open()
c2.open()
ssn1 = c1.session()
@@ -88,10 +86,6 @@ class Test(common.Test):
return snd, rcv
def cleanup(self):
- # release resources created by this class
- for w in self._wires:
- w[0]._transport = None
- w[2]._transport = None
self._wires = []
def pump(self, buffer_size=OUTPUT_SIZE):
@@ -215,10 +209,10 @@ class ConnectionTest(Test):
assert self.c1.remote_properties == p2, (self.c2.remote_properties, p2)
def test_channel_max(self, value=1234):
- self.c1._transport.channel_max = value
+ self.c1.transport.channel_max = value
self.c1.open()
self.pump()
- assert self.c2._transport.remote_channel_max == value, (self.c2._transport.remote_channel_max, value)
+ assert self.c2.transport.remote_channel_max == value, (self.c2.transport.remote_channel_max, value)
def test_cleanup(self):
self.c1.open()
@@ -226,8 +220,8 @@ class ConnectionTest(Test):
self.pump()
assert self.c1.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE
assert self.c2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE
- t1 = self.c1._transport
- t2 = self.c2._transport
+ t1 = self.c1.transport
+ t2 = self.c2.transport
c2 = self.c2
self.c1.close()
# release all references to C1, except that held by the transport
@@ -897,8 +891,8 @@ class MaxFrameTransferTest(Test):
self.snd.open()
self.rcv.open()
self.pump()
- assert self.rcv.session.connection._transport.max_frame_size == 512
- assert self.snd.session.connection._transport.remote_max_frame_size == 512
+ assert self.rcv.session.connection.transport.max_frame_size == 512
+ assert self.snd.session.connection.transport.remote_max_frame_size == 512
self.rcv.flow(1)
self.snd.delivery("tag")
@@ -926,8 +920,8 @@ class MaxFrameTransferTest(Test):
self.snd.open()
self.rcv.open()
self.pump()
- assert self.rcv.session.connection._transport.max_frame_size == 521
- assert self.snd.session.connection._transport.remote_max_frame_size == 521
+ assert self.rcv.session.connection.transport.max_frame_size == 521
+ assert self.snd.session.connection.transport.remote_max_frame_size == 521
self.rcv.flow(2)
self.snd.delivery("tag")
@@ -1023,10 +1017,10 @@ class IdleTimeoutTest(Test):
self.rcv.open()
self.pump()
# proton advertises 1/2 the configured timeout to the peer:
- assert self.rcv.session.connection._transport.idle_timeout == 2.0
- assert self.rcv.session.connection._transport.remote_idle_timeout == 0.5
- assert self.snd.session.connection._transport.idle_timeout == 1.0
- assert self.snd.session.connection._transport.remote_idle_timeout == 1.0
+ assert self.rcv.session.connection.transport.idle_timeout == 2.0
+ assert self.rcv.session.connection.transport.remote_idle_timeout == 0.5
+ assert self.snd.session.connection.transport.idle_timeout == 1.0
+ assert self.snd.session.connection.transport.remote_idle_timeout == 1.0
def testTimeout(self):
"""
@@ -1041,8 +1035,8 @@ class IdleTimeoutTest(Test):
self.rcv.open()
self.pump()
- t_snd = self.snd.session.connection._transport
- t_rcv = self.rcv.session.connection._transport
+ t_snd = self.snd.session.connection.transport
+ t_rcv = self.rcv.session.connection.transport
assert t_rcv.idle_timeout == 0.0
# proton advertises 1/2 the timeout (see spec)
assert t_rcv.remote_idle_timeout == 0.5
@@ -1346,8 +1340,8 @@ class CreditTest(Test):
assert self.rcv.credit == 0
assert self.rcv.queued == 0
- #self.rcv.session.connection._transport.trace(Transport.TRACE_FRM)
- #self.snd.session.connection._transport.trace(Transport.TRACE_FRM)
+ #self.rcv.session.connection.transport.trace(Transport.TRACE_FRM)
+ #self.snd.session.connection.transport.trace(Transport.TRACE_FRM)
## verify that a sender that has reached the drain state will respond
## promptly to a drain issued by the peer.
@@ -2158,9 +2152,11 @@ class EventTest(CollectorTest):
del ssn2
self.pump()
c1.free()
- c1._transport.unbind()
- self.expect(Event.CONNECTION_UNBOUND, Event.SESSION_FINAL, Event.LINK_FINAL,
- Event.SESSION_FINAL, Event.CONNECTION_FINAL)
+ c1.transport.unbind()
+ self.expect_oneof((Event.SESSION_FINAL, Event.LINK_FINAL, Event.SESSION_FINAL,
+ Event.CONNECTION_UNBOUND, Event.CONNECTION_FINAL),
+ (Event.CONNECTION_UNBOUND, Event.SESSION_FINAL, Event.LINK_FINAL,
+ Event.SESSION_FINAL, Event.CONNECTION_FINAL))
def testConnectionINIT_FINAL(self):
c = Connection()
@@ -2222,7 +2218,7 @@ class EventTest(CollectorTest):
snd.open()
self.pump()
self.expect(Event.LINK_REMOTE_OPEN, Event.DELIVERY)
- rcv.session.connection._transport.unbind()
+ rcv.session.connection.transport.unbind()
rcv.session.connection.free()
self.expect(Event.CONNECTION_UNBOUND, Event.TRANSPORT, Event.LINK_FINAL,
Event.SESSION_FINAL, Event.CONNECTION_FINAL)
@@ -2382,13 +2378,9 @@ class TeardownLeakTest(PeerTest):
self.pump()
if remote:
- self.expect_oneof((Event.TRANSPORT_HEAD_CLOSED, Event.LINK_REMOTE_CLOSE,
- Event.SESSION_REMOTE_CLOSE, Event.CONNECTION_REMOTE_CLOSE,
- Event.TRANSPORT_TAIL_CLOSED, Event.TRANSPORT_CLOSED),
- (Event.TRANSPORT_HEAD_CLOSED, Event.LINK_REMOTE_CLOSE,
- Event.LINK_FINAL, Event.SESSION_REMOTE_CLOSE,
- Event.CONNECTION_REMOTE_CLOSE, Event.TRANSPORT_TAIL_CLOSED,
- Event.TRANSPORT_CLOSED))
+ self.expect(Event.TRANSPORT_HEAD_CLOSED, Event.LINK_REMOTE_CLOSE,
+ Event.SESSION_REMOTE_CLOSE, Event.CONNECTION_REMOTE_CLOSE,
+ Event.TRANSPORT_TAIL_CLOSED, Event.TRANSPORT_CLOSED)
else:
self.expect(Event.TRANSPORT_HEAD_CLOSED, Event.SESSION_REMOTE_CLOSE,
Event.CONNECTION_REMOTE_CLOSE, Event.TRANSPORT_TAIL_CLOSED,
@@ -2397,8 +2389,7 @@ class TeardownLeakTest(PeerTest):
self.connection.free()
self.transport.unbind()
- self.expect_oneof((Event.LINK_FINAL, Event.CONNECTION_UNBOUND, Event.SESSION_FINAL, Event.CONNECTION_FINAL),
- (Event.CONNECTION_UNBOUND, Event.LINK_FINAL, Event.SESSION_FINAL, Event.CONNECTION_FINAL))
+ self.expect(Event.LINK_FINAL, Event.SESSION_FINAL, Event.CONNECTION_UNBOUND, Event.CONNECTION_FINAL)
def testLocalRemoteLeak(self):
self.doLeak(True, True)
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2ada5d2b/tests/python/proton_tests/sasl.py
----------------------------------------------------------------------
diff --git a/tests/python/proton_tests/sasl.py b/tests/python/proton_tests/sasl.py
index c2dc005..a14a0db 100644
--- a/tests/python/proton_tests/sasl.py
+++ b/tests/python/proton_tests/sasl.py
@@ -175,15 +175,23 @@ class SaslTest(Test):
def test_singleton(self):
"""Verify that only a single instance of SASL can exist per Transport"""
transport = Transport()
+ attr = object()
sasl1 = SASL(transport)
+ sasl1.my_attribute = attr
sasl2 = transport.sasl()
sasl3 = SASL(transport)
- assert sasl1 is sasl2
- assert sasl1 is sasl3
+ assert sasl1 == sasl2
+ assert sasl1 == sasl3
+ assert sasl1.my_attribute == attr
+ assert sasl2.my_attribute == attr
+ assert sasl3.my_attribute == attr
transport = Transport()
sasl1 = transport.sasl()
+ sasl1.my_attribute = attr
sasl2 = SASL(transport)
- assert sasl1 is sasl2
+ assert sasl1 == sasl2
+ assert sasl1.my_attribute == attr
+ assert sasl2.my_attribute == attr
def testSaslSkipped(self):
"""Verify that the server (with SASL) correctly handles a client without SASL"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org