Refactor link handling to use a more event driven approach.
This commit is contained in:
Kenneth Giusti
parent
9935c0cf50
commit
ac54d70c02
python
@ -51,7 +51,7 @@ class ConnectionEventHandler(object):
|
||||
|
||||
def sender_requested(self, connection, link_handle,
|
||||
name, requested_source,
|
||||
properties=None):
|
||||
properties):
|
||||
"""Peer has requested a SenderLink be created."""
|
||||
# call accept_sender to accept new link,
|
||||
# reject_sender to reject it.
|
||||
@ -147,8 +147,9 @@ class Connection(object):
|
||||
self._pn_transport.trace(proton.Transport.TRACE_FRM)
|
||||
|
||||
# indexed by link-name
|
||||
self._sender_links = {} # SenderLink or pn_link if pending
|
||||
self._receiver_links = {} # ReceiverLink or pn_link if pending
|
||||
self._sender_links = {} # SenderLink
|
||||
self._receiver_links = {} # ReceiverLink
|
||||
self._work_queue = set() # either, in need of processing
|
||||
|
||||
self._read_done = False
|
||||
self._write_done = False
|
||||
@ -240,33 +241,9 @@ class Connection(object):
|
||||
self._pn_transport = None
|
||||
self._user_context = None
|
||||
|
||||
def _link_requested(self, pn_link):
|
||||
if pn_link.is_sender and pn_link.name not in self._sender_links:
|
||||
LOG.debug("Remotely initiated Sender needs init")
|
||||
self._sender_links[pn_link.name] = pn_link
|
||||
pn_link.context = None # TODO(kgiusti) update proton.py
|
||||
req_source = ""
|
||||
if pn_link.remote_source.dynamic:
|
||||
req_source = None
|
||||
elif pn_link.remote_source.address:
|
||||
req_source = pn_link.remote_source.address
|
||||
self._handler.sender_requested(self, pn_link.name,
|
||||
pn_link.name, req_source,
|
||||
{"target-address":
|
||||
pn_link.remote_target.address})
|
||||
elif pn_link.is_receiver and pn_link.name not in self._receiver_links:
|
||||
LOG.debug("Remotely initiated Receiver needs init")
|
||||
self._receiver_links[pn_link.name] = pn_link
|
||||
pn_link.context = None # TODO(kgiusti) update proton.py
|
||||
req_target = ""
|
||||
if pn_link.remote_target.dynamic:
|
||||
req_target = None
|
||||
elif pn_link.remote_target.address:
|
||||
req_target = pn_link.remote_target.address
|
||||
self._handler.receiver_requested(self, pn_link.name,
|
||||
pn_link.name, req_target,
|
||||
{"source-address":
|
||||
pn_link.remote_source.address})
|
||||
def _add_work(self, link):
|
||||
"""Add the link to the work queue."""
|
||||
self._work_queue.add(link)
|
||||
|
||||
_REMOTE_REQ = (proton.Endpoint.LOCAL_UNINIT
|
||||
| proton.Endpoint.REMOTE_ACTIVE)
|
||||
@ -315,78 +292,40 @@ class Connection(object):
|
||||
|
||||
pn_link = self._pn_connection.link_head(self._REMOTE_REQ)
|
||||
while pn_link:
|
||||
next_link = pn_link.next(self._LOCAL_UNINIT)
|
||||
if (pn_link.is_sender and
|
||||
pn_link.name not in self._sender_links):
|
||||
LOG.debug("Remotely initiated Sender needs init")
|
||||
link = SenderLink(self, pn_link)
|
||||
self._sender_links[pn_link.name] = link
|
||||
self._add_work(link)
|
||||
elif (pn_link.is_receiver and
|
||||
pn_link.name not in self._receiver_links):
|
||||
LOG.debug("Remotely initiated Receiver needs init")
|
||||
link = ReceiverLink(self, pn_link)
|
||||
self._receiver_links[pn_link.name] = link
|
||||
self._add_work(link)
|
||||
else:
|
||||
LOG.debug("Ignoring request link - name in use %s",
|
||||
pn_link.name)
|
||||
pn_link = pn_link.next(self._REMOTE_REQ)
|
||||
|
||||
if pn_link.state == self._REMOTE_REQ:
|
||||
self._link_requested(pn_link)
|
||||
|
||||
pn_link = next_link
|
||||
|
||||
# TODO(kgiusti) won't scale?
|
||||
# TODO(kgiusti) won't scale - use new Enging event API
|
||||
pn_link = self._pn_connection.link_head(self._ACTIVE)
|
||||
while pn_link:
|
||||
next_link = pn_link.next(self._ACTIVE)
|
||||
|
||||
if pn_link.context and not pn_link.context._active:
|
||||
LOG.debug("Link is up")
|
||||
pn_link.context._active = True
|
||||
if pn_link.is_sender:
|
||||
sender_link = pn_link.context
|
||||
if sender_link._handler:
|
||||
sender_link._handler.sender_active(sender_link)
|
||||
else:
|
||||
receiver_link = pn_link.context
|
||||
if receiver_link._handler:
|
||||
receiver_link._handler.receiver_active(receiver_link)
|
||||
pn_link = next_link
|
||||
|
||||
# process the work queue
|
||||
|
||||
pn_delivery = self._pn_connection.work_head
|
||||
while pn_delivery:
|
||||
next_delivery = pn_delivery.work_next
|
||||
if pn_delivery.link.context:
|
||||
if pn_delivery.link.is_sender:
|
||||
sender_link = pn_delivery.link.context
|
||||
sender_link._delivery_updated(pn_delivery)
|
||||
else:
|
||||
receiver_link = pn_delivery.link.context
|
||||
receiver_link._delivery_updated(pn_delivery)
|
||||
pn_delivery = next_delivery
|
||||
self._add_work(pn_link.context)
|
||||
pn_link = pn_link.next(self._ACTIVE)
|
||||
|
||||
# do endpoint down handling:
|
||||
|
||||
pn_link = self._pn_connection.link_head(self._REMOTE_CLOSE)
|
||||
while pn_link:
|
||||
LOG.debug("Link closed remotely")
|
||||
next_link = pn_link.next(self._REMOTE_CLOSE)
|
||||
# TODO(kgiusti) error reporting
|
||||
if pn_link.context:
|
||||
if pn_link.is_sender:
|
||||
sender_link = pn_link.context
|
||||
handler = pn_link.context._handler
|
||||
if handler:
|
||||
handler.sender_remote_closed(sender_link, None)
|
||||
else:
|
||||
receiver_link = pn_link.context
|
||||
handler = pn_link.context._handler
|
||||
if handler:
|
||||
handler.receiver_remote_closed(receiver_link, None)
|
||||
pn_link = next_link
|
||||
self._add_work(pn_link.context)
|
||||
pn_link = pn_link.next(self._REMOTE_CLOSE)
|
||||
|
||||
pn_link = self._pn_connection.link_head(self._CLOSED)
|
||||
while pn_link:
|
||||
next_link = pn_link.next(self._CLOSED)
|
||||
if pn_link.context and pn_link.context._active:
|
||||
LOG.debug("Link close completed")
|
||||
pn_link.context._active = False
|
||||
if pn_link.is_sender:
|
||||
sender_link = pn_link.context
|
||||
sender_link._handler.sender_closed(sender_link)
|
||||
else:
|
||||
receiver_link = pn_link.context
|
||||
receiver_link._handler.receiver_closed(receiver_link)
|
||||
pn_link = next_link
|
||||
self._add_work(pn_link.context)
|
||||
pn_link = pn_link.next(self._CLOSED)
|
||||
|
||||
pn_session = self._pn_connection.session_head(self._REMOTE_CLOSE)
|
||||
while pn_session:
|
||||
@ -401,6 +340,19 @@ class Connection(object):
|
||||
LOG.debug("Connection close complete")
|
||||
self._handler.connection_closed(self)
|
||||
|
||||
# service links needing attention:
|
||||
while self._work_queue:
|
||||
link = self._work_queue.pop()
|
||||
link._process()
|
||||
|
||||
# process the delivery work queue
|
||||
|
||||
pn_delivery = self._pn_connection.work_head
|
||||
while pn_delivery:
|
||||
next_delivery = pn_delivery.work_next
|
||||
pn_delivery.link.context._process_delivery(pn_delivery)
|
||||
pn_delivery = next_delivery
|
||||
|
||||
# DEBUG LINK "LEAK"
|
||||
# count = 0
|
||||
# link = self._pn_connection.link_head(0)
|
||||
@ -500,36 +452,32 @@ class Connection(object):
|
||||
raise KeyError("Sender %s already exists!" % ident)
|
||||
|
||||
pn_link = self._pn_session.sender(ident)
|
||||
if pn_link:
|
||||
s = SenderLink(self, pn_link,
|
||||
source_address, target_address,
|
||||
event_handler, properties)
|
||||
self._sender_links[ident] = s
|
||||
return s
|
||||
return None
|
||||
sl = SenderLink(self, pn_link)
|
||||
sl.configure(target_address, source_address, event_handler, properties)
|
||||
self._sender_links[ident] = sl
|
||||
return sl
|
||||
|
||||
def accept_sender(self, link_handle, source_override=None,
|
||||
event_handler=None, properties=None):
|
||||
pn_link = self._sender_links.get(link_handle)
|
||||
if not pn_link or not isinstance(pn_link, proton.Sender):
|
||||
link = self._sender_links.get(link_handle)
|
||||
if not link:
|
||||
raise Exception("Invalid link_handle: %s" % link_handle)
|
||||
pn_link = link._pn_link
|
||||
if pn_link.remote_source.dynamic and not source_override:
|
||||
raise Exception("A source address must be supplied!")
|
||||
source_addr = source_override or pn_link.remote_source.address
|
||||
link = SenderLink(self, pn_link,
|
||||
source_addr,
|
||||
pn_link.remote_target.address,
|
||||
event_handler, properties)
|
||||
self._sender_links[link_handle] = link
|
||||
link.configure(pn_link.remote_target.address,
|
||||
source_addr,
|
||||
event_handler, properties)
|
||||
return link
|
||||
|
||||
def reject_sender(self, link_handle, reason):
|
||||
pn_link = self._sender_links.get(link_handle)
|
||||
if not pn_link or not isinstance(pn_link, proton.Sender):
|
||||
"""Rejects the SenderLink, and destroys the handle."""
|
||||
link = self._sender_links.get(link_handle)
|
||||
if not link:
|
||||
raise Exception("Invalid link_handle: %s" % link_handle)
|
||||
del self._sender_links[link_handle]
|
||||
# TODO(kgiusti) support reason for close
|
||||
pn_link.close()
|
||||
link.reject(reason)
|
||||
link.destroy()
|
||||
|
||||
def create_receiver(self, target_address, source_address=None,
|
||||
event_handler=None, name=None, properties=None):
|
||||
@ -539,35 +487,31 @@ class Connection(object):
|
||||
raise KeyError("Receiver %s already exists!" % ident)
|
||||
|
||||
pn_link = self._pn_session.receiver(ident)
|
||||
if pn_link:
|
||||
r = ReceiverLink(self, pn_link, target_address,
|
||||
source_address, event_handler, properties)
|
||||
self._receiver_links[ident] = r
|
||||
return r
|
||||
return None
|
||||
rl = ReceiverLink(self, pn_link)
|
||||
rl.configure(target_address, source_address, event_handler, properties)
|
||||
self._receiver_links[ident] = rl
|
||||
return rl
|
||||
|
||||
def accept_receiver(self, link_handle, target_override=None,
|
||||
event_handler=None, properties=None):
|
||||
pn_link = self._receiver_links.get(link_handle)
|
||||
if not pn_link or not isinstance(pn_link, proton.Receiver):
|
||||
link = self._receiver_links.get(link_handle)
|
||||
if not link:
|
||||
raise Exception("Invalid link_handle: %s" % link_handle)
|
||||
pn_link = link._pn_link
|
||||
if pn_link.remote_target.dynamic and not target_override:
|
||||
raise Exception("A target address must be supplied!")
|
||||
target_addr = target_override or pn_link.remote_target.address
|
||||
link = ReceiverLink(self, pn_link,
|
||||
target_addr,
|
||||
pn_link.remote_source.address,
|
||||
event_handler, properties)
|
||||
self._receiver_links[link_handle] = link
|
||||
link.configure(target_addr,
|
||||
pn_link.remote_source.address,
|
||||
event_handler, properties)
|
||||
return link
|
||||
|
||||
def reject_receiver(self, link_handle, reason):
|
||||
pn_link = self._receiver_links.get(link_handle)
|
||||
if not pn_link or not isinstance(pn_link, proton.Receiver):
|
||||
link = self._receiver_links.get(link_handle)
|
||||
if not link:
|
||||
raise Exception("Invalid link_handle: %s" % link_handle)
|
||||
del self._receiver_links[link_handle]
|
||||
# TODO(kgiusti) support reason for close
|
||||
pn_link.close()
|
||||
link.reject(reason)
|
||||
link.destroy()
|
||||
|
||||
def _remove_sender(self, name):
|
||||
if name in self._sender_links:
|
||||
|
||||
251
python/link.py
251
python/link.py
@ -28,30 +28,155 @@ import proton
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
# link lifecycle state:
|
||||
_STATE_UNINIT = 0 # initial state
|
||||
_STATE_PENDING = 1 # waiting for remote to open
|
||||
_STATE_REQUESTED = 2 # waiting for local to open
|
||||
_STATE_ACTIVE = 3
|
||||
_STATE_NEED_CLOSE = 4 # remote initiated close
|
||||
_STATE_CLOSING = 5 # locally closed, pending remote
|
||||
_STATE_CLOSED = 6 # terminal state
|
||||
|
||||
# proton endpoint states:
|
||||
_LOCAL_UNINIT = proton.Endpoint.LOCAL_UNINIT
|
||||
_LOCAL_ACTIVE = proton.Endpoint.LOCAL_ACTIVE
|
||||
_LOCAL_CLOSED = proton.Endpoint.LOCAL_CLOSED
|
||||
_REMOTE_UNINIT = proton.Endpoint.REMOTE_UNINIT
|
||||
_REMOTE_ACTIVE = proton.Endpoint.REMOTE_ACTIVE
|
||||
_REMOTE_CLOSED = proton.Endpoint.REMOTE_CLOSED
|
||||
|
||||
def _do_pending(link):
|
||||
"""The application has opened a new link."""
|
||||
# nothing to do but wait for the remote:
|
||||
return _STATE_PENDING
|
||||
|
||||
def _do_requested(link):
|
||||
"""The peer has requested a new link be created."""
|
||||
LOG.debug("Remote has initiated a link")
|
||||
pn_link = link._pn_link
|
||||
if isinstance(link, SenderLink):
|
||||
# has the remote requested a source address?
|
||||
req_source = ""
|
||||
if pn_link.remote_source.dynamic:
|
||||
req_source = None
|
||||
elif pn_link.remote_source.address:
|
||||
req_source = pn_link.remote_source.address
|
||||
handler = link._connection._handler
|
||||
if handler:
|
||||
handler.sender_requested(link._connection,
|
||||
pn_link.name, # handle
|
||||
pn_link.name,
|
||||
req_source,
|
||||
{"target-address":
|
||||
pn_link.remote_target.address})
|
||||
else:
|
||||
# has the remote requested a target address?
|
||||
req_target = ""
|
||||
if pn_link.remote_target.dynamic:
|
||||
req_target = None
|
||||
elif pn_link.remote_target.address:
|
||||
req_target = pn_link.remote_target.address
|
||||
handler = link._connection._handler
|
||||
if handler:
|
||||
handler.receiver_requested(link._connection,
|
||||
pn_link.name, # handle
|
||||
pn_link.name,
|
||||
req_target,
|
||||
{"source-address":
|
||||
pn_link.remote_source.address})
|
||||
return _STATE_REQUESTED
|
||||
|
||||
def _do_active(link):
|
||||
"""Both ends of the link have become active."""
|
||||
LOG.debug("Link is up")
|
||||
if link._handler:
|
||||
if isinstance(link, SenderLink):
|
||||
link._handler.sender_active(link)
|
||||
else:
|
||||
link._handler.receiver_active(link)
|
||||
return _STATE_ACTIVE
|
||||
|
||||
def _do_need_close(link):
|
||||
"""The remote has closed its end of the link."""
|
||||
# TODO(kgiusti) error reporting
|
||||
LOG.debug("Link remote closed")
|
||||
if link._handler:
|
||||
if isinstance(link, SenderLink):
|
||||
link._handler.sender_remote_closed(link, None)
|
||||
else:
|
||||
link._handler.receiver_remote_closed(link, None)
|
||||
return _STATE_NEED_CLOSE
|
||||
|
||||
def _do_closing(link):
|
||||
"""Locally closed, remote end still active."""
|
||||
# nothing to do but wait for the remote
|
||||
return _STATE_CLOSING
|
||||
|
||||
def _do_closed(link):
|
||||
"""Both ends of the link have closed."""
|
||||
LOG.debug("Link close completed")
|
||||
if link._handler:
|
||||
if isinstance(link, SenderLink):
|
||||
link._handler.sender_closed(link)
|
||||
else:
|
||||
link._handler.receiver_closed(link)
|
||||
return _STATE_CLOSED
|
||||
|
||||
# Given the current endpoint state for the link, move to the next state do any
|
||||
# state-specific processing:
|
||||
_EP_STATE_MACHINE = [
|
||||
# _STATE_UNINIT:
|
||||
{(_LOCAL_ACTIVE | _REMOTE_UNINIT): _do_pending,
|
||||
(_LOCAL_UNINIT | _REMOTE_ACTIVE): _do_requested},
|
||||
# _STATE_PENDING:
|
||||
{(_LOCAL_ACTIVE | _REMOTE_ACTIVE): _do_active,
|
||||
(_LOCAL_ACTIVE | _REMOTE_CLOSED): _do_need_close,
|
||||
(_LOCAL_CLOSED | _REMOTE_UNINIT): _do_closed,
|
||||
(_LOCAL_CLOSED | _REMOTE_CLOSED): _do_closed},
|
||||
# _STATE_REQESTED:
|
||||
{(_LOCAL_ACTIVE | _REMOTE_ACTIVE): _do_active,
|
||||
(_LOCAL_CLOSED | _REMOTE_ACTIVE): _do_closing,
|
||||
(_LOCAL_ACTIVE | _REMOTE_CLOSED): _do_need_close},
|
||||
# _STATE_ACTIVE:
|
||||
{(_LOCAL_ACTIVE | _REMOTE_CLOSED): _do_need_close,
|
||||
(_LOCAL_CLOSED | _REMOTE_ACTIVE): _do_closing,
|
||||
(_LOCAL_CLOSED | _REMOTE_CLOSED): _do_closed},
|
||||
# _STATE_NEED_CLOSE:
|
||||
{(_LOCAL_CLOSED | _REMOTE_CLOSED): _do_closed},
|
||||
# _STATE_CLOSING:
|
||||
{(_LOCAL_CLOSED | _REMOTE_CLOSED): _do_closed}]
|
||||
|
||||
class _Link(object):
|
||||
"""A generic Link base class."""
|
||||
def __init__(self, connection, pn_link,
|
||||
target_address, source_address,
|
||||
handler, properties):
|
||||
|
||||
def __init__(self, connection, pn_link):
|
||||
self._state = _STATE_UNINIT
|
||||
# last known endpoint state:
|
||||
self._ep_state = _LOCAL_UNINIT | _REMOTE_UNINIT
|
||||
self._connection = connection
|
||||
self._name = pn_link.name
|
||||
self._handler = handler
|
||||
self._properties = properties
|
||||
self._handler = None
|
||||
self._properties = None
|
||||
self._user_context = None
|
||||
self._active = False
|
||||
# TODO(kgiusti): raise jira to add 'context' attr to api
|
||||
self._pn_link = pn_link
|
||||
pn_link.context = self
|
||||
|
||||
def configure(self, target_address, source_address, handler, properties):
|
||||
"""Assign addresses, properties, etc."""
|
||||
self._handler = handler
|
||||
self._properties = properties
|
||||
|
||||
if target_address is None:
|
||||
assert pn_link.is_sender, "Dynamic target not allowed"
|
||||
if not self._pn_link.is_sender:
|
||||
raise Exception("Dynamic target not allowed")
|
||||
self._pn_link.target.dynamic = True
|
||||
elif target_address:
|
||||
self._pn_link.target.address = target_address
|
||||
|
||||
if source_address is None:
|
||||
assert pn_link.is_receiver, "Dynamic source not allowed"
|
||||
if not self._pn_link.is_receiver:
|
||||
raise Exception("Dynamic source not allowed")
|
||||
self._pn_link.source.dynamic = True
|
||||
elif source_address:
|
||||
self._pn_link.source.address = source_address
|
||||
@ -73,7 +198,9 @@ class _Link(object):
|
||||
return self._name
|
||||
|
||||
def open(self):
|
||||
LOG.debug("Opening the link.")
|
||||
self._pn_link.open()
|
||||
self._connection._add_work(self)
|
||||
|
||||
def _get_user_context(self):
|
||||
return self._user_context
|
||||
@ -106,7 +233,9 @@ class _Link(object):
|
||||
return self._pn_link.remote_target.address
|
||||
|
||||
def close(self, error=None):
|
||||
LOG.debug("Closing the link.")
|
||||
self._pn_link.close()
|
||||
self._connection._add_work(self)
|
||||
|
||||
@property
|
||||
def closed(self):
|
||||
@ -122,6 +251,20 @@ class _Link(object):
|
||||
self._pn_link.free()
|
||||
self._pn_link = None
|
||||
|
||||
def _process(self):
|
||||
"""Link state machine processing."""
|
||||
# check for transitions in Endpoint state:
|
||||
pn_link = self._pn_link
|
||||
if self._state != _STATE_CLOSED:
|
||||
ep_state = pn_link.state
|
||||
if ep_state != self._ep_state:
|
||||
LOG.debug("link state: %s old ep: %s new ep: %s",
|
||||
self._state, hex(self._ep_state), hex(ep_state))
|
||||
self._ep_state = ep_state
|
||||
self._state = _EP_STATE_MACHINE[self._state][ep_state](self)
|
||||
|
||||
def _process_delivery(self, pn_delivery):
|
||||
pass
|
||||
|
||||
class SenderEventHandler(object):
|
||||
def sender_active(self, sender_link):
|
||||
@ -146,11 +289,8 @@ class SenderLink(_Link):
|
||||
RELEASED = 3
|
||||
MODIFIED = 4
|
||||
|
||||
def __init__(self, connection, pn_link, source_address,
|
||||
target_address, event_handler, properties):
|
||||
super(SenderLink, self).__init__(connection, pn_link,
|
||||
target_address, source_address,
|
||||
event_handler, properties)
|
||||
def __init__(self, connection, pn_link):
|
||||
super(SenderLink, self).__init__(connection, pn_link)
|
||||
self._pending_sends = collections.deque()
|
||||
self._pending_acks = {}
|
||||
self._next_deadline = 0
|
||||
@ -163,17 +303,18 @@ class SenderLink(_Link):
|
||||
self._pending_sends.append((message, delivery_callback, handle,
|
||||
deadline))
|
||||
# TODO(kgiusti) deadline not supported yet
|
||||
assert not deadline, "send timeout not supported yet!"
|
||||
if deadline:
|
||||
raise NotImplementedError("send timeout not supported yet!")
|
||||
if deadline and (self._next_deadline == 0 or
|
||||
self._next_deadline > deadline):
|
||||
self._next_deadline = deadline
|
||||
|
||||
delivery = self._pn_link.delivery("tag-%x" % self._next_tag)
|
||||
pn_delivery = self._pn_link.delivery("tag-%x" % self._next_tag)
|
||||
self._next_tag += 1
|
||||
|
||||
if delivery.writable:
|
||||
if pn_delivery.writable:
|
||||
send_req = self._pending_sends.popleft()
|
||||
self._write_msg(delivery, send_req)
|
||||
self._write_msg(pn_delivery, send_req)
|
||||
|
||||
return 0
|
||||
|
||||
@ -197,14 +338,20 @@ class SenderLink(_Link):
|
||||
self._pending_acks.clear()
|
||||
super(SenderLink, self).close()
|
||||
|
||||
def reject(self, reason):
|
||||
"""See Link Reject, AMQP1.0 spec."""
|
||||
# TODO(kgiusti) support reason for close
|
||||
self._pn_link.source.type = proton.Terminus.UNSPECIFIED
|
||||
self._pn_link.open()
|
||||
self._pn_link.close()
|
||||
|
||||
def destroy(self):
|
||||
self._connection._remove_sender(self._name)
|
||||
self._connection = None
|
||||
super(SenderLink, self).destroy()
|
||||
|
||||
def _delivery_updated(self, delivery):
|
||||
# A delivery has changed state.
|
||||
# Do we need to know?
|
||||
def _process_delivery(self, pn_delivery):
|
||||
"""Check if the delivery can be processed."""
|
||||
_disposition_state_map = {
|
||||
proton.Disposition.ACCEPTED: SenderLink.ACCEPTED,
|
||||
proton.Disposition.REJECTED: SenderLink.REJECTED,
|
||||
@ -212,41 +359,41 @@ class SenderLink(_Link):
|
||||
proton.Disposition.MODIFIED: SenderLink.MODIFIED,
|
||||
}
|
||||
|
||||
if delivery.tag in self._pending_acks:
|
||||
if delivery.settled: # remote has finished
|
||||
LOG.debug("delivery updated, remote state=%s",
|
||||
str(delivery.remote_state))
|
||||
|
||||
send_req = self._pending_acks.pop(delivery.tag)
|
||||
state = _disposition_state_map.get(delivery.remote_state,
|
||||
if pn_delivery.tag in self._pending_acks:
|
||||
if pn_delivery.settled: # remote has finished
|
||||
LOG.debug("Remote has settled a sent msg")
|
||||
send_req = self._pending_acks.pop(pn_delivery.tag)
|
||||
state = _disposition_state_map.get(pn_delivery.remote_state,
|
||||
self.UNKNOWN)
|
||||
cb = send_req[1]
|
||||
handle = send_req[2]
|
||||
|
||||
cb(self, handle, state, None)
|
||||
delivery.settle()
|
||||
pn_delivery.settle()
|
||||
else:
|
||||
# not for a sent msg, use it to send the next
|
||||
if delivery.writable and self._pending_sends:
|
||||
if pn_delivery.writable and self._pending_sends:
|
||||
send_req = self._pending_sends.popleft()
|
||||
self._write_msg(delivery, send_req)
|
||||
self._write_msg(pn_delivery, send_req)
|
||||
else:
|
||||
# what else is there???
|
||||
delivery.settle()
|
||||
pn_delivery.settle()
|
||||
|
||||
def _write_msg(self, delivery, send_req):
|
||||
def _write_msg(self, pn_delivery, send_req):
|
||||
# given a writable delivery, send a message
|
||||
# send_req = (msg, cb, handle, deadline)
|
||||
LOG.debug("Sending a pending message")
|
||||
msg = send_req[0]
|
||||
cb = send_req[1]
|
||||
self._pn_link.send(msg.encode())
|
||||
self._pn_link.advance()
|
||||
if cb: # delivery callback given
|
||||
assert delivery.tag not in self._pending_acks
|
||||
self._pending_acks[delivery.tag] = send_req
|
||||
if pn_delivery.tag in self._pending_acks:
|
||||
raise Exception("Duplicate delivery tag?")
|
||||
self._pending_acks[pn_delivery.tag] = send_req
|
||||
else:
|
||||
# no status required, so settle it now.
|
||||
delivery.settle()
|
||||
pn_delivery.settle()
|
||||
|
||||
|
||||
class ReceiverEventHandler(object):
|
||||
@ -265,11 +412,8 @@ class ReceiverEventHandler(object):
|
||||
|
||||
|
||||
class ReceiverLink(_Link):
|
||||
def __init__(self, connection, pn_link, target_address,
|
||||
source_address, event_handler, properties):
|
||||
super(ReceiverLink, self).__init__(connection, pn_link,
|
||||
target_address, source_address,
|
||||
event_handler, properties)
|
||||
def __init__(self, connection, pn_link):
|
||||
super(ReceiverLink, self).__init__(connection, pn_link)
|
||||
self._next_handle = 0
|
||||
self._unsettled_deliveries = {} # indexed by handle
|
||||
|
||||
@ -294,17 +438,24 @@ class ReceiverLink(_Link):
|
||||
def message_modified(self, handle):
|
||||
self._settle_delivery(handle, proton.Delivery.MODIFIED)
|
||||
|
||||
def reject(self, reason):
|
||||
"""See Link Reject, AMQP1.0 spec."""
|
||||
# TODO(kgiusti) support reason for close
|
||||
self._pn_link.target.type = proton.Terminus.UNSPECIFIED
|
||||
self._pn_link.open()
|
||||
self._pn_link.close()
|
||||
|
||||
def destroy(self):
|
||||
self._connection._remove_receiver(self._name)
|
||||
self._connection = None
|
||||
super(ReceiverLink, self).destroy()
|
||||
|
||||
def _delivery_updated(self, delivery):
|
||||
# a receive delivery changed state
|
||||
def _process_delivery(self, pn_delivery):
|
||||
"""Check if the delivery can be processed."""
|
||||
# TODO(kgiusti): multi-frame message transfer
|
||||
LOG.debug("Receive delivery updated")
|
||||
if delivery.readable:
|
||||
data = self._pn_link.recv(delivery.pending)
|
||||
if pn_delivery.readable:
|
||||
LOG.debug("Receive delivery readable")
|
||||
data = self._pn_link.recv(pn_delivery.pending)
|
||||
msg = proton.Message()
|
||||
msg.decode(data)
|
||||
self._pn_link.advance()
|
||||
@ -312,16 +463,16 @@ class ReceiverLink(_Link):
|
||||
if self._handler:
|
||||
handle = "rmsg-%s:%x" % (self._name, self._next_handle)
|
||||
self._next_handle += 1
|
||||
self._unsettled_deliveries[handle] = delivery
|
||||
self._unsettled_deliveries[handle] = pn_delivery
|
||||
self._handler.message_received(self, msg, handle)
|
||||
else:
|
||||
# TODO(kgiusti): is it ok to assume Delivery.REJECTED?
|
||||
delivery.settle()
|
||||
pn_delivery.settle()
|
||||
|
||||
def _settle_delivery(self, handle, result):
|
||||
# settle delivery associated with a handle
|
||||
if handle not in self._unsettled_deliveries:
|
||||
raise Exception("Invalid message handle: %s" % str(handle))
|
||||
delivery = self._unsettled_deliveries.pop(handle)
|
||||
delivery.update(result)
|
||||
delivery.settle()
|
||||
pn_delivery = self._unsettled_deliveries.pop(handle)
|
||||
pn_delivery.update(result)
|
||||
pn_delivery.settle()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user