_PulsectlBackend

def __init__(self, client_name: str, connect_timeout: float, op_timeout: float) -> None:
    self._name = client_name
    self._connect_timeout = connect_timeout
    self._op_timeout = op_timeout
    self._pulse: Any = None
    self._finalizer: Optional[weakref.finalize] = None
    self._loop: Optional[asyncio.AbstractEventLoop] = None
    self._inflight: set = set()

def _bind_loop(self) -> asyncio.AbstractEventLoop:
    loop = asyncio.get_running_loop()
    if self._loop is None:
        self._loop = loop
    elif self._loop is not loop:
        raise RuntimeError("AudioControl client used from a foreign event loop")
    return loop

async def connect(self) -> None:
    """Open the connection; raises when the server cannot be reached.

    The connect runs in its own task: a caller cancelled mid-handshake
    leaves the task to finish and the client is closed the moment it does,
    so libpulse never holds the state callback of a collected object.
    """
    loop = self._bind_loop()
    self._discard()
    pulse = pulsectl_asyncio.PulseAsync(self._name)
    task = loop.create_task(pulse.connect(timeout=self._connect_timeout))
    try:
        await asyncio.shield(task)
    except asyncio.CancelledError:
        task.add_done_callback(lambda t: (_retrieve(t), _close_quietly(pulse)))
        raise
    except BaseException:
        _close_quietly(pulse)
        raise
    self._pulse = pulse
    self._finalizer = weakref.finalize(self, _close_quietly, pulse)
    self._finalizer.atexit = False

async def call(self, op: Callable[[Any], Awaitable[T]]) -> T:
    """Run `op(pulse)` in an uncancellable task, bounded by the op timeout.

    A timeout abandons the connection: libpulse cancels the pending
    operation's callbacks on disconnect and the task then ends on its own.
    A cancelled caller leaves the task running; the operation completes
    normally and the client stays usable.

    Raises:
        AudioControlError: Not connected, or the operation failed.
        asyncio.TimeoutError: The server did not answer in time.
    """
    loop = self._bind_loop()
    pulse = self._pulse
    if pulse is None or not pulse.connected:
        raise AudioControlError("not connected to the sound server")

    async def body() -> T:
        if self._pulse is not pulse or not pulse.connected:
            raise AudioControlError("sound server connection closed")
        try:
            return await op(pulse)
        except AudioControlError:
            raise
        except Exception as e:
            raise AudioControlError(str(e) or type(e).__name__) from e

    task = loop.create_task(body())
    self._inflight.add(task)
    task.add_done_callback(self._inflight.discard)
    try:
        return await asyncio.wait_for(asyncio.shield(task), self._op_timeout)
    except asyncio.TimeoutError:
        task.add_done_callback(_retrieve)
        self._discard()
        raise
    except asyncio.CancelledError:
        task.add_done_callback(_retrieve)
        raise

def _take(self) -> Any:
    """Detach the current client (and its finalizer) for closing."""
    pulse, self._pulse = self._pulse, None
    if self._finalizer is not None:
        self._finalizer.detach()
        self._finalizer = None
    return pulse

def _discard(self) -> None:
    pulse = self._take()
    if pulse is not None:
        _close_quietly(pulse)

async def aclose(self) -> None:
    """Let in-flight operations finish, then close the connection."""
    pulse = self._take()
    if pulse is None:
        return
    try:
        if self._inflight:
            await asyncio.wait(set(self._inflight), timeout=self._op_timeout)
    finally:
        _close_quietly(pulse)
        if not self._inflight:
            self._loop = None

@staticmethod
def _node(obj: Any) -> PulseNode:
    return PulseNode(
        index=int(obj.index),
        name=str(obj.name),
        owner_module=getattr(obj, "owner_module", None),
        proplist=dict(getattr(obj, "proplist", None) or {}),
        source=getattr(obj, "source", None),
    )

async def sink_list(self) -> List[PulseNode]:
    return [self._node(s) for s in await self.call(lambda p: p.sink_list())]

async def source_list(self) -> List[PulseNode]:
    return [self._node(s) for s in await self.call(lambda p: p.source_list())]

async def source_output_list(self) -> List[PulseNode]:
    return [self._node(s) for s in await self.call(lambda p: p.source_output_list())]

async def server_defaults(self) -> Tuple[Optional[str], Optional[str]]:
    info = await self.call(lambda p: p.server_info())
    return (getattr(info, "default_sink_name", None) or None,
            getattr(info, "default_source_name", None) or None)

async def module_load(self, name: str, args: str) -> int:
    return int(await self.call(lambda p: p.module_load(name, args)))

async def module_unload(self, index: int) -> None:
    await self.call(lambda p: p.module_unload(index))

async def sink_default_set(self, name: str) -> None:
    await self.call(lambda p: p.sink_default_set(name))

async def source_default_set(self, name: str) -> None:
    await self.call(lambda p: p.source_default_set(name))

async def source_output_move(self, output_index: int, source_index: int) -> None:
    await self.call(lambda p: p.source_output_move(output_index, source_index))