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

# Mixin Classes - `kombu.mixins`

Mixins.

class kombu.mixins.ConsumerMixin

  Convenience mixin for implementing consumer programs.

  It can be used outside of threads, with threads, or greenthreads (eventlet/gevent) too.

  The basic class would need a `connection` attribute which must be a `Connection` instance, and define a `get_consumers()` method that returns a list of `kombu.Consumer` instances to use. Supporting multiple consumers is important so that multiple channels can be used for different QoS requirements.

  ## Example:

  >     class Worker(ConsumerMixin):
  >         task_queue = Queue('tasks', Exchange('tasks'), 'tasks')

  >         def __init__(self, connection):
  >             self.connection = None

  >         def get_consumers(self, Consumer, channel):
  >             return [Consumer(queues=[self.task_queue],
  >                              callbacks=[self.on_task])]

  >         def on_task(self, body, message):
  >             print('Got task: {0!r}'.format(body))
  >             message.ack()

  \* :meth:`extra_context`

    Optional extra context manager that will be entered after the connection and consumers have been set up.

    Takes arguments `(connection, channel)`.

  \* :meth:`on_connection_error`

    Handler called if the connection is lost/ or is unavailable.

    Takes arguments `(exc, interval)`, where interval is the time in seconds when the connection will be retried.

    The default handler will log the exception.

  \* :meth:`on_connection_revived`

    Handler called as soon as the connection is re-established after connection failure.

    Takes no arguments.

  \* :meth:`on_consume_ready`

    Handler called when the consumer is ready to accept messages.

    Takes arguments `(connection, channel, consumers)`. Also keyword arguments to `consume` are forwarded to this handler.

  \* :meth:`on_consume_end`

    Handler called after the consumers are canceled. Takes arguments `(connection, channel)`.

  \* :meth:`on_iteration`

    Handler called for every iteration while draining events.

    Takes no arguments.

  \* :meth:`on_decode_error`

    Handler called if a consumer was unable to decode the body of a message.

    Takes arguments `(message, exc)` where message is the original message object.

    The default handler will log the error and acknowledge the message, so if you override make sure to call super, or perform these steps yourself.

  Consumer()
  property channel_errors
  connect_max_retries = None

    maximum number of retries trying to re-establish the connection, if the connection is lost/unavailable.

  property connection_errors
  consume(limit=None, timeout=None, safety_interval=1, **kwargs)
  consumer_context(**kwargs)
  create_connection()
  establish_connection()
  extra_context(connection, channel)
  get_consumers(Consumer, channel)
  maybe_conn_error(fun)

    Use `kombu.common.ignore_errors()` instead.

  on_connection_error(exc, interval)
  on_connection_revived()
  on_consume_end(connection, channel)
  on_consume_ready(connection, channel, consumers, **kwargs)
  on_decode_error(message, exc)
  on_iteration()
  property restart_limit
  run(_tokens=1, **kwargs)
  should_stop = False

    When this is set to true the consumer should stop consuming and return, so that it can be joined if it is the implementation of a thread.

class kombu.mixins.ConsumerProducerMixin

  Consumer and Producer mixin.

  Version of ConsumerMixin having separate connection for also publishing messages.

  ## Example:

  >     class Worker(ConsumerProducerMixin):

  >         def __init__(self, connection):
  >             self.connection = connection

  >         def get_consumers(self, Consumer, channel):
  >             return [Consumer(queues=Queue('foo'),
  >                              on_message=self.handle_message,
  >                              accept='application/json',
  >                              prefetch_count=10)]

  >         def handle_message(self, message):
  >             self.producer.publish(
  >                 {'message': 'hello to you'},
  >                 exchange='',
  >                 routing_key=message.properties['reply_to'],
  >                 correlation_id=message.properties['correlation_id'],
  >                 retry=True,
  >             )

  on_consume_end(connection, channel)
  property producer
  property producer_connection