Gracefully stop the Worker and all of its consumers and
        workers.

        Parameters:
          timeout(int): The number of milliseconds to wait for
            everything to shut down.

def stop(self, timeout=600000):
        """Gracefully stop the Worker and all of its consumers and
        workers.

        Parameters:
          timeout(int): The number of milliseconds to wait for
            everything to shut down.
        """
        self.broker.emit_before("worker_shutdown", self)
        self.logger.info("Shutting down...")

        # Stop workers before consumers.  The consumers are kept alive
        # during this process so that heartbeats keep being sent to
        # the broker while workers finish their current tasks.
        self.logger.debug("Stopping workers...")
        for thread in self.workers:
            thread.stop()

        join_all(self.workers, timeout)
        self.logger.debug("Workers stopped.")
        self.logger.debug("Stopping consumers...")
        for thread in self.consumers.values():
            thread.stop()

        join_all(self.consumers.values(), timeout)
        self.logger.debug("Consumers stopped.")

        self.logger.debug("Requeueing in-memory messages...")
        messages_by_queue = defaultdict(list)
        for _, message in iter_queue(self.work_queue):
            messages_by_queue[message.queue_name].append(message)

        for queue_name, messages in messages_by_queue.items():
            try:
                self.consumers[queue_name].requeue_messages(messages)
            except ConnectionError:
                self.logger.warning("Failed to requeue messages on queue %r.", queue_name, exc_info=True)
        self.logger.debug("Done requeueing in-progress messages.")

        self.logger.debug("Closing consumers...")
        for consumer in self.consumers.values():
            consumer.close()

        self.logger.debug("Consumers closed.")
        self.broker.emit_after("worker_shutdown", self)
        self.logger.info("Worker has been shut down.")