code: def close(self, reason=None):
          
          with self._closing:
              if self._closed:
                  return
  
              # Stop consuming messages.
              if self.is_active:
                  _LOGGER.debug("Stopping consumer.")
                  self._consumer.stop()
              self._consumer = None
  
              self._rpc.close()
              self._rpc = None
              self._closed = True
              _LOGGER.debug("Finished stopping manager.")
  
          if reason:
              # Raise an exception if a reason is provided
              _LOGGER.debug("reason for closing: %s" % reason)
              if isinstance(reason, Exception):
                  raise reason
              raise RuntimeError(reason)
docs: Stop consuming messages and shutdown all helper threads.
  
          This method is idempotent. Additional calls will have no effect.
  
          Args:
              reason (Any): The reason to close this. If None, this is considered
                  an "intentional" shutdown.