1"""Worker Event Heartbeat Bootstep.""" 2from __future__ import absolute_import, unicode_literals 3 4from celery import bootsteps 5from celery.worker import heartbeat 6 7from .events import Events 8 9__all__ = ('Heart',) 10 11 12class Heart(bootsteps.StartStopStep): 13 """Bootstep sending event heartbeats. 14 15 This service sends a ``worker-heartbeat`` message every n seconds. 16 17 Note: 18 Not to be confused with AMQP protocol level heartbeats. 19 """ 20 21 requires = (Events,) 22 23 def __init__(self, c, 24 without_heartbeat=False, heartbeat_interval=None, **kwargs): 25 self.enabled = not without_heartbeat 26 self.heartbeat_interval = heartbeat_interval 27 c.heart = None 28 super(Heart, self).__init__(c, **kwargs) 29 30 def start(self, c): 31 c.heart = heartbeat.Heart( 32 c.timer, c.event_dispatcher, self.heartbeat_interval, 33 ) 34 c.heart.start() 35 36 def stop(self, c): 37 c.heart = c.heart and c.heart.stop() 38 shutdown = stop 39