Skip to content

System.IO.Pipelines API adapter for System.Net.WebSockets

License

Notifications You must be signed in to change notification settings

devlooped/WebSocketPipe

Repository files navigation

Icon WebSocketPipe

High-performance System.IO.Pipelines API adapter for System.Net.WebSockets

Version Downloads License Build

Usage

using Devlooped.Net;

var client = new ClientWebSocket();
await client.ConnectAsync(serverUri, CancellationToken.None);

using IWebSocketPipe pipe = WebSocketPipe.Create(client, closeWhenCompleted: true);

// Start the pipe before hooking up the processing
var run = pipe.RunAsync();

The IWebSocketPipe interface extends IDuplexPipe, exposing Input and Output properties that can be used to read incoming messages and write outgoing ones.

For example, to read incoming data and write it to the console, we could write the following code:

await ReadIncoming(pipe.Input);

async Task ReadIncoming(PipeReader reader)
{
    while (await reader.ReadAsync() is var result && !result.IsCompleted)
    {
        Console.WriteLine($"Received: {Encoding.UTF8.GetString(result.Buffer)}");
        reader.AdvanceTo(result.Buffer.End);
    }
    Console.WriteLine($"Done reading.");
}

Similarly, to write to the underlying websocket the input entered in the console, we use code like the following:

await SendOutgoing(pipe.Output);

async Task SendOutgoing(PipeWriter writer)
{
    while (Console.ReadLine() is var line && line?.Length > 0)
    {
        Encoding.UTF8.GetBytes(line, writer);
    }
    await writer.CompleteAsync();
    Console.WriteLine($"Done writing.");
}

If we wanted to simultaneously read and write and wait for completion of both operations, we could just wait for both tasks:

// Wait for completion of processing code
await Task.WhenAny(
    ReadIncoming(pipe.Input),
    SendOutgoing(pipe.Output));

Note that completing the PipeWriter automatically causes the reader to reveive a completed result and exit the loop. In addition, the overall IWebSocketPipe.RunAsync task will also run to completion.

The IWebSocketPipe takes care of gracefully closing the connection when the input or output are completed, if closeWhenCompleted is set to true when creating it.

Alternatively, it's also possible to complete the entire pipe explicitly, while setting an optional socket close status and status description for the server to act on:

await pipe.CompleteAsync(WebSocketCloseStatus.NormalClosure, "Done processing");

Specifying a close status will always close the underlying socket.

The WebSocketPipe can also be used on the server. The following example is basically taken from the documentation on WebSockets in ASP.NET Core and adapted to use a WebSocketPipe to echo messages to the client:

app.Use(async (context, next) =>
{
    if (context.Request.Path == "/ws")
    {
        if (context.WebSockets.IsWebSocketRequest)
        {
            using var websocket = await context.WebSockets.AcceptWebSocketAsync();
            using var pipe = WebSocketPipe.Create(websocket, true);
            await Task.WhenAll(Echo(pipe), pipe.RunAsync(context.RequestAborted));
        }
        else
        {
            context.Response.StatusCode = (int) HttpStatusCode.BadRequest;
        }
    }
    else
    {
        await next();
    }
});

The sample Echo method is simply:

async Task Echo(IDuplexPipe pipe)
{
    while (await pipe.Input.ReadAsync() is var result && !result.IsCompleted)
    {
        // Just assume we get a single-segment entry, for simplicity
        await pipe.Output.WriteAsync(result.Buffer.First);
        pipe.Input.AdvanceTo(result.Buffer.End);
    }
}

Dogfooding

CI Version Build

We also produce CI packages from branches and pull requests so you can dogfood builds as quickly as they are produced.

The CI feed is https://pkg.kzu.io/index.json.

The versioning scheme for packages is:

  • PR builds: 42.42.42-pr[NUMBER]
  • Branch builds: 42.42.42-[BRANCH].[COMMITS]

Sponsors

Clarius Org Kirill Osenkov MFB Technologies, Inc. Stephen Shaw Torutek DRIVE.NET, Inc. Ashley Medway Keith Pickford Thomas Bolon Kori Francis Toni Wenzel Giorgi Dalakishvili Uno Platform Dan Siegel Reuben Swartz Jacob Foshee Eric Johnson Ix Technologies B.V. David JENNI Jonathan Oleg Kyrylchuk Charley Wu Jakob Tikjøb Andersen Seann Alexander Tino Hager Mark Seemann Ken Bonny Simon Cropp agileworks-eu sorahex Zheyu Shen Vezel ChilliCream 4OTC

Sponsor this project  

Learn more about GitHub Sponsors