[issue tracker]: pipe dead lock in WebSocketProxy `callback`

[WSH032]

I just found a bug. The current implementation only supports a strict one-receive-one-send mode within a single loop. If this pattern is violated, such as multiple receives and one send, one receive and multiple sends, or sending before receiving within a single loop, it will result in a deadlock.

async def callback(ctx: CallbackPipeContextType[str]) -> None:
    with ctx as (sender, receiver):
        # multiple receives and one send, dead lock!
        await receiver.receive()
        await receiver.receive()
        await sender.send("foo")

async def callback(ctx: CallbackPipeContextType[str]) -> None:
    with ctx as (sender, receiver):
        # one receive and multiple sends, dead lock!
        async for message in receiver:
            await sender.send("foo")
            await sender.send("bar")

async def callback(ctx: CallbackPipeContextType[str]) -> None:
    with ctx as (sender, receiver):
        # sending before receiving, dead lock!
        await sender.send("foo")
        async for message in receiver:
            await sender.send(message)

Unfortunately, we can't resolve this issue until the underlying logic is rewritten using anyio and memory-object-streams.

There is already a PR for anyio: #34. However, this PR still has many issues, and I currently don't have time to merge it.

Originally posted by @WSH032 in https://github.com/WSH032/fastapi-proxy-lib/issues/41#issuecomment-2242214863

[WSH032]

The current implementation processes message receiving and sending serially within the same task, which prevents multiple receives and sends.

https://github.com/WSH032/fastapi-proxy-lib/blob/52f048f5c48e7582169bc97f215b8a558bb7f018/src/fastapi_proxy_lib/core/websocket.py#L344-L421

To resolve this issue, we need to modify the underlying implementation by creating four asynchronous tasks specifically for client receive, client send, server receive, and server send, while using pipelines to pass messages between them.

However, managing so many asynchronous tasks robustly is a challenge. We must switch to the anyio backend and rely on its structured concurrency to manage the tasks.

---

I’m currently not motivated to resolve this issue because I’m unsure if there’s demand from the community.

If there is interest, PRs are welcome. Alternatively, you can give a thumbs up on the issue to indicate community feedback.