WebSocketPipe 0.9.0

使用方法

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();

IWebSocketPipe 接口扩展了 IDuplexPipe，公开了可用来读取传入消息和写入传出消息的 Input 和 Output 属性。

例如，为了读取传入数据并将其写入控制台，我们可以编写以下代码：

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.");
}

同样，为了将控制台输入写入底层 WebSocket，我们使用类似以下代码：

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.");
}

如果我们想要同时读取和写入，并等待这两个操作完成，只需等待这两个任务：

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

请注意，完成 PipeWriter 将自动使读者收到完成的结果并退出循环。此外，整体 IWebSocketPipe.RunAsync 任务也将运行到完成。

如果创建时将 closeWhenCompleted 设置为 true，则 IWebSocketPipe 会负责在输入或输出完成时优雅地关闭连接。

或者，也可以显式完成整个管道，同时设置可选的套接字关闭状态和状态描述，供服务器操作使用

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

指定关闭状态将始终关闭底层套接字。

您也可以在服务器上使用它。以下示例基本上来自于 ASP.NET Core 中的 WebSocket 文档，并适当地使用 WebSocketPipe 读取/写入客户端。

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();
    }
});

示例 Echo 方法很简单：

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);
    }
}

赞助商

也会在这里提到!