.NET 10 giới thiệu một API mới, WebSocketStream, nhằm đơn giản hóa việc xử lý luồng (streaming) qua WebSocket trong .NET.

API WebSocket truyền thống ở mức độ thấp và yêu cầu nhiều mã nguồn để xử lý các vấn đề như quản lý bộ đệm, khung, tái tạo tin nhắn, mã hóa/ giải mã, và viết các wrapper tùy chỉnh để tích hợp với luồng, kênh hoặc các trừu tượng truyền tải khác. Điều này khiến việc sử dụng WebSocket trở nên phức tạp, đặc biệt là cho các ứng dụng có giao thức dựa trên luồng hoặc xử lý sự kiện.

WebSocketStream giải quyết vấn đề này bằng cách cung cấp một lớp trừu tượng dựa trên Stream cho WebSocket, giúp tích hợp liền mạch với các API hiện có để đọc, ghi và phân tích dữ liệu, dù đó là dữ liệu nhị phân hay văn bản, và giảm thiểu nhu cầu viết các pipeline thủ công.

Vấn đề mà WebSocketStream giải quyết

Trước .NET 10:

• System.Net.WebSockets.WebSocket
• Mô hình dựa trên tin nhắn (Message-based)
• Phải tự xử lý:
• Các phần nhỏ (Fragment)
• Ranh giới tin nhắn
• Vòng lặp Receive / Send
• Không thể trừu tượng hóa như Stream

Kết quả là:

• Mã nguồn WebSocket phức tạp, lặp đi lặp lại, dễ gây lỗi
• Khó tích hợp với hệ sinh thái Stream (nén, mã hóa, pipeline, serialize)
• Mô hình lập trình WebSocket tách biệt so với TCP / NamedPipe / HTTP Body

Thay đổi cơ bản của WebSocketStream

WebSocketStream trong .NET 10:

Trừu tượng hóa kết nối WebSocket thành một Stream thực sự

Điều này nghĩa là:

• Không cần quan tâm đến khung tin nhắn, phần nhỏ, ranh giới
• Có thể trực tiếp:
• ReadAsync
• WriteAsync
• Tích hợp liền mạch với:
• PipeReader / PipeWriter
• Hệ thống middleware của System.IO.Stream

Đây là một cải tiến mô hình, không chỉ là cú pháp mới.

Ví dụ về mã nguồn

using System;
using System.IO;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;

// Giao thức nhị phân streaming (ví dụ: AMQP).
Stream transportStream = WebSocketStream.Create(
    connectedWebSocket,
    WebSocketMessageType.Binary,
    closeTimeout: TimeSpan.FromSeconds(10));
await message.SerializeToStreamAsync(transportStream, cancellationToken);
var receivePayload = new byte[payloadLength];
await transportStream.ReadExactlyAsync(receivePayload, cancellationToken);
transportStream.Dispose();
// `Dispose` tự động xử lý đóng kết nối.

Xử lý giao thức văn bản (ví dụ: STOMP)

using System.IO;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;

// Giao thức văn bản streaming (ví dụ: STOMP).
using Stream transportStream = WebSocketStream.Create(
    connectedWebSocket, 
    WebSocketMessageType.Text,
    ownsWebSocket: true);
// Tích hợp với API dựa trên Stream.
// Không đóng stream vì nó cũng được sử dụng để ghi.
using var transportReader = new StreamReader(transportStream, leaveOpen: true); 
var line = await transportReader.ReadLineAsync(cancellationToken); // Xử lý UTF-8 và dòng mới tự động.
transportStream.Dispose(); // Xử lý đóng kết nối tự động khi `Dispose`.

Xử lý giao thức nhị phân (ví dụ: AMQP)

using System;
using System.IO;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;

// Giao thức nhị phân streaming (ví dụ: AMQP).
Stream transportStream = WebSocketStream.Create(
    connectedWebSocket,
    WebSocketMessageType.Binary,
    closeTimeout: TimeSpan.FromSeconds(10));
await message.SerializeToStreamAsync(transportStream, cancellationToken);
var receivePayload = new byte[payloadLength];
await transportStream.ReadExactlyAsync(receivePayload, cancellationToken);
transportStream.Dispose();
// `Dispose` tự động xử lý đóng kết nối.

Đọc tin nhắn dưới dạng stream (ví dụ: JSON deserialize)

using System.IO;
using System.Net.WebSockets;
using System.Text.Json;

// Đọc một tin nhắn dưới dạng stream (ví dụ: JSON deserialize).
using Stream messageStream = WebSocketStream.CreateReadableMessageStream(connectedWebSocket, WebSocketMessageType.Text);
// JsonSerializer.DeserializeAsync đọc đến cuối stream.
var appMessage = await JsonSerializer.DeserializeAsync<AppMessage>(messageStream);

Ghi tin nhắn dưới dạng stream (ví dụ: serialize nhị phân)

using System;
using System.IO;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;

// Ghi một tin nhắn dưới dạng stream (ví dụ: serialize nhị phân).
public async Task SendMessageAsync(AppMessage message, CancellationToken cancellationToken)
{
    using Stream messageStream = WebSocketStream.CreateWritableMessageStream(_connectedWebSocket, WebSocketMessageType.Binary);
    foreach (ReadOnlyMemory<byte> chunk in message.SplitToChunks())
    {
        await messageStream.WriteAsync(chunk, cancellationToken);
    }
} // EOM gửi khi messageStream.Dispose().

Các trường hợp sử dụng thực tế của WebSocketStream

Trường hợp 1: Dữ liệu thời gian thực (Streaming Data)

Các ví dụ kinh doanh

• Push trạng thái trạm sạc điện (công suất, dòng điện, điện áp)
• Màn hình giám sát thời gian thực (luồng chỉ số)
• Luồng giá / đơn hàng
• Đăng ký theo dõi log / Trace thời gian thực

Sau khi sử dụng WebSocketStreamWebSocket = một luồng dữ liệu liên tục

Trường hợp 2: Đầu ra AI / LLM thời gian thực (Token Streaming)

Các ví dụ kinh doanh

• Trả về kết quả mô hình lớn theo luồng
• Quá trình suy luận Copilot / Chat / Agent
• Giải pháp SSE + WebSocket

Trường hợp 3: Truyền tải tệp lớn / luồng nhị phân

using var file = File.OpenRead(path);  
await file.CopyToAsync(webSocketStream);

Giống như TCP, nhưng vẫn là WebSocket

Trường hợp 4: RPC / Tunnel giao thức (Protocol Tunneling)

• Có thể chạy giao thức Stream hiện có
• Tương tự gRPC
• Framing tùy chỉnh
• Giao thức thông tin thiết bị

await protocolHandler.RunAsync(webSocketStream);

Trường hợp 5: Tích hợp sâu với System.IO.Pipelines

Xây dựng dịch vụ máy chủ hiệu suất cao

var reader = PipeReader.Create(webSocketStream);  
var writer = PipeWriter.Create(webSocketStream);

• Sử dụng Pipelines để:
• Phân tích hiệu suất cao
• Xử lý không sao chép
• Giống như Kestrel / gRPC / giao thức tự phát triển

Phù hợp cho:

• Giao tiếp thời gian thực tần suất cao
• Dịch vụ có triệu kết nối
• Dịch vụ trung tâm quan trọng