code: def close(self):
          
          try:
              if self.consumer:
                  self.requeue_messages(m for _, m in iter_queue(self.delay_queue))
                  self.consumer.close()
          except ConnectionError:
              pass
docs: Close this consumer thread and its underlying connection.