For the first API, it is necessary to consider exceptions and disconnections:

if an exception occurs in the proxy, what do you want the callback to receive; or conversely, if an exception occurs in the callback, what should the proxy send to the client?

We use a closable pipeline to implement this functionality. The diagram is as follows:

flowchart LR;
    Client -->|A| CallbackReceive;
    CallbackReceive -->|B| CallbackSend;
    CallbackSend -->|C| TargetServer;

A and C are Anyio's memory-object-streams (closable pipeline).

For Exceptions

If any of the pipelines(A, B, or C) encounter an exception, the proxy will automatically close the WebSocket connections with the client and the target server. The close code is hardcoded to 1011, but this might change in the future, so you should not rely on it.

if an exception occurs in the proxy, what do you want the callback to receive?

The proxy will close the sending end of A and the receiving end of C, causing the callback to receive an exception(also).

However, note that for the current implementation, the exceptions received by callback are raised by Anyio, but this might change in the future, so you should not rely on it.

if an exception occurs in the callback, what should the proxy send to the client?

In any case, the callback needs to ensure that it closes the receiving end of A and the sending end of C so that the proxy knows the callback has become invalid. In this situation, the proxy will automatically close the WebSocket connections with 1011.

the callback needs to ensure that it closes the receiving end of A and the sending end of C

To make this more user-friendly, we will provide a context manager. The API will be as follows:

from contextlib import AbstractContextManager
from typing import Tuple

from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream


async def server_to_client_callback(
    pipe_builder: AbstractContextManager[
        Tuple[MemoryObjectReceiveStream[str], MemoryObjectSendStream[str]]
    ],
):
    # The ContextManager will automatically close both `receiver` and `sender`
    with pipe_builder as (receiver, sender):
        async for recv in receiver:
            print(f"Received: {recv}")
            resp = f"Modified: {recv}"
            await sender.send(resp)
            print(time.time)

For Disconnections

If the proxy receives a disconnection request

The proxy will close the pipeline, and in this case, the callback will receive an exception.

If the callback wants to disconnect

This is not possible for now. The callback cannot disconnect normally; it can only close the pipeline. In this case, the proxy will treat it as an exception.

CC @IvesAwadi

What do you think?

CC @IvesAwadi

What do you think?

Looks good to me.