"""

Sending/Receiving Messages.

"""
from itertools import count
from carrot.utils import gen_unique_id
import warnings

from carrot import serialization


class Consumer(object):
    """Message consumer.

    :param connection: see :attr:`connection`.
    :param queue: see :attr:`queue`.
    :param exchange: see :attr:`exchange`.
    :param routing_key: see :attr:`routing_key`.

    :keyword durable: see :attr:`durable`.
    :keyword auto_delete: see :attr:`auto_delete`.
    :keyword exclusive: see :attr:`exclusive`.
    :keyword exchange_type: see :attr:`exchange_type`.
    :keyword auto_ack: see :attr:`auto_ack`.
    :keyword no_ack: see :attr:`no_ack`.
    :keyword auto_declare: see :attr:`auto_declare`.


    .. attribute:: connection

        The connection to the broker.
        A :class:`carrot.connection.BrokerConnection` instance.

    .. attribute:: queue

       Name of the queue.

    .. attribute:: exchange

        Name of the exchange the queue binds to.

    .. attribute:: routing_key

        The routing key (if any). The interpretation of the routing key
        depends on the value of the :attr:`exchange_type` attribute:

            * direct exchange

                Matches if the routing key property of the message and
                the :attr:`routing_key` attribute are identical.

            * fanout exchange

                Always matches, even if the binding does not have a key.

            * topic exchange

                Matches the routing key property of the message by a primitive
                pattern matching scheme. The message routing key then consists
                of words separated by dots (``"."``, like domain names), and
                two special characters are available; star (``"*"``) and hash
                (``"#"``). The star matches any word, and the hash matches
                zero or more words. For example ``"*.stock.#"`` matches the
                routing keys ``"usd.stock"`` and ``"eur.stock.db"`` but not
                ``"stock.nasdaq"``.

    .. attribute:: durable

        Durable exchanges remain active when a server restarts. Non-durable
        exchanges (transient exchanges) are purged when a server restarts.
        Default is ``True``.

    .. attribute:: auto_delete

        If set, the exchange is deleted when all queues have finished
        using it. Default is ``False``.

    .. attribute:: exclusive

        Exclusive queues may only be consumed from by the current connection.
        When :attr:`exclusive` is on, this also implies :attr:`auto_delete`.
        Default is ``False``.

    .. attribute:: exchange_type

        AMQP defines four default exchange types (routing algorithms) that
        covers most of the common messaging use cases. An AMQP broker can
        also define additional exchange types, so see your message brokers
        manual for more information about available exchange types.

            * Direct

                Direct match between the routing key in the message, and the
                routing criteria used when a queue is bound to this exchange.

            * Topic

                Wildcard match between the routing key and the routing pattern
                specified in the binding. The routing key is treated as zero
                or more words delimited by ``"."`` and supports special
                wildcard characters. ``"*"`` matches a single word and ``"#"``
                matches zero or more words.

            * Fanout

                Queues are bound to this exchange with no arguments. Hence any
                message sent to this exchange will be forwarded to all queues
                bound to this exchange.

            * Headers

                Queues are bound to this exchange with a table of arguments
                containing headers and values (optional). A special argument
                named "x-match" determines the matching algorithm, where
                ``"all"`` implies an ``AND`` (all pairs must match) and
                ``"any"`` implies ``OR`` (at least one pair must match).

                Use the :attr:`routing_key`` is used to specify the arguments,
                the same when sending messages.

            This description of AMQP exchange types was shamelessly stolen
            from the blog post `AMQP in 10 minutes: Part 4`_ by
            Rajith Attapattu. Recommended reading.

            .. _`AMQP in 10 minutes: Part 4`:
                http://bit.ly/amqp-exchange-types

    .. attribute:: callbacks

        List of registered callbacks to trigger when a message is received
        by :meth:`wait`, :meth:`process_next` or :meth:`iterqueue`.

    .. attribute:: warn_if_exists

        Emit a warning if the queue has already been declared. If a queue
        already exists, and you try to redeclare the queue with new settings,
        the new settings will be silently ignored, so this can be
        useful if you've recently changed the :attr:`routing_key` attribute
        or other settings.

    .. attribute:: auto_ack

        Acknowledgement is handled automatically once messages are received.
        This means that the :meth:`carrot.backends.base.BaseMessage.ack` and
        :meth:`carrot.backends.base.BaseMessage.reject` methods
        on the message object are no longer valid.
        By default :attr:`auto_ack` is set to ``False``, and the receiver is
        required to manually handle acknowledgment.

    .. attribute:: no_ack

        Disable acknowledgement on the server-side. This is different from
        :attr:`auto_ack` in that acknowledgement is turned off altogether.
        This functionality increases performance but at the cost of
        reliability. Messages can get lost if a client dies before it can
        deliver them to the application.

    .. attribute auto_declare

        If this is ``True`` the following will be automatically declared:

            * The queue if :attr:`queue` is set.
            * The exchange if :attr:`exchange` is set.
            * The :attr:`queue` will be bound to the :attr:`exchange`.

        This is the default behaviour.


    :raises `amqplib.client_0_8.channel.AMQPChannelException`: if the queue is
        exclusive and the queue already exists and is owned by another
        connection.


    Example Usage

        >>> consumer = Consumer(connection=DjangoBrokerConnection(),
        ...               queue="foo", exchange="foo", routing_key="foo")
        >>> def process_message(message_data, message):
        ...     print("Got message %s: %s" % (
        ...             message.delivery_tag, message_data))
        >>> consumer.register_callback(process_message)
        >>> consumer.wait() # Go into receive loop

    """
    queue = ""
    exchange = ""
    routing_key = ""
    durable = True
    exclusive = False
    auto_delete = False
    exchange_type = "direct"
    channel_open = False
    warn_if_exists = False
    auto_declare = True
    auto_ack = False
    queue_arguments = None
    no_ack = False
    _closed = True
    _init_opts = ("durable", "exclusive", "auto_delete",
                  "exchange_type", "warn_if_exists",
                  "auto_ack", "auto_declare",
                  "queue_arguments")
    _next_consumer_tag = count(1).next

    def __init__(self, connection, queue=None, exchange=None,
            routing_key=None, **kwargs):
        self.connection = connection
        self.backend = kwargs.get("backend", None)
        if not self.backend:
            self.backend = self.connection.create_backend()
        self.queue = queue or self.queue

        # Binding.
        self.queue = queue or self.queue
        self.exchange = exchange or self.exchange
        self.routing_key = routing_key or self.routing_key
        self.callbacks = []

        # Options
        for opt_name in self._init_opts:
            opt_value = kwargs.get(opt_name)
            if opt_value is not None:
                setattr(self, opt_name, opt_value)

        # exclusive implies auto-delete.
        if self.exclusive:
            self.auto_delete = True

        self.consumer_tag = self._generate_consumer_tag()

        if self.auto_declare:
            self.declare()

    def __enter__(self):
        return self

    def __exit__(self, e_type, e_value, e_trace):
        if e_type:
            raise e_type(e_value)
        self.close()

    def __iter__(self):
        """iter(Consumer) -> Consumer.iterqueue(infinite=True)"""
        return self.iterqueue(infinite=True)

    def _generate_consumer_tag(self):
        """Generate a unique consumer tag.

        :rtype string:

        """
        return "%s.%s%s" % (
                self.__class__.__module__,
                self.__class__.__name__,
                self._next_consumer_tag())

    def declare(self):
        """Declares the queue, the exchange and binds the queue to
        the exchange."""
        arguments = None
        routing_key = self.routing_key
        if self.exchange_type == "headers":
            arguments, routing_key = routing_key, ""

        if self.queue:
            self.backend.queue_declare(queue=self.queue, durable=self.durable,
                                       exclusive=self.exclusive,
                                       auto_delete=self.auto_delete,
                                       arguments=self.queue_arguments,
                                       warn_if_exists=self.warn_if_exists)
        if self.exchange:
            self.backend.exchange_declare(exchange=self.exchange,
                                          type=self.exchange_type,
                                          durable=self.durable,
                                          auto_delete=self.auto_delete)
        if self.queue:
            self.backend.queue_bind(queue=self.queue,
                                    exchange=self.exchange,
                                    routing_key=routing_key,
                                    arguments=arguments)
        self._closed = False
        return self

    def _receive_callback(self, raw_message):
        """Internal method used when a message is received in consume mode."""
        message = self.backend.message_to_python(raw_message)

        if self.auto_ack and not message.acknowledged:
            message.ack()
        self.receive(message.payload, message)

    def fetch(self, no_ack=None, auto_ack=None, enable_callbacks=False):
        """Receive the next message waiting on the queue.

        :returns: A :class:`carrot.backends.base.BaseMessage` instance,
            or ``None`` if there's no messages to be received.

        :keyword enable_callbacks: Enable callbacks. The message will be
            processed with all registered callbacks. Default is disabled.
        :keyword auto_ack: Override the default :attr:`auto_ack` setting.
        :keyword no_ack: Override the default :attr:`no_ack` setting.

        """
        no_ack = no_ack or self.no_ack
        auto_ack = auto_ack or self.auto_ack
        message = self.backend.get(self.queue, no_ack=no_ack)
        if message:
            if auto_ack and not message.acknowledged:
                message.ack()
            if enable_callbacks:
                self.receive(message.payload, message)
        return message

    def process_next(self):
        """**DEPRECATED** Use :meth:`fetch` like this instead:

            >>> message = self.fetch(enable_callbacks=True)

        """
        warnings.warn(DeprecationWarning(
            "Consumer.process_next has been deprecated in favor of \
            Consumer.fetch(enable_callbacks=True)"))
        return self.fetch(enable_callbacks=True)

    def receive(self, message_data, message):
        """This method is called when a new message is received by
        running :meth:`wait`, :meth:`process_next` or :meth:`iterqueue`.

        When a message is received, it passes the message on to the
        callbacks listed in the :attr:`callbacks` attribute.
        You can register callbacks using :meth:`register_callback`.

        :param message_data: The deserialized message data.

        :param message: The :class:`carrot.backends.base.BaseMessage` instance.

        :raises NotImplementedError: If no callbacks has been registered.

        """
        if not self.callbacks:
            raise NotImplementedError("No consumer callbacks registered")
        for callback in self.callbacks:
            callback(message_data, message)

    def register_callback(self, callback):
        """Register a callback function to be triggered by :meth:`receive`.

        The ``callback`` function must take two arguments:

            * message_data

                The deserialized message data

            * message

                The :class:`carrot.backends.base.BaseMessage` instance.
        """
        self.callbacks.append(callback)

    def discard_all(self, filterfunc=None):
        """Discard all waiting messages.

        :param filterfunc: A filter function to only discard the messages this
            filter returns.

        :returns: the number of messages discarded.

        *WARNING*: All incoming messages will be ignored and not processed.

        Example using filter:

            >>> def waiting_feeds_only(message):
            ...     try:
            ...         message_data = message.decode()
            ...     except: # Should probably be more specific.
            ...         pass
            ...
            ...     if message_data.get("type") == "feed":
            ...         return True
            ...     else:
            ...         return False
        """
        if not filterfunc:
            return self.backend.queue_purge(self.queue)

        if self.no_ack or self.auto_ack:
            raise Exception("discard_all: Can't use filter with auto/no-ack.")

        discarded_count = 0
        while True:
            message = self.fetch()
            if message is None:
                return discarded_count

            if filterfunc(message):
                message.ack()
                discarded_count += 1

    def iterconsume(self, limit=None, no_ack=None):
        """Iterator processing new messages as they arrive.
        Every new message will be passed to the callbacks, and the iterator
        returns ``True``. The iterator is infinite unless the ``limit``
        argument is specified or someone closes the consumer.

        :meth:`iterconsume` uses transient requests for messages on the
        server, while :meth:`iterequeue` uses synchronous access. In most
        cases you want :meth:`iterconsume`, but if your environment does not
        support this behaviour you can resort to using :meth:`iterqueue`
        instead.

        Also, :meth:`iterconsume` does not return the message
        at each step, something which :meth:`iterqueue` does.

        :keyword limit: Maximum number of messages to process.

        :raises StopIteration: if limit is set and the message limit has been
            reached.

        """
        self.consume(no_ack=no_ack)
        return self.backend.consume(limit=limit)

    def consume(self, no_ack=None):
        """Declare consumer."""
        no_ack = no_ack or self.no_ack
        self.backend.declare_consumer(queue=self.queue, no_ack=no_ack,
                                      callback=self._receive_callback,
                                      consumer_tag=self.consumer_tag,
                                      nowait=True)
        self.channel_open = True

    def wait(self, limit=None):
        """Go into consume mode.

        Mostly for testing purposes and simple programs, you probably
        want :meth:`iterconsume` or :meth:`iterqueue` instead.

        This runs an infinite loop, processing all incoming messages
        using :meth:`receive` to apply the message to all registered
        callbacks.

        """
        it = self.iterconsume(limit)
        while True:
            it.next()

    def iterqueue(self, limit=None, infinite=False):
        """Infinite iterator yielding pending messages, by using
        synchronous direct access to the queue (``basic_get``).

        :meth:`iterqueue` is used where synchronous functionality is more
        important than performance. If you can, use :meth:`iterconsume`
        instead.

        :keyword limit: If set, the iterator stops when it has processed
            this number of messages in total.

        :keyword infinite: Don't raise :exc:`StopIteration` if there is no
            messages waiting, but return ``None`` instead. If infinite you
            obviously shouldn't consume the whole iterator at once without
            using a ``limit``.

        :raises StopIteration: If there is no messages waiting, and the
            iterator is not infinite.

        """
        for items_since_start in count():
            item = self.fetch()
            if (not infinite and item is None) or \
                    (limit and items_since_start >= limit):
                raise StopIteration
            yield item

    def cancel(self):
        """Cancel a running :meth:`iterconsume` session."""
        if self.channel_open:
            try:
                self.backend.cancel(self.consumer_tag)
            except KeyError:
                pass

    def close(self):
        """Close the channel to the queue."""
        self.cancel()
        self.backend.close()
        self._closed = True

    def flow(self, active):
        """This method asks the peer to pause or restart the flow of
        content data.

        This is a simple flow-control mechanism that a
        peer can use to avoid oveflowing its queues or otherwise
        finding itself receiving more messages than it can process.
        Note that this method is not intended for window control.  The
        peer that receives a request to stop sending content should
        finish sending the current content, if any, and then wait
        until it receives the ``flow(active=True)`` restart method.

        """
        self.backend.flow(active)

    def qos(self, prefetch_size=0, prefetch_count=0, apply_global=False):
        """Request specific Quality of Service.

        This method requests a specific quality of service.  The QoS
        can be specified for the current channel or for all channels
        on the connection.  The particular properties and semantics of
        a qos method always depend on the content class semantics.
        Though the qos method could in principle apply to both peers,
        it is currently meaningful only for the server.

        :param prefetch_size: Prefetch window in octets.
            The client can request that messages be sent in
            advance so that when the client finishes processing a
            message, the following message is already held
            locally, rather than needing to be sent down the
            channel.  Prefetching gives a performance improvement.
            This field specifies the prefetch window size in
            octets.  The server will send a message in advance if
            it is equal to or smaller in size than the available
            prefetch size (and also falls into other prefetch
            limits). May be set to zero, meaning "no specific
            limit", although other prefetch limits may still
            apply. The ``prefetch_size`` is ignored if the
            :attr:`no_ack` option is set.

        :param prefetch_count: Specifies a prefetch window in terms of whole
            messages. This field may be used in combination with
            ``prefetch_size``; A message will only be sent
            in advance if both prefetch windows (and those at the
            channel and connection level) allow it. The prefetch-
            count is ignored if the :attr:`no_ack` option is set.

        :keyword apply_global: By default the QoS settings apply to the
            current channel only. If this is set, they are applied
            to the entire connection.

        """
        return self.backend.qos(prefetch_size, prefetch_count, apply_global)


