This document describes the current stable version of Celery (4.0). For development docs, go here.

Source code for celery.app.events

"""Implementation for the app.events shortcuts."""
from __future__ import absolute_import, unicode_literals
from contextlib import contextmanager
from kombu.utils.objects import cached_property


[docs]class Events(object): """Implements app.events.""" receiver_cls = 'celery.events.receiver:EventReceiver' dispatcher_cls = 'celery.events.dispatcher:EventDispatcher' state_cls = 'celery.events.state:State' def __init__(self, app=None): self.app = app @cached_property
[docs] def Receiver(self): return self.app.subclass_with_self( self.receiver_cls, reverse='events.Receiver')
@cached_property
[docs] def Dispatcher(self): return self.app.subclass_with_self( self.dispatcher_cls, reverse='events.Dispatcher')
@cached_property
[docs] def State(self): return self.app.subclass_with_self( self.state_cls, reverse='events.State')
@contextmanager
[docs] def default_dispatcher(self, hostname=None, enabled=True, buffer_while_offline=False): with self.app.amqp.producer_pool.acquire(block=True) as prod: # pylint: disable=too-many-function-args # This is a property pylint... with self.Dispatcher(prod.connection, hostname, enabled, prod.channel, buffer_while_offline) as d: yield d