[API Proposal]: Add APIs to WebSocket which allow it to be read as a Stream

[christothes]

Background and motivation

Utilizing WebSockets is a convenient approach to writing real-time audio processing code for ASP.NET applications. One such scenario is implementing a real-time conversation with Open AI.

OpenAI's real-time API SendInputAudioAsync accept a Stream as input which leaves it up to the developer to write a custom Stream implementation that reads from an underlying WebSocket. It would be a nice enhancement to the WebSocket APIs if one could wrap read operations in a Stream.

API Proposal

public class WebSocket
{
    public Stream AsStream();
}

API Usage

using var webSocket = await context.WebSockets.AcceptWebSocketAsync();
using RealtimeConversationSession session = await InitSession(realtime);
// <...>
using var stream = webSocket.AsStream();
await session.SendInputAudioAsync(stream);

Alternative Designs

No response

Risks

WebSocket doesn’t provide synchronous methods for wire-based operations, so all of the Stream sync APIs (including Dispose, which presumably would need to not just Dispose the WebSocket but also CloseAsync it) would be sync-over-async.

[dotnet-policy-service]

Tagging subscribers to this area: @dotnet/ncl
See info in area-owners.md if you want to be subscribed.

[MihaZupan]

It's an interesting idea, though its use seems limited only to cases where you know that only the binary content is being transmitted (e.g. only the audio, no control data, no extra framing).
There's also the question of what happens with close messages -- does the user not care about the data, who's responsible for responding to them, do you send them during disposal.

I can see it being useful in cases where you just need an opaque Stream and happen to be using WebSocket as the transport.

Sample code if someone needed such a Stream could be something like this (untested):

public sealed class WebSocketStream : Stream
{
    private readonly WebSocket _webSocket;

    public WebSocketStream(WebSocket webSocket) => _webSocket = webSocket;

    public override bool CanRead => _webSocket.State is WebSocketState.Open or WebSocketState.CloseSent;
    public override bool CanWrite => _webSocket.State is WebSocketState.Open or WebSocketState.CloseReceived;
    public override bool CanSeek => false;

    public override void Flush() { }
    public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;

    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
        ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();

    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
        WriteAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();

    public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
    {
        ValueWebSocketReceiveResult result = await _webSocket.ReceiveAsync(buffer, cancellationToken);

        if (result.MessageType != WebSocketMessageType.Binary)
        {
            if (result.MessageType == WebSocketMessageType.Close)
            {
                await _webSocket.SendAsync(ReadOnlyMemory<byte>.Empty, WebSocketMessageType.Close, endOfMessage: true, cancellationToken);
                return 0;
            }

            throw new Exception("Expected binary messages");
        }

        return result.Count;
    }

    public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default) =>
        _webSocket.SendAsync(buffer, WebSocketMessageType.Binary, endOfMessage: true, cancellationToken);

    public override ValueTask DisposeAsync()
    {
        Dispose(true);
        return default;
    }

    protected override void Dispose(bool disposing) => _webSocket.Dispose();

    public override int Read(byte[] buffer, int offset, int count) => ReadAsync(buffer, offset, count).GetAwaiter().GetResult();
    public override void Write(byte[] buffer, int offset, int count) => WriteAsync(buffer, offset, count).GetAwaiter().GetResult();

    public override long Length => throw new NotSupportedException();
    public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
    public override void SetLength(long value) => throw new NotSupportedException();
}