Function interactive_client

src/websockets/cli.py:104–142

(uri: str)

async def interactive_client(uri: str) -> None:
    try:
        websocket = await connect(uri)
    except Exception as exc:
        print(f"Failed to connect to {uri}: {exc}.")
        sys.exit(1)
    else:
        print(f"Connected to {uri}.")

    loop = asyncio.get_running_loop()
    transport, protocol = await loop.connect_read_pipe(ReadLines, sys.stdin)
    incoming = asyncio.create_task(
        print_incoming_messages(websocket),
    )
    outgoing = asyncio.create_task(
        send_outgoing_messages(websocket, protocol.messages),
    )
    try:
        await asyncio.wait(
            [incoming, outgoing],
            # Clean up and exit when the server closes the connection
            # or the user enters EOT (^D), whichever happens first.
            return_when=asyncio.FIRST_COMPLETED,
        )
    # asyncio.run() cancels the main task when the user triggers SIGINT (^C).
    # https://docs.python.org/3/library/asyncio-runner.html#handling-keyboard-interruption
    # Clean up and exit without re-raising CancelledError to prevent Python
    # from raising KeyboardInterrupt and displaying a stack track.
    except asyncio.CancelledError:  # pragma: no cover
        pass
    finally:
        incoming.cancel()
        outgoing.cancel()
        transport.close()

    await websocket.close()
    assert websocket.close_code is not None and websocket.close_reason is not None
    close_status = Close(websocket.close_code, websocket.close_reason)
    print_over_input(f"Connection closed: {close_status}.")


def main(argv: list[str] | None = None) -> None: