Source code for proton._handlers

#
# 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 __future__ import absolute_import

import errno
import logging
import socket
import time
import weakref

from ._condition import Condition
from ._delivery import Delivery
from ._endpoints import Endpoint
from ._events import Event, Handler, _dispatch
from ._exceptions import ProtonException
from ._io import IO
from ._message import Message
from ._selectable import Selectable
from ._transport import Transport
from ._url import Url

log = logging.getLogger("proton")


class OutgoingMessageHandler(Handler):
    """
    A utility for simpler and more intuitive handling of delivery
    events related to outgoing i.e. sent messages.

    :param auto_settle: If ``True``, settle all messages (default). Otherwise
        messages must be explicitly settled.
    :type auto_settle: ``bool``
    :param delegate: A client handler for the endpoint event
    """

    def __init__(self, auto_settle=True, delegate=None):
        self.auto_settle = auto_settle
        self.delegate = delegate

    def on_link_flow(self, event):
        if event.link.is_sender and event.link.credit \
                and event.link.state & Endpoint.LOCAL_ACTIVE \
                and event.link.state & Endpoint.REMOTE_ACTIVE:
            self.on_sendable(event)

    def on_delivery(self, event):
        dlv = event.delivery
        if dlv.link.is_sender and dlv.updated:
            if dlv.remote_state == Delivery.ACCEPTED:
                self.on_accepted(event)
            elif dlv.remote_state == Delivery.REJECTED:
                self.on_rejected(event)
            elif dlv.remote_state == Delivery.RELEASED or dlv.remote_state == Delivery.MODIFIED:
                self.on_released(event)
            if dlv.settled:
                self.on_settled(event)
                if self.auto_settle:
                    dlv.settle()

    def on_sendable(self, event):
        """
        Called when the sender link has credit and messages can
        therefore be transferred.

        :param event: The underlying event object. Use this to obtain further
            information on the event.
        :type event: :class:`proton.Event`
        """
        if self.delegate is not None:
            _dispatch(self.delegate, 'on_sendable', event)

    def on_accepted(self, event):
        """
        Called when the remote peer accepts an outgoing message.

        :param event: The underlying event object. Use this to obtain further
            information on the event.
        :type event: :class:`proton.Event`
        """
        if self.delegate is not None:
            _dispatch(self.delegate, 'on_accepted', event)

    def on_rejected(self, event):
        """
        Called when the remote peer rejects an outgoing message.

        :param event: The underlying event object. Use this to obtain further
            information on the event.
        :type event: :class:`proton.Event`
        """
        if self.delegate is not None:
            _dispatch(self.delegate, 'on_rejected', event)

    def on_released(self, event):
        """
        Called when the remote peer releases an outgoing message. Note
        that this may be in response to either the ``RELEASE`` or ``MODIFIED``
        state as defined by the AMQP specification.

        :param event: The underlying event object. Use this to obtain further
            information on the event.
        :type event: :class:`proton.Event`
        """
        if self.delegate is not None:
            _dispatch(self.delegate, 'on_released', event)

    def on_settled(self, event):
        """
        Called when the remote peer has settled the outgoing
        message. This is the point at which it should never be
        retransmitted.

        :param event: The underlying event object. Use this to obtain further
            information on the event.
        :type event: :class:`proton.Event`
        """
        if self.delegate is not None:
            _dispatch(self.delegate, 'on_settled', event)


def recv_msg(delivery):
    msg = Message()
    msg.decode(delivery.link.recv(delivery.pending))
    delivery.link.advance()
    return msg


