/usr/share/pyshared/carrot/backends/queue.py is in python-carrot 0.10.7-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """
Backend for unit-tests, using the Python :mod:`Queue` module.
"""
from Queue import Queue
from carrot.backends.base import BaseMessage, BaseBackend
import time
import itertools
mqueue = Queue()
class Message(BaseMessage):
"""Message received from the backend.
See :class:`carrot.backends.base.BaseMessage`.
"""
class Backend(BaseBackend):
"""Backend using the Python :mod:`Queue` library. Usually only
used while executing unit tests.
Please not that this backend does not support queues, exchanges
or routing keys, so *all messages will be sent to all consumers*.
"""
Message = Message
def get(self, *args, **kwargs):
"""Get the next waiting message from the queue.
:returns: A :class:`Message` instance, or ``None`` if there is
no messages waiting.
"""
if not mqueue.qsize():
return None
message_data, content_type, content_encoding = mqueue.get()
return self.Message(backend=self, body=message_data,
content_type=content_type,
content_encoding=content_encoding)
def establish_connection(self):
# for drain_events
return self
def drain_events(self, timeout=None):
message = self.get()
if message:
self.callback(message)
else:
time.sleep(0.1)
def consume(self, limit=None):
"""Go into consume mode."""
for total_message_count in itertools.count():
if limit and total_message_count >= limit:
raise StopIteration
self.drain_events()
yield True
def declare_consumer(self, queue, no_ack, callback, consumer_tag,
nowait=False):
self.queue = queue
self.no_ack = no_ack
self.callback = callback
self.consumer_tag = consumer_tag
self.nowait = nowait
def queue_purge(self, queue, **kwargs):
"""Discard all messages in the queue."""
qsize = mqueue.qsize()
mqueue.queue.clear()
return qsize
def prepare_message(self, message_data, delivery_mode,
content_type, content_encoding, **kwargs):
"""Prepare message for sending."""
return (message_data, content_type, content_encoding)
def publish(self, message, exchange, routing_key, **kwargs):
"""Publish a message to the queue."""
mqueue.put(message)
|