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

Source code for kombu.simple

"""Simple messaging interface."""

import socket
from collections import deque
from queue import Empty
from time import monotonic

from . import entity, messaging
from .connection import maybe_channel

__all__ = ('SimpleQueue', 'SimpleBuffer')


class SimpleBase:
    Empty = Empty
    _consuming = False

    def __enter__(self):
        return self

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

    def __init__(self, channel, producer, consumer, no_ack=False):
        self.channel = maybe_channel(channel)
        self.producer = producer
        self.consumer = consumer
        self.no_ack = no_ack
        self.queue = self.consumer.queues[0]
        self.buffer = deque()
        self.consumer.register_callback(self._receive)

    def get(self, block=True, timeout=None):
        if not block:
            return self.get_nowait()

        self._consume()

        time_start = monotonic()
        remaining = timeout
        while True:
            if self.buffer:
                return self.buffer.popleft()

            if remaining is not None and remaining <= 0.0:
                raise self.Empty()

            try:
                # The `drain_events` method will
                # block on the socket connection to rabbitmq. if any
                # application-level messages are received, it will put them
                # into `self.buffer`.
                # * The method will block for UP TO `timeout` milliseconds.
                # * The method may raise a socket.timeout exception; or...
                # * The method may return without having put anything on
                #    `self.buffer`.  This is because internal heartbeat
                #    messages are sent over the same socket; also POSIX makes
                #    no guarantees against socket calls returning early.
                self.channel.connection.client.drain_events(timeout=remaining)
            except socket.timeout:
                raise self.Empty()

            if remaining is not None:
                elapsed = monotonic() - time_start
                remaining = timeout - elapsed

    def get_nowait(self):
        m = self.queue.get(no_ack=self.no_ack, accept=self.consumer.accept)
        if not m:
            raise self.Empty()
        return m

    def put(self, message, serializer=None, headers=None, compression=None,
            routing_key=None, **kwargs):
        self.producer.publish(message,
                              serializer=serializer,
                              routing_key=routing_key,
                              headers=headers,
                              compression=compression,
                              **kwargs)

    def clear(self):
        return self.consumer.purge()

    def qsize(self):
        _, size, _ = self.queue.queue_declare(passive=True)
        return size

    def close(self):
        self.consumer.cancel()

    def _receive(self, message_data, message):
        self.buffer.append(message)

    def _consume(self):
        if not self._consuming:
            self.consumer.consume(no_ack=self.no_ack)
            self._consuming = True

    def __len__(self):
        """`len(self) -> self.qsize()`."""
        return self.qsize()

    def __bool__(self):
        return True
    __nonzero__ = __bool__


[docs]class SimpleQueue(SimpleBase): """Simple API for persistent queues.""" no_ack = False queue_opts = {} queue_args = {} exchange_opts = {'type': 'direct'} def __init__(self, channel, name, no_ack=None, queue_opts=None, queue_args=None, exchange_opts=None, serializer=None, compression=None, accept=None): queue = name queue_opts = dict(self.queue_opts, **queue_opts or {}) queue_args = dict(self.queue_args, **queue_args or {}) exchange_opts = dict(self.exchange_opts, **exchange_opts or {}) if no_ack is None: no_ack = self.no_ack if not isinstance(queue, entity.Queue): exchange = entity.Exchange(name, **exchange_opts) queue = entity.Queue(name, exchange, name, queue_arguments=queue_args, **queue_opts) routing_key = name else: exchange = queue.exchange routing_key = queue.routing_key consumer = messaging.Consumer(channel, queue, accept=accept) producer = messaging.Producer(channel, exchange, serializer=serializer, routing_key=routing_key, compression=compression) super().__init__(channel, producer, consumer, no_ack)
[docs]class SimpleBuffer(SimpleQueue): """Simple API for ephemeral queues.""" no_ack = True queue_opts = {'durable': False, 'auto_delete': True} exchange_opts = {'durable': False, 'delivery_mode': 'transient', 'auto_delete': True}