Source code for kombu.transport.memory

"""
kombu.transport.memory
======================

In-memory transport.

"""
from __future__ import absolute_import

from kombu.five import Queue, values

from . import virtual


[docs]class Channel(virtual.Channel): queues = {} do_restore = False supports_fanout = True def _has_queue(self, queue, **kwargs): return queue in self.queues def _new_queue(self, queue, **kwargs): if queue not in self.queues: self.queues[queue] = Queue() def _get(self, queue, timeout=None): return self._queue_for(queue).get(block=False) def _queue_for(self, queue): if queue not in self.queues: self.queues[queue] = Queue() return self.queues[queue] def _queue_bind(self, *args): pass def _put_fanout(self, exchange, message, routing_key=None, **kwargs): for queue in self._lookup(exchange, routing_key): self._queue_for(queue).put(message) def _put(self, queue, message, **kwargs): self._queue_for(queue).put(message) def _size(self, queue): return self._queue_for(queue).qsize() def _delete(self, queue, *args): self.queues.pop(queue, None) def _purge(self, queue): q = self._queue_for(queue) size = q.qsize() q.queue.clear() return size
[docs] def close(self): super(Channel, self).close() for queue in values(self.queues): queue.empty() self.queues = {}
[docs] def after_reply_message_received(self, queue): pass
[docs]class Transport(virtual.Transport): Channel = Channel #: memory backend state is global. state = virtual.BrokerState() driver_type = 'memory' driver_name = 'memory'
[docs] def driver_version(self): return 'N/A'