def METHOD_NAME(self) -> None:
    """Stop the duplexer. Most likely idempotent."""
    with contextlib.suppress(Exception):
        if self._loop.is_running():
            if self._ws:
                asyncio.run_coroutine_threadsafe(
                    self._ws.close(), loop=self._loop
                ).result()
            try:
                self._loop.call_soon_threadsafe(self._connection.cancel)
                asyncio.run_coroutine_threadsafe(
                    asyncio.wait_for(self._connection, None),
                    loop=self._loop,
                ).result()
            except (OSError, asyncio.CancelledError, CancelledError):
                # The OSError will have been raised in send/receive already.
                pass
    self._loop.call_soon_threadsafe(self._loop.METHOD_NAME)
    self._loop_thread.join()