class Publisher(object):
    """Message publisher.

    :param connection: see :attr:`connection`.
    :param exchange: see :attr:`exchange`.
    :param routing_key: see :attr:`routing_key`.

    :keyword exchange_type: see :attr:`Consumer.exchange_type`.
    :keyword durable: see :attr:`Consumer.durable`.
    :keyword auto_delete: see :attr:`Consumer.auto_delete`.
    :keyword serializer: see :attr:`serializer`.
    :keyword auto_declare: See :attr:`auto_declare`.


    .. attribute:: connection

        The connection to the broker.
        A :class:`carrot.connection.BrokerConnection` instance.

    .. attribute:: exchange

        Name of the exchange we send messages to.

    .. attribute:: routing_key

        The default routing key for messages sent using this publisher.
        See :attr:`Consumer.routing_key` for more information.
        You can override the routing key by passing an explicit
        ``routing_key`` argument to :meth:`send`.

    .. attribute:: delivery_mode

        The default delivery mode used for messages. The value is an integer.
        The following delivery modes are supported by (at least) RabbitMQ:

            * 1 or "transient"

                The message is transient. Which means it is stored in
                memory only, and is lost if the server dies or restarts.

            * 2 or "persistent"
                The message is persistent. Which means the message is
                stored both in-memory, and on disk, and therefore
                preserved if the server dies or restarts.

        The default value is ``2`` (persistent).

    .. attribute:: exchange_type

        See :attr:`Consumer.exchange_type`.

    .. attribute:: durable

        See :attr:`Consumer.durable`.

    .. attribute:: auto_delete

        See :attr:`Consumer.auto_delete`.

    .. attribute:: auto_declare

        If this is ``True`` and the :attr:`exchange` name is set, the exchange
        will be automatically declared at instantiation.
        You can manually the declare the exchange by using the :meth:`declare`
        method.

        Auto declare is on by default.

    .. attribute:: serializer

        A string identifying the default serialization method to use.
        Defaults to ``json``. Can be ``json`` (default), ``raw``,
        ``pickle``, ``hessian``, ``yaml``, or any custom serialization
        methods that have been registered with
        :mod:`carrot.serialization.registry`.

    """

    NONE_PERSISTENT_DELIVERY_MODE = 1
    TRANSIENT_DELIVERY_MODE = 1
    PERSISTENT_DELIVERY_MODE = 2
    DELIVERY_MODES = {
            "transient": TRANSIENT_DELIVERY_MODE,
            "persistent": PERSISTENT_DELIVERY_MODE,
            "non-persistent": TRANSIENT_DELIVERY_MODE,
    }

    exchange = ""
    routing_key = ""
    delivery_mode = PERSISTENT_DELIVERY_MODE
    _closed = True
    exchange_type = "direct"
    durable = True
    auto_delete = False
    auto_declare = True
    serializer = None
    _init_opts = ("exchange_type", "durable", "auto_delete",
                  "serializer", "delivery_mode", "auto_declare")

    def __init__(self, connection, exchange=None, routing_key=None, **kwargs):
        self.connection = connection
        self.backend = self.connection.create_backend()
        self.exchange = exchange or self.exchange
        self.routing_key = routing_key or self.routing_key
        for opt_name in self._init_opts:
            opt_value = kwargs.get(opt_name)
            if opt_value is not None:
                setattr(self, opt_name, opt_value)
        self.delivery_mode = self.DELIVERY_MODES.get(self.delivery_mode,
                                                     self.delivery_mode)
        self._closed = False

        if self.auto_declare and self.exchange:
            self.declare()


    def declare(self):
        """Declare the exchange.

        Creates the exchange on the broker.

        """
        self.backend.exchange_declare(exchange=self.exchange,
                                      type=self.exchange_type,
                                      durable=self.durable,
                                      auto_delete=self.auto_delete)

    def __enter__(self):
        return self

    def __exit__(self, e_type, e_value, e_trace):
        self.close()

    def create_message(self, message_data, delivery_mode=None, priority=None,
                       content_type=None, content_encoding=None,
                       serializer=None):
        """With any data, serialize it and encapsulate it in a AMQP
        message with the proper headers set."""

        delivery_mode = delivery_mode or self.delivery_mode

        # No content_type? Then we're serializing the data internally.
        if not content_type:
            serializer = serializer or self.serializer
            (content_type, content_encoding,
             message_data) = serialization.encode(message_data,
                                                  serializer=serializer)
        else:
            # If the programmer doesn't want us to serialize,
            # make sure content_encoding is set.
            if isinstance(message_data, unicode):
                if not content_encoding:
                    content_encoding = 'utf-8'
                message_data = message_data.encode(content_encoding)

            # If they passed in a string, we can't know anything
            # about it.  So assume it's binary data.
            elif not content_encoding:
                content_encoding = 'binary'

        return self.backend.prepare_message(message_data, delivery_mode,
                                            priority=priority,
                                            content_type=content_type,
                                            content_encoding=content_encoding)

    def send(self, message_data, routing_key=None, delivery_mode=None,
            mandatory=False, immediate=False, priority=0, content_type=None,
            content_encoding=None, serializer=None, exchange=None):
        """Send a message.

        :param message_data: The message data to send. Can be a list,
            dictionary or a string.

        :keyword routing_key: A custom routing key for the message.
            If not set, the default routing key set in the :attr:`routing_key`
            attribute is used.

        :keyword mandatory: If set, the message has mandatory routing.
            By default the message is silently dropped by the server if it
            can't be routed to a queue. However - If the message is mandatory,
            an exception will be raised instead.

        :keyword immediate: Request immediate delivery.
            If the message cannot be routed to a queue consumer immediately,
            an exception will be raised. This is instead of the default
            behaviour, where the server will accept and queue the message,
            but with no guarantee that the message will ever be consumed.

        :keyword delivery_mode: Override the default :attr:`delivery_mode`.

        :keyword priority: The message priority, ``0`` to ``9``.

        :keyword content_type: The messages content_type. If content_type
            is set, no serialization occurs as it is assumed this is either
            a binary object, or you've done your own serialization.
            Leave blank if using built-in serialization as our library
            properly sets content_type.

        :keyword content_encoding: The character set in which this object
            is encoded. Use "binary" if sending in raw binary objects.
            Leave blank if using built-in serialization as our library
            properly sets content_encoding.

        :keyword serializer: Override the default :attr:`serializer`.

        :keyword exchange: Override the exchange to publish to.
            Note that this exchange must have been declared.

        """
        headers = None
        routing_key = routing_key or self.routing_key

        if self.exchange_type == "headers":
            headers, routing_key = routing_key, ""

        exchange = exchange or self.exchange

        message = self.create_message(message_data, priority=priority,
                                      delivery_mode=delivery_mode,
                                      content_type=content_type,
                                      content_encoding=content_encoding,
                                      serializer=serializer)
        self.backend.publish(message,
                             exchange=exchange, routing_key=routing_key,
                             mandatory=mandatory, immediate=immediate,
                             headers=headers)

    def close(self):
        """Close connection to queue."""
        self.backend.close()
        self._closed = True


