code: def stop(self, timeout=600000):
          
          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.")
docs: 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.