FM / consumer

/opt/imh-python/lib/python3.9/site-packages/celery/worker/consumer/

/opt/imh-python/lib/python3.9/site-packages/celery/worker/consumer

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
agent.py	525	0644	edit dl rm
connection.py	1026	0644	edit dl rm
consumer.py	29091	0644	edit dl rm
control.py	946	0644	edit dl rm
events.py	2054	0644	edit dl rm
gossip.py	6863	0644	edit dl rm
heart.py	930	0644	edit dl rm
mingle.py	2519	0644	edit dl rm
tasks.py	1960	0644	edit dl rm
__init__.py	391	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/celery/worker/consumer/events.py (2054B)

"""Worker Event Dispatcher Bootstep.

``Events`` -> :class:`celery.events.EventDispatcher`.
"""
from kombu.common import ignore_errors

from celery import bootsteps

from .connection import Connection

__all__ = ('Events',)

class Events(bootsteps.StartStopStep):
    """Service used for sending monitoring events."""

requires = (Connection,)

def __init__(self, c,
                 task_events=True,
                 without_heartbeat=False,
                 without_gossip=False,
                 **kwargs):
        self.groups = None if task_events else ['worker']
        self.send_events = (
            task_events or
            not without_gossip or
            not without_heartbeat
        )
        self.enabled = self.send_events
        c.event_dispatcher = None
        super().__init__(c, **kwargs)

def start(self, c):
        # flush events sent while connection was down.
        prev = self._close(c)
        dis = c.event_dispatcher = c.app.events.Dispatcher(
            c.connection_for_write(),
            hostname=c.hostname,
            enabled=self.send_events,
            groups=self.groups,
            # we currently only buffer events when the event loop is enabled
            # XXX This excludes eventlet/gevent, which should actually buffer.
            buffer_group=['task'] if c.hub else None,
            on_send_buffered=c.on_send_event_buffered if c.hub else None,
        )
        if prev:
            dis.extend_buffer(prev)
            dis.flush()

def stop(self, c):
        pass

def _close(self, c):
        if c.event_dispatcher:
            dispatcher = c.event_dispatcher
            # remember changes from remote control commands:
            self.groups = dispatcher.groups

# close custom connection
            if dispatcher.connection:
                ignore_errors(c, dispatcher.connection.close)
            ignore_errors(c, dispatcher.close)
            c.event_dispatcher = None
            return dispatcher

def shutdown(self, c):
        self._close(c)

cmd: