kombu/examples/simple_task_queue/worker.py

43 lines
1.2 KiB
Python
Raw Normal View History

2011-09-12 09:25:42 +00:00
from kombu.mixins import ConsumerMixin
2012-11-02 14:32:45 +00:00
from kombu.log import get_logger
2014-05-19 21:27:36 +00:00
from kombu.utils import reprcall
2011-09-12 09:25:42 +00:00
2013-09-10 16:26:12 +00:00
from .queues import task_queues
2011-09-12 09:25:42 +00:00
2012-11-02 14:32:45 +00:00
logger = get_logger(__name__)
2011-09-12 09:47:22 +00:00
2011-09-12 09:25:42 +00:00
class Worker(ConsumerMixin):
2011-09-12 09:47:22 +00:00
def __init__(self, connection):
self.connection = connection
2011-09-12 09:25:42 +00:00
def get_consumers(self, Consumer, channel):
2011-09-12 09:47:22 +00:00
return [Consumer(queues=task_queues,
accept=['pickle', 'json'],
2011-09-12 09:47:22 +00:00
callbacks=[self.process_task])]
2011-09-12 09:25:42 +00:00
2011-09-12 09:47:22 +00:00
def process_task(self, body, message):
fun = body['fun']
args = body['args']
kwargs = body['kwargs']
2012-11-02 14:32:45 +00:00
logger.info('Got task: %s', reprcall(fun.__name__, args, kwargs))
2011-09-12 09:47:22 +00:00
try:
2014-05-19 21:27:36 +00:00
fun(*args, **kwargs)
2013-02-12 15:06:49 +00:00
except Exception as exc:
2012-11-02 14:32:45 +00:00
logger.error('task raised exception: %r', exc)
2011-09-12 09:25:42 +00:00
message.ack()
if __name__ == '__main__':
from kombu import Connection
from kombu.utils.debug import setup_logging
# setup root logger
setup_logging(loglevel='INFO', loggers=[''])
2011-09-12 09:25:42 +00:00
with Connection('amqp://guest:guest@localhost:5672//') as conn:
2011-09-12 11:57:12 +00:00
try:
worker = Worker(conn)
worker.run()
2011-09-12 11:57:12 +00:00
except KeyboardInterrupt:
print('bye bye')