[docs]class Reject(ProtonException): """ An exception that indicates a message should be rejected. """ pass
[docs]class Release(ProtonException): """ An exception that indicates a message should be released. """ pass
class Acking(object): """ A class containing methods for handling received messages. """ def accept(self, delivery): """ Accepts a received message. .. note:: This method cannot currently be used in combination with transactions. See :class:`proton.reactor.Transaction` for transactional methods. :param delivery: The message delivery tracking object :type delivery: :class:`proton.Delivery` """ self.settle(delivery, Delivery.ACCEPTED) def reject(self, delivery): """ Rejects a received message that is considered invalid or unprocessable. .. note:: This method cannot currently be used in combination with transactions. See :class:`proton.reactor.Transaction` for transactional methods. :param delivery: The message delivery tracking object :type delivery: :class:`proton.Delivery` """ self.settle(delivery, Delivery.REJECTED) def release(self, delivery, delivered=True): """ Releases a received message, making it available at the source for any (other) interested receiver. The ``delivered`` parameter indicates whether this should be considered a delivery attempt (and the delivery count updated) or not. .. note:: This method cannot currently be used in combination with transactions. See :class:`proton.reactor.Transaction` for transactional methods. :param delivery: The message delivery tracking object :type delivery: :class:`proton.Delivery` :param delivered: If ``True``, the message will be annotated with a delivery attempt (setting delivery flag :const:`proton.Delivery.MODIFIED`). Otherwise, the message will be returned without the annotation and released (setting delivery flag :const:`proton.Delivery.RELEASED` :type delivered: ``bool`` """ if delivered: self.settle(delivery, Delivery.MODIFIED) else: self.settle(delivery, Delivery.RELEASED) def settle(self, delivery, state=None): """ Settles the message delivery, and optionally updating the delivery state. :param delivery: The message delivery tracking object :type delivery: :class:`proton.Delivery` :param state: The delivery state, or ``None`` if not update is to be performed. :type state: ``int`` or ``None`` """ if state: delivery.update(state) delivery.settle() class IncomingMessageHandler(Handler, Acking): """ A utility for simpler and more intuitive handling of delivery events related to incoming i.e. received messages. :type auto_accept: ``bool`` :param auto_accept: If ``True``, accept all messages (default). Otherwise messages must be individually accepted or rejected. :param delegate: A client handler for the endpoint event """ def __init__(self, auto_accept=True, delegate=None): self.delegate = delegate self.auto_accept = auto_accept def on_delivery(self, event): dlv = event.delivery if not dlv.link.is_receiver: return if dlv.aborted: self.on_aborted(event) dlv.settle() elif dlv.readable and not dlv.partial: event.message = recv_msg(dlv) if event.link.state & Endpoint.LOCAL_CLOSED: if self.auto_accept: dlv.update(Delivery.RELEASED) dlv.settle() else: try: self.on_message(event) if self.auto_accept: dlv.update(Delivery.ACCEPTED) dlv.settle() except Reject: dlv.update(Delivery.REJECTED) dlv.settle() except Release: dlv.update(Delivery.MODIFIED) dlv.settle() elif dlv.updated and dlv.settled: self.on_settled(event) def on_message(self, event): """ Called when a message is received. The message itself can be obtained as a property on the event. For the purpose of referring to this message in further actions (e.g. if explicitly accepting it, the ``delivery`` should be used, also obtainable via a property on the event. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_message', event) def on_settled(self, event): """ Callback for when a message delivery is settled by the remote peer. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_settled', event) def on_aborted(self, event): """ Callback for when a message delivery is aborted by the remote peer. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_aborted', event) class EndpointStateHandler(Handler): """ A utility that exposes 'endpoint' events - ie the open/close for links, sessions and connections in a more intuitive manner. A ``XXX_opened()`` method will be called when both local and remote peers have opened the link, session or connection. This can be used to confirm a locally initiated action for example. A ``XXX_opening()`` method will be called when the remote peer has requested an open that was not initiated locally. By default this will simply open locally, which then triggers the ``XXX_opened()`` call. The same applies to close. :param peer_close_is_error: If ``True``, a peer endpoint closing will be treated as an error with an error callback. Otherwise (default), the normal callbacks for the closing will occur. :type peer_close_is_error: ``bool`` :param delegate: A client handler for the endpoint event """ def __init__(self, peer_close_is_error=False, delegate=None): self.delegate = delegate self.peer_close_is_error = peer_close_is_error @classmethod def is_local_open(cls, endpoint): """ Test if local ``enpoint`` is open (ie has state :const:`proton.Endpoint.LOCAL_ACTIVE`). :param endpoint: The local endpoint to be tested. :type endpoint: Any child of :class:`proton.Endpoint` :return: ``True`` if local endpoint is in state :const:`proton.Endpoint.LOCAL_ACTIVE`, ``False`` otherwise. :rtype: ``bool`` """ return endpoint.state & Endpoint.LOCAL_ACTIVE @classmethod def is_local_uninitialised(cls, endpoint): """ Test if local ``enpoint`` is uninitialised (ie has state :const:`proton.Endpoint.LOCAL_UNINIT`). :param endpoint: The local endpoint to be tested. :type endpoint: Any child of :class:`proton.Endpoint` :return: ``True`` if local endpoint is in state :const:`proton.Endpoint.LOCAL_UNINIT`, ``False`` otherwise. :rtype: ``bool`` """ return endpoint.state & Endpoint.LOCAL_UNINIT @classmethod def is_local_closed(cls, endpoint): """ Test if local ``enpoint`` is closed (ie has state :const:`proton.Endpoint.LOCAL_CLOSED`). :param endpoint: The local endpoint to be tested. :type endpoint: Any child of :class:`proton.Endpoint` :return: ``True`` if local endpoint is in state :const:`proton.Endpoint.LOCAL_CLOSED`, ``False`` otherwise. :rtype: ``bool`` """ return endpoint.state & Endpoint.LOCAL_CLOSED @classmethod def is_remote_open(cls, endpoint): """ Test if remote ``enpoint`` is open (ie has state :const:`proton.Endpoint.LOCAL_ACTIVE`). :param endpoint: The remote endpoint to be tested. :type endpoint: Any child of :class:`proton.Endpoint` :return: ``True`` if remote endpoint is in state :const:`proton.Endpoint.LOCAL_ACTIVE`, ``False`` otherwise. :rtype: ``bool`` """ return endpoint.state & Endpoint.REMOTE_ACTIVE @classmethod def is_remote_closed(cls, endpoint): """ Test if remote ``enpoint`` is closed (ie has state :const:`proton.Endpoint.REMOTE_CLOSED`). :param endpoint: The remote endpoint to be tested. :type endpoint: Any child of :class:`proton.Endpoint` :return: ``True`` if remote endpoint is in state :const:`proton.Endpoint.REMOTE_CLOSED`, ``False`` otherwise. :rtype: ``bool`` """ return endpoint.state & Endpoint.REMOTE_CLOSED @classmethod def print_error(cls, endpoint, endpoint_type): """ Logs an error message related to an error condition at an endpoint. :param endpoint: The endpoint to be tested :type endpoint: :class:`proton.Endpoint` :param endpoint_type: The endpoint type as a string to be printed in the log message. :type endpoint_type: ``str`` """ if endpoint.remote_condition: log.error(endpoint.remote_condition.description or endpoint.remote_condition.name) elif cls.is_local_open(endpoint) and cls.is_remote_closed(endpoint): log.error("%s closed by peer" % endpoint_type) def on_link_remote_close(self, event): if event.link.remote_condition: self.on_link_error(event) elif self.is_local_closed(event.link): self.on_link_closed(event) else: self.on_link_closing(event) event.link.close() def on_session_remote_close(self, event): if event.session.remote_condition: self.on_session_error(event) elif self.is_local_closed(event.session): self.on_session_closed(event) else: self.on_session_closing(event) event.session.close() def on_connection_remote_close(self, event): if event.connection.remote_condition: if event.connection.remote_condition.name == "amqp:connection:forced": # Treat this the same as just having the transport closed by the peer without # sending any events. Allow reconnection to happen transparently. return self.on_connection_error(event) elif self.is_local_closed(event.connection): self.on_connection_closed(event) else: self.on_connection_closing(event) event.connection.close() def on_connection_local_open(self, event): if self.is_remote_open(event.connection): self.on_connection_opened(event) def on_connection_remote_open(self, event): if self.is_local_open(event.connection): self.on_connection_opened(event) elif self.is_local_uninitialised(event.connection): self.on_connection_opening(event) event.connection.open() def on_session_local_open(self, event): if self.is_remote_open(event.session): self.on_session_opened(event) def on_session_remote_open(self, event): if self.is_local_open(event.session): self.on_session_opened(event) elif self.is_local_uninitialised(event.session): self.on_session_opening(event) event.session.open() def on_link_local_open(self, event): if self.is_remote_open(event.link): self.on_link_opened(event) def on_link_remote_open(self, event): if self.is_local_open(event.link): self.on_link_opened(event) elif self.is_local_uninitialised(event.link): self.on_link_opening(event) event.link.open() def on_connection_opened(self, event): """ Callback for when both the local and remote endpoints of a connection have opened. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_connection_opened', event) def on_session_opened(self, event): """ Callback for when both the local and remote endpoints of a session have opened. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_session_opened', event) def on_link_opened(self, event): """ Callback for when both the local and remote endpoints of a link have opened. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_link_opened', event) def on_connection_opening(self, event): """ Callback for when a remote peer initiates the opening of a connection. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_connection_opening', event) def on_session_opening(self, event): """ Callback for when a remote peer initiates the opening of a session. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_session_opening', event) def on_link_opening(self, event): """ Callback for when a remote peer initiates the opening of a link. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_link_opening', event) def on_connection_error(self, event): """ Callback for when an initiated connection open fails. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_connection_error', event) else: self.print_error(event.connection, "connection") def on_session_error(self, event): """ Callback for when an initiated session open fails. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_session_error', event) else: self.print_error(event.session, "session") event.connection.close() def on_link_error(self, event): """ Callback for when an initiated link open fails. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_link_error', event) else: self.print_error(event.link, "link") event.connection.close() def on_connection_closed(self, event): """ Callback for when both the local and remote endpoints of a connection have closed. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_connection_closed', event) def on_session_closed(self, event): """ Callback for when both the local and remote endpoints of a session have closed. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_session_closed', event) def on_link_closed(self, event): """ Callback for when both the local and remote endpoints of a link have closed. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_link_closed', event) def on_connection_closing(self, event): """ Callback for when a remote peer initiates the closing of a connection. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_connection_closing', event) elif self.peer_close_is_error: self.on_connection_error(event) def on_session_closing(self, event): """ Callback for when a remote peer initiates the closing of a session. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_session_closing', event) elif self.peer_close_is_error: self.on_session_error(event) def on_link_closing(self, event): """ Callback for when a remote peer initiates the closing of a link. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None: _dispatch(self.delegate, 'on_link_closing', event) elif self.peer_close_is_error: self.on_link_error(event) def on_transport_tail_closed(self, event): """ Callback for when the transport tail has closed (ie no further input will be accepted by the transport). :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ self.on_transport_closed(event) def on_transport_closed(self, event): """ Callback for when the transport has closed - ie both the head (input) and tail (output) of the transport pipeline are closed. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if self.delegate is not None and event.connection and self.is_local_open(event.connection): _dispatch(self.delegate, 'on_disconnected', event)
[docs]class MessagingHandler(Handler, Acking): """ A general purpose handler that makes the proton-c events somewhat simpler to deal with and/or avoids repetitive tasks for common use cases. :param prefetch: Initial flow credit for receiving messages, defaults to 10. :type prefetch: ``int`` :param auto_accept: If ``True``, accept all messages (default). Otherwise messages must be individually accepted or rejected. :type auto_accept: ``bool`` :param auto_settle: If ``True``, settle all messages (default). Otherwise messages must be explicitly settled. :type auto_settle: ``bool`` :param peer_close_is_error: If ``True``, a peer endpoint closing will be treated as an error with an error callback. Otherwise (default), the normal callbacks for the closing will occur. :type peer_close_is_error: ``bool`` """ def __init__(self, prefetch=10, auto_accept=True, auto_settle=True, peer_close_is_error=False): self.handlers = [] if prefetch: self.handlers.append(FlowController(prefetch)) self.handlers.append(EndpointStateHandler(peer_close_is_error, weakref.proxy(self))) self.handlers.append(IncomingMessageHandler(auto_accept, weakref.proxy(self))) self.handlers.append(OutgoingMessageHandler(auto_settle, weakref.proxy(self))) self.fatal_conditions = ["amqp:unauthorized-access"]
[docs] def on_transport_error(self, event): """ Called when some error is encountered with the transport over which the AMQP connection is to be established. This includes authentication errors as well as socket errors. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if event.transport.condition: if event.transport.condition.info: log.error("%s: %s: %s" % ( event.transport.condition.name, event.transport.condition.description, event.transport.condition.info)) else: log.error("%s: %s" % (event.transport.condition.name, event.transport.condition.description)) if event.transport.condition.name in self.fatal_conditions: event.connection.close() else: logging.error("Unspecified transport error")
[docs] def on_connection_error(self, event): """ Called when the peer closes the connection with an error condition. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ EndpointStateHandler.print_error(event.connection, "connection")
[docs] def on_session_error(self, event): """ Called when the peer closes the session with an error condition. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ EndpointStateHandler.print_error(event.session, "session") event.connection.close()
[docs] def on_reactor_init(self, event): """ Called when the event loop - the reactor - starts. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ if hasattr(event.reactor, 'subclass'): setattr(event, event.reactor.subclass.__name__.lower(), event.reactor) self.on_start(event)
[docs] def on_start(self, event): """ Called when the event loop starts. (Just an alias for on_reactor_init) :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_connection_closed(self, event): """ Called when the connection is closed. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_session_closed(self, event): """ Called when the session is closed. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_connection_closing(self, event): """ Called when the peer initiates the closing of the connection. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_session_closing(self, event): """ Called when the peer initiates the closing of the session. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_disconnected(self, event): """ Called when the socket is disconnected. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_sendable(self, event): """ Called when the sender link has credit and messages can therefore be transferred. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_accepted(self, event): """ Called when the remote peer accepts an outgoing message. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_rejected(self, event): """ Called when the remote peer rejects an outgoing message. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_released(self, event): """ Called when the remote peer releases an outgoing message. Note that this may be in response to either the RELEASE or MODIFIED state as defined by the AMQP specification. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_settled(self, event): """ Called when the remote peer has settled the outgoing message. This is the point at which it should never be retransmitted. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_message(self, event): """ Called when a message is received. The message itself can be obtained as a property on the event. For the purpose of referring to this message in further actions (e.g. if explicitly accepting it, the ``delivery`` should be used, also obtainable via a property on the event. :param event: The underlying event object. Use this to obtain further information on the event. In particular, the message itself may be obtained by accessing ``event.message``. :type event: :class:`proton.Event` """ pass
[docs]class TransactionHandler(object): """ The interface for transaction handlers - ie objects that want to be notified of state changes related to a transaction. """
[docs] def on_transaction_declared(self, event): """ Called when a local transaction is declared. :param event: The underlying event object. Use this to obtain further information on the event. In particular, the :class:`proton.reactor.Transaction` object may be obtained by accessing ``event.transaction``. :type event: :class:`proton.Event` """ pass
[docs] def on_transaction_committed(self, event): """ Called when a local transaction is discharged successfully (committed). :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_transaction_aborted(self, event): """ Called when a local transaction is discharged unsuccessfully (aborted). :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_transaction_declare_failed(self, event): """ Called when a local transaction declare fails. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs] def on_transaction_commit_failed(self, event): """ Called when the commit of a local transaction fails. :param event: The underlying event object. Use this to obtain further information on the event. :type event: :class:`proton.Event` """ pass
[docs]class TransactionalClientHandler(MessagingHandler, TransactionHandler): """ An extension to the MessagingHandler for applications using transactions. This handler provides all of the callbacks found in :class:`MessagingHandler` and :class:`TransactionHandler`, and provides a convenience method :meth:`accept` for performing a transactional acceptance of received messages. :param prefetch: Initial flow credit for receiving messages, defaults to 10. :type prefetch: ``int`` :param auto_accept: If ``True``, accept all messages (default). Otherwise messages must be individually accepted or rejected. :type auto_accept: ``bool`` :param auto_settle: If ``True``, settle all messages (default). Otherwise messages must be explicitly settled. :type auto_settle: ``bool`` :param peer_close_is_error: If ``True``, a peer endpoint closing will be treated as an error with an error callback. Otherwise (default), the normal callbacks for the closing will occur. :type peer_close_is_error: ``bool`` """ def __init__(self, prefetch=10, auto_accept=False, auto_settle=True, peer_close_is_error=False): super(TransactionalClientHandler, self).__init__(prefetch, auto_accept, auto_settle, peer_close_is_error)
[docs] def accept(self, delivery, transaction=None): """ A convenience method for accepting a received message as part of a transaction. If no transaction object is supplied, a regular non-transactional acceptance will be performed. :param delivery: Delivery tracking object for received message. :type delivery: :class:`proton.Delivery` :param transaction: Transaction tracking object which is required if the message is being accepted under the transaction. If ``None`` (default), then a normal non-transactional accept occurs. :type transaction: :class:`proton.reactor.Transaction` """ if transaction: transaction.accept(delivery) else: super(TransactionalClientHandler, self).accept(delivery)
class FlowController(Handler): def __init__(self, window=1024): self._window = window self._drained = 0 def on_link_local_open(self, event): self._flow(event.link) def on_link_remote_open(self, event): self._flow(event.link) def on_link_flow(self, event): self._flow(event.link) def on_delivery(self, event): self._flow(event.link) def _flow(self, link): if link.is_receiver: self._drained += link.drained() if self._drained == 0: delta = self._window - link.credit link.flow(delta) class Handshaker(Handler): @staticmethod def on_connection_remote_open(event): conn = event.connection if conn.state & Endpoint.LOCAL_UNINIT: conn.open() @staticmethod def on_session_remote_open(event): ssn = event.session if ssn.state & Endpoint.LOCAL_UNINIT: ssn.open() @staticmethod def on_link_remote_open(event): link = event.link if link.state & Endpoint.LOCAL_UNINIT: link.source.copy(link.remote_source) link.target.copy(link.remote_target) link.open() @staticmethod def on_connection_remote_close(event): conn = event.connection if not conn.state & Endpoint.LOCAL_CLOSED: conn.close() @staticmethod def on_session_remote_close(event): ssn = event.session if not ssn.state & Endpoint.LOCAL_CLOSED: ssn.close() @staticmethod def on_link_remote_close(event): link = event.link if not link.state & Endpoint.LOCAL_CLOSED: link.close() # Back compatibility definitions CFlowController = FlowController CHandshaker = Handshaker class PythonIO: def __init__(self): self.selectables = [] self.delegate = IOHandler() def on_unhandled(self, method, event): event.dispatch(self.delegate) def on_selectable_init(self, event): self.selectables.append(event.context) def on_selectable_updated(self, event): pass def on_selectable_final(self, event): sel = event.context if sel.is_terminal: self.selectables.remove(sel) sel.close() def on_reactor_quiesced(self, event): reactor = event.reactor # check if we are still quiesced, other handlers of # on_reactor_quiesced could have produced events to process if not reactor.quiesced: return reading = [] writing = [] deadline = None for sel in self.selectables: if sel.reading: reading.append(sel) if sel.writing: writing.append(sel) if sel.deadline: if deadline is None: deadline = sel.deadline else: deadline = min(sel.deadline, deadline) if deadline is not None: timeout = deadline - time.time() else: timeout = reactor.timeout if timeout < 0: timeout = 0 timeout = min(timeout, reactor.timeout) readable, writable, _ = IO.select(reading, writing, [], timeout) now = reactor.mark() for s in readable: s.readable() for s in writable: s.writable() for s in self.selectables: if s.deadline and now > s.deadline: s.expired() reactor.yield_() # For C style IO handler need to implement Selector class IOHandler(Handler): def __init__(self): self._selector = IO.Selector() def on_selectable_init(self, event): s = event.selectable self._selector.add(s) s._reactor._selectables += 1 def on_selectable_updated(self, event): s = event.selectable self._selector.update(s) def on_selectable_final(self, event): s = event.selectable self._selector.remove(s) s._reactor._selectables -= 1 s.close() def on_reactor_quiesced(self, event): r = event.reactor if not r.quiesced: return d = r.timer_deadline readable, writable, expired = self._selector.select(r.timeout) now = r.mark() for s in readable: s.readable() for s in writable: s.writable() for s in expired: s.expired() r.yield_() def on_selectable_readable(self, event): s = event.selectable t = s._transport # If we're an acceptor we can't have a transport # and we don't want to do anything here in any case if not t: return capacity = t.capacity() if capacity > 0: try: b = s.recv(capacity) if len(b) > 0: n = t.push(b) else: # EOF handling self.on_selectable_error(event) except socket.error as e: # TODO: What's the error handling to be here? log.error("Couldn't recv: %r" % e) t.close_tail() # Always update as we may have gone to not reading or from # not writing to writing when processing the incoming bytes r = s._reactor self.update(t, s, r.now) def on_selectable_writable(self, event): s = event.selectable t = s._transport # If we're an acceptor we can't have a transport # and we don't want to do anything here in any case if not t: return pending = t.pending() if pending > 0: try: n = s.send(t.peek(pending)) t.pop(n) except socket.error as e: log.error("Couldn't send: %r" % e) # TODO: Error? or actually an exception t.close_head() newpending = t.pending() if newpending != pending: r = s._reactor self.update(t, s, r.now) def on_selectable_error(self, event): s = event.selectable t = s._transport t.close_head() t.close_tail() s.terminate() s._transport = None t._selectable = None s.update() def on_selectable_expired(self, event): s = event.selectable t = s._transport r = s._reactor self.update(t, s, r.now) def on_connection_local_open(self, event): c = event.connection if not c.state & Endpoint.REMOTE_UNINIT: return t = Transport() # It seems perverse, but the C code ignores bind errors too! # and this is required or you get errors because Connector() has already # bound the transport and connection! t.bind_nothrow(c) def on_connection_bound(self, event): c = event.connection t = event.transport reactor = c._reactor # link the new transport to its reactor: t._reactor = reactor if c._acceptor: # this connection was created by the acceptor. There is already a # socket assigned to this connection. Nothing needs to be done. return url = c.url or Url(c.hostname) url.defaults() host = url.host port = int(url.port) if not c.user: user = url.username if user: c.user = user password = url.password if password: c.password = password addrs = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM) # Try first possible address log.debug("Connect trying first transport address: %s", addrs[0]) sock = IO.connect(addrs[0]) # At this point we need to arrange to be called back when the socket is writable ConnectSelectable(sock, reactor, addrs[1:], t, self) # TODO: Don't understand why we need this now - how can we get PN_TRANSPORT until the connection succeeds? t._selectable = None @staticmethod def update(transport, selectable, now): try: capacity = transport.capacity() selectable.reading = capacity > 0 except: if transport.closed: selectable.terminate() selectable._transport = None transport._selectable = None try: pending = transport.pending() selectable.writing = pending > 0 except: if transport.closed: selectable.terminate() selectable._transport = None transport._selectable = None selectable.deadline = transport.tick(now) selectable.update() def on_transport(self, event): t = event.transport r = t._reactor s = t._selectable if s and not s.is_terminal: self.update(t, s, r.now) def on_transport_closed(self, event): t = event.transport r = t._reactor s = t._selectable if s and not s.is_terminal: s.terminate() s._transport = None t._selectable = None r.update(s) t.unbind() class ConnectSelectable(Selectable): def __init__(self, sock, reactor, addrs, transport, iohandler): super(ConnectSelectable, self).__init__(sock, reactor) self.writing = True self._addrs = addrs self._transport = transport self._iohandler = iohandler transport._connect_selectable = self def readable(self): pass def writable(self): e = self._delegate.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) t = self._transport t._connect_selectable = None # Always cleanup this ConnectSelectable: either we failed or created a new one # Do it first to ensure the socket gets deregistered before being registered again # in the case of connecting self.terminate() self._transport = None self.update() if e == 0: log.debug("Connection succeeded") # Disassociate from the socket (which will be passed on) self.release() s = self._reactor.selectable(delegate=self._delegate) s._transport = t t._selectable = s self._iohandler.update(t, s, t._reactor.now) return elif e == errno.ECONNREFUSED: if len(self._addrs) > 0: log.debug("Connection refused: trying next transport address: %s", self._addrs[0]) sock = IO.connect(self._addrs[0]) # New ConnectSelectable for the new socket with rest of addresses ConnectSelectable(sock, self._reactor, self._addrs[1:], t, self._iohandler) return else: log.debug("Connection refused, but tried all transport addresses") t.condition = Condition("proton.pythonio", "Connection refused to all addresses") else: log.error("Couldn't connect: %s", e) t.condition = Condition("proton.pythonio", "Connection error: %s" % e) t.close_tail() t.close_head()