This document is for Kombu's development version, which can be significantly different from previous releases. Get the stable docs here: 3.0.
Source code for kombu.transport.memory
"""In-memory transport."""
from __future__ import absolute_import, unicode_literals
from kombu.five import Queue, values
from . import base
from . import virtual
[docs]class Channel(virtual.Channel):
"""In-memory 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, **kwargs):
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 = {}