This document is for py-amqp's development version, which can be significantly different from previous releases. Get the stable docs here: 5.0.

Source code for amqp.abstract_channel

"""Code common to Connection and Channel objects."""
# Copyright (C) 2007-2008 Barry Pederson <bp@barryp.org>)

import logging

from vine import ensure_promise, promise

from .exceptions import AMQPNotImplementedError, RecoverableConnectionError
from .serialization import dumps, loads

__all__ = ('AbstractChannel',)

AMQP_LOGGER = logging.getLogger('amqp')

IGNORED_METHOD_DURING_CHANNEL_CLOSE = """\
Received method %s during closing channel %s. This method will be ignored\
"""


class AbstractChannel:
    """Superclass for Connection and Channel.

    The connection is treated as channel 0, then comes
    user-created channel objects.

    The subclasses must have a _METHOD_MAP class property, mapping
    between AMQP method signatures and Python methods.
    """

    def __init__(self, connection, channel_id):
        self.is_closing = False
        self.connection = connection
        self.channel_id = channel_id
        connection.channels[channel_id] = self
        self.method_queue = []  # Higher level queue for methods
        self.auto_decode = False
        self._pending = {}
        self._callbacks = {}

        self._setup_listeners()

    __slots__ = (
        "is_closing",
        "connection",
        "channel_id",
        "method_queue",
        "auto_decode",
        "_pending",
        "_callbacks",
        # adding '__dict__' to get dynamic assignment
        "__dict__",
        "__weakref__",
        )

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        self.close()

    def send_method(self, sig,
                    format=None, args=None, content=None,
                    wait=None, callback=None, returns_tuple=False):
        p = promise()
        conn = self.connection
        if conn is None:
            raise RecoverableConnectionError('connection already closed')
        args = dumps(format, args) if format else ''
        try:
            conn.frame_writer(1, self.channel_id, sig, args, content)
        except StopIteration:
            raise RecoverableConnectionError('connection already closed')

        # TODO temp: callback should be after write_method ... ;)
        if callback:
            p.then(callback)
        p()
        if wait:
            return self.wait(wait, returns_tuple=returns_tuple)
        return p

    def close(self):
        """Close this Channel or Connection."""
        raise NotImplementedError('Must be overridden in subclass')

    def wait(self, method, callback=None, timeout=None, returns_tuple=False):
        p = ensure_promise(callback)
        pending = self._pending
        prev_p = []
        if not isinstance(method, list):
            method = [method]

        for m in method:
            prev_p.append(pending.get(m))
            pending[m] = p

        try:
            while not p.ready:
                self.connection.drain_events(timeout=timeout)

            if p.value:
                args, kwargs = p.value
                args = args[1:]  # We are not returning method back
                return args if returns_tuple else (args and args[0])
        finally:
            for i, m in enumerate(method):
                if prev_p[i] is not None:
                    pending[m] = prev_p[i]
                else:
                    pending.pop(m, None)

    def dispatch_method(self, method_sig, payload, content):
        if self.is_closing and method_sig not in (
            self._ALLOWED_METHODS_WHEN_CLOSING
        ):
            # When channel.close() was called we must ignore all methods except
            # Channel.close and Channel.CloseOk
            AMQP_LOGGER.warning(
                IGNORED_METHOD_DURING_CHANNEL_CLOSE,
                method_sig, self.channel_id
            )
            return

        if content and \
                self.auto_decode and \
                hasattr(content, 'content_encoding'):
            try:
                content.body = content.body.decode(content.content_encoding)
            except Exception:
                pass

        try:
            amqp_method = self._METHODS[method_sig]
        except KeyError:
            raise AMQPNotImplementedError(
                f'Unknown AMQP method {method_sig!r}')

        try:
            listeners = [self._callbacks[method_sig]]
        except KeyError:
            listeners = []
        one_shot = None
        try:
            one_shot = self._pending.pop(method_sig)
        except KeyError:
            if not listeners:
                return

        args = []
        if amqp_method.args:
            args, _ = loads(amqp_method.args, payload, 4)
        if amqp_method.content:
            args.append(content)

        for listener in listeners:
            listener(*args)

        if one_shot:
            one_shot(method_sig, *args)

    #: Placeholder, the concrete implementations will have to
    #: supply their own versions of _METHOD_MAP
    _METHODS = {}