class Messaging(object):
    """A combined message publisher and consumer."""
    queue = ""
    exchange = ""
    routing_key = ""
    publisher_cls = Publisher
    consumer_cls = Consumer
    _closed = True

    def __init__(self, connection, **kwargs):
        self.connection = connection
        self.exchange = kwargs.get("exchange", self.exchange)
        self.queue = kwargs.get("queue", self.queue)
        self.routing_key = kwargs.get("routing_key", self.routing_key)
        self.publisher = self.publisher_cls(connection,
                exchange=self.exchange, routing_key=self.routing_key)
        self.consumer = self.consumer_cls(connection, queue=self.queue,
                exchange=self.exchange, routing_key=self.routing_key)
        self.consumer.register_callback(self.receive)
        self.callbacks = []
        self._closed = False

    def __enter__(self):
        return self

    def __exit__(self, e_type, e_value, e_trace):
        if e_type:
            raise e_type(e_value)
        self.close()

    def register_callback(self, callback):
        """See :meth:`Consumer.register_callback`"""
        self.callbacks.append(callback)

    def receive(self, message_data, message):
        """See :meth:`Consumer.receive`"""
        if not self.callbacks:
            raise NotImplementedError("No consumer callbacks registered")
        for callback in self.callbacks:
            callback(message_data, message)

    def send(self, message_data, delivery_mode=None):
        """See :meth:`Publisher.send`"""
        self.publisher.send(message_data, delivery_mode=delivery_mode)

    def fetch(self, **kwargs):
        """See :meth:`Consumer.fetch`"""
        return self.consumer.fetch(**kwargs)

    def close(self):
        """Close any open channels."""
        self.consumer.close()
        self.publisher.close()
        self._closed = True


class ConsumerSet(object):
    """Receive messages from multiple consumers.

    :param connection: see :attr:`connection`.
    :param from_dict: see :attr:`from_dict`.
    :param consumers: see :attr:`consumers`.
    :param callbacks: see :attr:`callbacks`.

    .. attribute:: connection

        The connection to the broker.
        A :class:`carrot.connection.BrokerConnection` instance.

    .. attribute:: callbacks

        A list of callbacks to be called when a message is received.
        See :class:`Consumer.register_callback`.

    .. attribute:: from_dict

        Add consumers from a dictionary configuration::

            {
                "webshot": {
                            "exchange": "link_exchange",
                            "exchange_type": "topic",
                            "binding_key": "links.webshot",
                            "default_routing_key": "links.webshot",
                    },
                "retrieve": {
                            "exchange": "link_exchange",
                            "exchange_type" = "topic",
                            "binding_key": "links.*",
                            "default_routing_key": "links.retrieve",
                            "auto_delete": True,
                            # ...
                    },
            }

    .. attribute:: consumers

        Add consumers from a list of :class:`Consumer` instances.

    .. attribute:: auto_ack

        Default value for the :attr:`Consumer.auto_ack` attribute.

    """
    auto_ack = False

    def __init__(self, connection, from_dict=None, consumers=None,
            callbacks=None, **options):
        self.connection = connection
        self.options = options
        self.from_dict = from_dict or {}
        self.consumers = []
        self.callbacks = callbacks or []
        self._open_consumers = {}

        self.backend = self.connection.create_backend()

        self.auto_ack = options.get("auto_ack", self.auto_ack)

        if consumers:
            [self.add_consumer(consumer) for consumer in consumers]

        [self.add_consumer_from_dict(queue_name, **queue_options)
                for queue_name, queue_options in self.from_dict.items()]

    def _receive_callback(self, raw_message):
        """Internal method used when a message is received in consume mode."""
        message = self.backend.message_to_python(raw_message)
        if self.auto_ack and not message.acknowledged:
            message.ack()
        self.receive(message.decode(), message)

    def add_consumer_from_dict(self, queue, **options):
        """Add another consumer from dictionary configuration."""
        consumer = Consumer(self.connection, queue=queue,
                backend=self.backend, **options)
        self.consumers.append(consumer)
        return consumer

    def add_consumer(self, consumer):
        """Add another consumer from a :class:`Consumer` instance."""
        consumer.backend = self.backend
        self.consumers.append(consumer)

    def register_callback(self, callback):
        """Register new callback to be called when a message is received.
        See :meth:`Consumer.register_callback`"""
        self.callbacks.append(callback)

    def receive(self, message_data, message):
        """What to do when a message is received.
        See :meth:`Consumer.receive`."""
        if not self.callbacks:
            raise NotImplementedError("No consumer callbacks registered")
        for callback in self.callbacks:
            callback(message_data, message)

    def _declare_consumer(self, consumer, nowait=False):
        """Declare consumer so messages can be received from it using
        :meth:`iterconsume`."""
        if consumer.queue not in self._open_consumers:
            # Use the ConsumerSet's consumer by default, but if the
            # child consumer has a callback, honor it.
            callback = consumer.callbacks and \
                consumer._receive_callback or self._receive_callback
            self.backend.declare_consumer(queue=consumer.queue,
                                          no_ack=consumer.no_ack,
                                          nowait=nowait,
                                          callback=callback,
                                          consumer_tag=consumer.consumer_tag)
            self._open_consumers[consumer.queue] = consumer.consumer_tag

    def consume(self):
        """Declare consumers."""
        head = self.consumers[:-1]
        tail = self.consumers[-1]
        [self._declare_consumer(consumer, nowait=True)
                for consumer in head]
        self._declare_consumer(tail, nowait=False)

    def iterconsume(self, limit=None):
        """Cycle between all consumers in consume mode.

        See :meth:`Consumer.iterconsume`.
        """
        self.consume()
        return self.backend.consume(limit=limit)

    def discard_all(self):
        """Discard all messages. Does not support filtering.
        See :meth:`Consumer.discard_all`."""
        return sum([consumer.discard_all()
                        for consumer in self.consumers])

    def flow(self, active):
        """This method asks the peer to pause or restart the flow of
        content data.

        See :meth:`Consumer.flow`.

        """
        self.backend.flow(active)

    def qos(self, prefetch_size=0, prefetch_count=0, apply_global=False):
        """Request specific Quality of Service.

        See :meth:`Consumer.cos`.

        """
        self.backend.qos(prefetch_size, prefetch_count, apply_global)

    def cancel(self):
        """Cancel a running :meth:`iterconsume` session."""
        for consumer_tag in self._open_consumers.values():
            try:
                self.backend.cancel(consumer_tag)
            except KeyError:
                pass
        self._open_consumers.clear()

    def cancel_by_queue(self, queue):
        consumer_tag = self._open_consumers.pop(queue)
        self.backend.cancel(consumer_tag)

    def close(self):
        """Close all consumers."""
        self.cancel()
        for consumer in self.consumers:
            consumer.close()