def stop(self) -> None:
        """Closes the open stream and resets all management resources."""
        if (
            self._manage_stream_loop_future is not None
            and not self._manage_stream_loop_future.done()
        ):
            self._manage_stream_loop_future.cancel()
        self._response_demux.publish_exception(asyncio.CancelledError())
        self._reset()
Closes the open stream and resets all management resources.
cirq-google/cirq_google/engine/stream_manager.py