Method run

dramatiq/worker.py:296–344

(self)

        self.delay_queue: PriorityQueue[_WorkQueueItem] = PriorityQueue()

    def run(self) -> None:
        self.logger.debug("Running consumer thread...")
        self.running = True
        self.broker.emit_after("consumer_thread_boot", self)
        while self.running:
            if self.paused:
                self.logger.debug("Consumer is paused. Sleeping for %.02fms...", self.worker_timeout)
                self.paused_event.set()
                time.sleep(self.worker_timeout / 1000)
                continue

            try:
                self.consumer = self.broker.consume(
                    queue_name=self.queue_name,
                    prefetch=self.prefetch,
                    timeout=self.worker_timeout,
                )

                for message in self.consumer:
                    if message is not None:
                        self.handle_message(message)

                    elif self.paused:
                        break

                    self.handle_delayed_messages()
                    if not self.running:
                        break

            except BrokerConnectionError as e:
                self.logger.critical("Consumer encountered a connection error: %s", e)
                self.delay_queue = PriorityQueue()

            except Exception:
                self.logger.critical("Consumer encountered an unexpected error.", exc_info=True)
                # Avoid leaving any open file descriptors around when
                # an exception occurs.
                self.close()

            # While the consumer is running (i.e. hasn't been shut down),
            # try to restart it once a second.
            if self.running:
                self.logger.info("Restarting consumer in %0.2f seconds.", CONSUMER_RESTART_DELAY_SECS)
                self.close()
                time.sleep(CONSUMER_RESTART_DELAY_SECS)

        # If it's no longer running, then shut it down gracefully.
        self.broker.emit_before("consumer_thread_shutdown", self)
        self.logger.debug("Consumer thread stopped.")

    def handle_delayed_messages(self) -> None:
        """Enqueue any delayed messages whose eta has passed."""