StreamFrame

English | 简体中文

A general-purpose socket communication library that pluggably separates frame delimiting from payload encoding/decoding. Built for scenarios where peers exchange data over sockets (device communication, logistics/WMS integration, etc.): the common parts (connection management, reconnection, I/O, TCP packet fragmentation/reassembly, message dispatching) live in the core, while the parts that vary (framing, codec) are pluggable drivers.

Why StreamFrame

	Hand-rolled	StreamFrame
Onboarding a new device/protocol	Rewrite connection, delimiting, encoding	Write one `ICodec<T>` driver
Frame boundaries (length-prefix / STX-ETX / custom)	Every project rolls its own	Two built-in + pluggable
Fragmented/partial TCP packets	Manual buffer stitching	Handled by Pipelines
Reconnection / state machine	Hand-rolled	Built-in auto-reconnect
Send performance	Multiple copies	Optional single-buffer streaming encode

Installation

dotnet add package StreamFrame

XML message driver (optional):

dotnet add package StreamFrame.Protocols.Xml

Quick start

This is a single-connection device integration library. A passive listener serves one client at a time; AcceptFirstClientOnly = false keeps the listener open without adding multi-client handling. The snippets below represent separate processes/roles, with a host-provided ct. See the production configuration below and dispose connections with await using.

One connection = one framer + one codec + address/port/mode. Framing and codec are both fixed per connection: a connection carries exactly one message type in exactly one frame format.

using System.Net;
using System.Xml.Linq;
using StreamFrame;
using StreamFrame.Protocols.Xml;

// Server (passive listener)
var server = new StreamConnection<XDocument>(
    new LengthPrefixFramer(),              // 4-byte big-endian length header; or StxEtxFramer
    new XmlDocumentCodec(),                // swap in your own ICodec<T> for a custom protocol
    IPAddress.Any, 5100, isActive: false);
server.Start(ct);

await foreach (var doc in server.GetMessages(ct))
{
    var id = doc.Root?.Element("Id")?.Value;
    await server.SendAsync(XDocument.Parse($"<Reply><Echo>{id}</Echo></Reply>"), ct);
}

// Client (active connector)
var client = new StreamConnection<XDocument>(
    new LengthPrefixFramer(),
    new XmlDocumentCodec(),
    IPAddress.Parse("127.0.0.1"), 5100, isActive: true);
client.Start(ct);
await client.SendAsync(XDocument.Parse("<Message><Id>1</Id></Message>"), ct);

Concepts

Framing (`IFramer`) — slicing frames out of a byte stream

Implementation	Delimiting	Suitable for
`LengthPrefixFramer`	4-byte big-endian length header + payload	General purpose, binary-safe
`StxEtxFramer`	Wrapped in STX `0x02` … ETX `0x03`	XML / plain-text payloads known to be marker-safe

Both built-in implementations support streaming single-buffer encoding (optional, removes the payload memcpy on send). Implement IFramer to define a custom frame format.

Codec (`ICodec<T>`) — parsing/writing the payload inside a frame

public interface ICodec<TMessage>
{
    TMessage Decode(in ReadOnlySequence<byte> frame, CancellationToken ct = default);
    void Encode(TMessage message, IBufferWriter<byte> writer, CancellationToken ct = default);
}

Onboarding a new device = writing one driver: implement ICodec<TMessage>, define your message type, pick a framer — everything else is reused. See StreamFrame.Protocols.Xml for the official example. A minimal JSON-message driver (System.Text.Json, span-written, AOT-safe):

sealed class SystemTextJsonCodec : ICodec<JsonElement>
{
    public static readonly SystemTextJsonCodec Instance = new();

    public JsonElement Decode(in ReadOnlySequence<byte> frame, CancellationToken ct = default)
    {
        using var document = JsonDocument.Parse(frame.ToArray());
        return document.RootElement.Clone(); // Own data beyond document disposal.
    }

    public void Encode(JsonElement message, IBufferWriter<byte> writer, CancellationToken ct = default)
    {
        var raw = message.GetRawText(); // serialized form (non-ASCII already escaped)
        var span = writer.GetSpan(Encoding.UTF8.GetMaxByteCount(raw.Length));
        writer.Advance(Encoding.UTF8.GetBytes(raw, span));
    }
}

Connection (`IStreamConnection<T>`) — the transport

Client/server dual mode: isActive: true connects out, false listens; IPv4/IPv6 dual-stack (listening on IPAddress.Any is normalized to dual-stack; IPv4 client addresses are displayed as plain IPv4)
Auto-reconnect: Connecting → Connected → Retry state machine; optional exponential backoff (MaxRetryDelayMs, doubling with cap + ±20% jitter, auto-reset on success); GetMessages is a stable stream across reconnections — received messages are not lost and enumeration does not break when the connection drops and recovers
Start & stop: the ct passed to Start(ct) is the connection's lifetime token — cancelling it stops connection/reconnection and tears the link down (state enters the terminal Disconnected, GetMessages completes naturally; create a new connection afterwards). DisposeAsync does the same. After shutdown, SendAsync throws ChannelClosedException
Waiting for readiness: await conn.WaitForConnectedAsync(ct) — completes immediately when connected; otherwise waits for the next successful connection (or caller cancellation, lifetime cancellation/disposal). Future waits after terminal shutdown are cancelled too, even without a caller token
Robustness: payload decode failures, incomplete-frame overflows/timeouts, send failures, and receive idle timeouts all invalidate the session and rebuild it automatically (no more "connection looks alive while messages silently vanish")
Liveness detection (optional): TCP KeepAlive and receive idle timeout to catch half-open connections (peer power loss / unplugged cable)
Events: ConnectionChanged state changes, FrameError frame-level diagnostics, RawBytesReceived/Sent raw bytes (HEX debugging)
Built-in metrics: System.Diagnostics.Metrics (meter StreamFrame) — frame/byte counters, reconnects, session duration, send-queue watermark; see the "Built-in metrics" section
Backpressure: bounded send queue — SendAsync waits when full; the receive side buffers without limit by default, or set ReceiveQueueCapacity — decoding pauses when the consumer lags, propagating TCP backpressure to the peer and preventing unbounded memory growth
Session-aware messaging (optional, advanced): CurrentSessionId / SendInSessionAsync (completes only after the whole frame is written to the socket; fails on session loss and never replays across sessions) / GetSessionMessages (messages carry their session id) — for protocols with strict session boundaries such as HSMS; see the dedicated section below

Buffer ownership and concurrency

Decode borrows frame only during the synchronous call. The returned message, including nested fields, must independently own retained data. Never return frame.First, slices or ReadOnlyMemory<byte> backed by pipeline storage: the pipe advances/releases its buffers, while business processing may happen after an await. Read-only memory does not imply ownership. This interface offers no transfer of pooled pipeline storage.

Encode must finish writing before returning. Do not retain/dispose writer or retain its buffers for asynchronous use. The send queue holds message objects: do not mutate or return their backing arrays after enqueue. Plain SendAsync completion means enqueue only; encoding may not have started. A dedicated array that is never modified afterwards is the simplest approach.

A session's send worker serializes its own encoding calls, not the codec instance. Encode and Decode may run concurrently. Session teardown waits at most two seconds: old decoders (for example, stalled by receive backpressure) may overlap new session decoders; a blocked old encoder cannot be assumed to have exited either. Use call-local state or synchronization and avoid blocking synchronous methods. Callers sharing a codec/framer across connections must ensure concurrency safety. One instance per connection still needs to support concurrent read/write and old/new session calls.

The same rules apply to IFramer: do not retain input sequences; output payload may borrow input for the immediately following synchronous codec call. Use the encoding writer only within the call. Other calls may interleave between IStreamingFramer.BeginFrame and EndFrame; keep pair state in the supplied writer rather than an instance's "current frame" field.

Complete, compilable OwnedBytesCodec / OwnedMemoryCodec examples return frame.ToArray() for both message types, including single-segment input. Their encoders synchronously call writer.Write(message) or writer.Write(message.Span). These exact sources are compiled in the test project for net8.0/net10.0/net48. OwnershipExampleTests overwrites single/multi-segment input and runs the real FrameDecoder with a pool that fills returned storage with 0xDD; asynchronous consumption after pipe disposal must still produce the original bytes.

Production configuration starting point

The complete ProductionExample.RunAsync demonstrates bounded queues, a ten-second readiness timeout, asynchronous consumption, lifetime cancellation and await using disposal, and is compiled for all three test targets. These values assume payloads up to 64 KiB: tune them to the protocol, peak traffic and processing time. Library defaults remain unchanged.

var framer = new LengthPrefixFramer(64 * 1024);
var options = new StreamConnectionOptions
{
    SendQueueCapacity = 128,
    ReceiveQueueCapacity = 128, // Default 0 is unbounded.
    MaxIncompleteFrameBufferBytes = 64 * 1024 + 4, // Includes length header.
    IncompleteFrameTimeoutMs = 5_000,
    ReceiveIdleTimeoutMs = 0, // Silence is legal; consider 15_000 with a 5s heartbeat.
    TcpKeepAlive = true,
    KeepAliveTimeMs = 30_000,
    KeepAliveIntervalMs = 1_500, // Modern .NET rounds up to 2 seconds (#62).
    AcceptFirstClientOnly = true,
};

Capacity counts messages, not bytes: 128 × 64 KiB is approximately 8 MiB of payload per queue, plus object overhead, encoding/pipe/socket buffers, in-flight processing, waiting producers and lingering old-session work. Decoded objects may exceed their wire size. Bound producer concurrency and await sends sequentially rather than creating unlimited waiting tasks. Bounded queues are not a hard process-memory limit.

Slow consumers block receive-channel writes and pause decoding; backpressure propagates through the pipe and TCP to the peer. The incomplete-frame byte limit checks unconsumed partial frames, not complete messages or total memory. While the channel is blocked, the incomplete-frame timer does not run and the byte check must wait for decoding to resume. Receive idle timeout suits periodic traffic; keep it at zero when silence is valid and choose a partial-frame timeout from the protocol's transfer deadline. KeepAlive uses positive milliseconds, rounded up to seconds on modern .NET; netstandard2.0 IOControl retains milliseconds.

A caller timeout for WaitForConnectedAsync cancels only that wait. Lifetime cancellation/disposal stops the connection and cancels future waits too, even without a caller token (#65). Plain SendAsync guarantees neither socket write nor peer acknowledgement: queued entries can continue in a new session, but a failed dequeued send cannot be treated as delivered (#67). Reliable business delivery needs protocol ACKs, retries and deduplication. Session-bound send success also means only that the entire frame reached the local socket.

Session-aware messaging (advanced)

For typical business traffic that tolerates replay, SendAsync (completes on enqueue; queued messages continue on the next session) + GetMessages (a stable stream across reconnects) is enough. Some protocols have strict session boundaries — re-handshake after reconnect, no replay of old-session messages, protocol timers starting when the frame is actually written (HSMS Select/T3/T6/T8). For those there is an optional capability interface ISessionAwareStreamConnection<TMessage> (implemented by StreamConnection<TMessage>; upper layers depending on the interface abstraction detect it with is):

if (connection is ISessionAwareStreamConnection<MyMessage> sessionAware)
{
    long id = sessionAware.CurrentSessionId;   // allocated per established TCP session, monotonic, never reused; 0 when no session

    // Completes only after the whole frame is handed to the local socket;
    // session ends before that → SessionExpiredException, the message is never replayed
    await sessionAware.SendInSessionAsync(id, message, ct);

    // Receive view: every message carries the session it arrived on
    // (late deliveries from an old session's decoder still carry the old id)
    await foreach (var m in sessionAware.GetSessionMessages(ct))
        Handle(m.SessionId, m.Message);
}

Semantics at a glance:

Id linearization: reading CurrentSessionId from a ConnectionChanged callback or after WaitForConnectedAsync always yields a valid id (allocation happens before Connected becomes visible); it is already zero when a non-Connected state becomes visible.
"Written" means: the whole frame has been handed to the local socket (kernel buffer), peer ACK not included — the best signal available at the application layer. When the task fails, treat the remote outcome as unknown (the peer may have received part or all of the bytes); reconcile via your protocol's transaction correlation, idempotency or recovery flow.
Cancellation commit point: cancelling before the send worker claims the entry cancels the task and the message is never sent; cancelling after the claim (frame write in progress) has no effect — cancelling one message neither tears frames nor kills the connection.
Relation to SendAsync: both share one FIFO (serialized in enqueue order); the plain SendAsync cross-session replay behavior is unchanged. Session-bound sends never transfer to a new session on any failure path.
The two receive APIs are not a broadcast: GetMessages and GetSessionMessages are competing consumer views of the same channel — enumerating both splits messages between them; pick one.

Diagnostics & debugging

FrameError — frame-level diagnostics

When the peer sends bad data, the FrameError event hands you the offending bytes and the reason directly — no more eyeballing HEX dumps:

client.FrameError += (_, e) =>
{
    // e.Kind: DecodeFailed (frame intact, payload parsing failed)
    //         DiscardedByResync (bytes discarded as noise by the framer)
    //         IncompleteFrameOverflow (incomplete-frame buffer over its limit)
    //         IncompleteFrameTimeout (incomplete frame timed out waiting for the rest)
    // e.Bytes: copied; safe to retain long-term
    // e.Exception: the original exception for DecodeFailed
    // e.SessionId: the session whose decoder detected the error (same ID space as
    //              CurrentSessionId/SessionMessage.SessionId; a delayed event from
    //              an old session keeps the old ID after reconnecting)
    // e.ObservedByteCount / e.IsTruncated: the originally observed byte count and
    //              whether Bytes is truncated. Timeout/overflow snapshots are capped
    //              at 8 KB — when IsTruncated is true, Bytes is only a prefix and
    //              ObservedByteCount is the real size
    Console.WriteLine($"[session {e.SessionId}] [{e.Kind}] {Convert.ToHexString(e.Bytes.Span)}" +
        (e.IsTruncated ? $" (first {e.Bytes.Length}/{e.ObservedByteCount} bytes)") : "") +
        $" {e.Exception?.Message}");
};

The policy for payload decode failures is set by StreamConnectionOptions.DecodeErrorPolicy:

Policy	Behavior
`Disconnect` (default)	Reconnect — after content corruption the stream state is usually untrustworthy
`SkipFrame`	Drop the bad frame and continue; suits noisy lines

RawBytesReceived / RawBytesSent — raw byte taps

Full output from the socket layer (including noise bytes later discarded by the framer); the send side fires per chunk actually written (bytes already on the wire remain visible after a partial-send failure). Memory contract: callback arguments are slices of internal buffers, valid only during the synchronous callback — copy them if you need to retain; handler exceptions are isolated and never affect the session.

Incomplete-frame protection

A peer that announces a huge frame but never completes it (or a stream with STX but no closing ETX) would hold the buffer forever. Two complementary guards:

MaxIncompleteFrameBufferBytes (default = frame limit + 4 KB) — a cap in bytes: exceeding it disconnects, blocking memory-flooding peers;
IncompleteFrameTimeoutMs (default 0 = off) — a cap in time: if a started frame receives no further bytes for this long, the session is torn down and FrameError reports IncompleteFrameTimeout with a buffer snapshot capped at 8 KB.

The incomplete-frame timeout only counts while a frame is actually in progress: an idle connection with an empty buffer never trips it (each received byte resets it; it clears once a whole frame is cut). It complements ReceiveIdleTimeoutMs (see below) — the latter also times total silence, which suits protocols with periodic traffic; for protocols that may stay idle for a long time but must not tolerate a stalled half-frame (e.g. HSMS T8), use the incomplete-frame timeout. Note its scope is "waiting for further bytes from the network": when ReceiveQueueCapacity is set and the consumer stalls completely, the decode loop blocks on the message-channel write and the timeout does not tick during that period (the memory guard MaxIncompleteFrameBufferBytes still applies).

Built-in metrics

Connections ship System.Diagnostics.Metrics instruments (meter name StreamFrame, tag endpoint) with zero external dependencies — subscribe with a MeterListener or OpenTelemetry in production; unsubscribed, each record costs nanoseconds:

Instrument	Kind	Meaning
`streamframe.frames_sent` / `frames_received`	Counter	Business frames sent/received
`streamframe.bytes_sent` / `bytes_received`	Counter	Bytes sent/received (framing bytes/noise included)
`streamframe.reconnects`	Counter	Reconnect count
`streamframe.session_duration`	Histogram	Lifetime of one TCP session (seconds)
`streamframe.send_queue_length`	Histogram	Send-queue watermark (sampled per enqueue)

// Minimal subscription (example): OTel's MeterProvider.AddMeter("StreamFrame") hooks the full ecosystem
using var listener = new MeterListener { InstrumentPublished = (i, l) => l.EnableMeasurementEvents(i) };
listener.SetMeasurementEventCallback<long>((instrument, value, _, _) => { /* report */ });
listener.Start();

The netstandard2.0 target gets the same API via the System.Diagnostics.DiagnosticSource package (works on .NET Framework).

Logging (optional)

Pass an ILogger when constructing the connection to route internal events (connect retries, session faults, user-callback exceptions) to your logs — no more silence in production:

var conn = new StreamConnection<XDocument>(..., logger: loggerFactory.CreateLogger("StreamFrame"));

Without one, no logging happens (zero-dependency usage).

Liveness & heartbeat pattern

KeepAlive is disabled by default. When enabled, KeepAliveTimeMs / KeepAliveIntervalMs must be positive milliseconds (defaults: 30000 / 5000). Modern .NET Socket options use whole seconds, rounded up (1ms → 1s, 1500ms → 2s, defaults → 30s / 5s); the netstandard2.0 SIO_KEEPALIVE_VALS path retains millisecond granularity.

For production, enable TcpKeepAlive = true; an application-level heartbeat combined with ReceiveIdleTimeoutMs (3× the heartbeat period, tolerating 1-2 lost beats) is an even stronger combo. The framework does not ship a heartbeat (the message shape belongs to your protocol) — the pattern:

var options = new StreamConnectionOptions { ReceiveIdleTimeoutMs = 1500 };

// One side beats periodically; the other echoes PONG on any message (bidirectional
// traffic resets both idle timers). Full runnable sample: demo scenario 4.
_ = Task.Run(async () =>
{
    while (!cts.IsCancellationRequested)
    {
        await client.SendAsync("PING", cts.Token);
        await Task.Delay(TimeSpan.FromMilliseconds(500), cts.Token); // ~1/3 of the timeout
    }
});

When the peer dies silently (power loss / unplugged cable, no FIN/RST), the idle timeout kills the session and auto-reconnect kicks in; if the peer is unreachable, reconnection keeps failing and the state stays in Connecting/Retry.

Choosing between the two receive timeouts: ReceiveIdleTimeoutMs demands periodic traffic — use it when the protocol has heartbeats/periodic reports (it also covers half-open connections); when the protocol may stay silent for long stretches (traffic only expected while a frame is in progress), use IncompleteFrameTimeoutMs instead — silence is not a fault, only a stalled half-frame is.

Performance

The benchmarks cover same-machine TCP loopback one-way transfer and sequential round trips, with both Framers, 64B/1KB/64KB payloads, byte arrays and two string encoders. Both transports encode, frame, materialize and validate complete contents. One-way timing ends after reception completes; RTT waits for every echo. Commands, raw results and limitations are in the benchmark guide.

The direct TCP control still omits queues, Pipe and general stream assembly, so its difference cannot isolate framework overhead. Earlier claims of “faster than raw TCP”, “3–4× large-message cost” and “only 20–30% byte-payload overhead” used mismatched workloads and have been withdrawn. A single-machine, single-launch measurement does not establish a general performance advantage.

Streaming encode removes one whole-payload copy from the payload buffer into the frame buffer. Codecs, sockets, buffer growth and message materialization still copy or allocate; this is not end-to-end zero-copy. XML, framing, metrics and session features have independent benchmarks and should be measured for the intended application.

Large-message guide (≥64KB)

For 64KB-class messages, examine codec intermediate arrays, message materialization and transport costs separately (measurement contract in bench/README.en.md). Three recommendations:

Use the span-based write overload in your codec — Encoding.GetBytes(ReadOnlySpan<char>, IBufferWriter<byte>) produces no intermediate array (the naive style adds a full-size allocation + copy per message):

// Recommended: zero intermediate arrays
public void Encode(string message, IBufferWriter<byte> writer, CancellationToken ct)
    => Encoding.UTF8.GetBytes(message.AsSpan(), writer);
// Not recommended: writer.Write(Encoding.UTF8.GetBytes(message));  // full-size byte[] per message

Pick byte[] / ReadOnlyMemory as the message type: the ASCII string payload in this benchmark produces about 2× its payload size in UTF-16 character data; byte messages also need an owning codec, which does not imply zero-copy or zero allocation;
Keep streaming encode on (default, UseStreamingEncode) — send buffers now start at the previous frame's size (adaptive, capped at 1MB).

Supported frameworks

Package target	Runtime
`net10.0` (recommended, LTS until 2028-11)	.NET 10
`net8.0` (LTS until 2026-11)	.NET 8
`netstandard2.0`	.NET Framework 4.6.2+, Unity, Mono, ...

The netstandard2.0 asset is validated by the full net48 test suite (real TCP loopback); TCP KeepAlive parameters are set via SIO_KEEPALIVE_VALS on .NET Framework. CI runs all tests on Ubuntu (net8/net10) and Windows (net48).

Dependencies

System.IO.Pipelines
System.Threading.Channels
Microsoft.Extensions.Logging.Abstractions (logging abstractions only — no logging framework is pulled in)

Tests & samples

dotnet build StreamFrame.slnx
dotnet test
dotnet run --project samples/StreamFrame.Demo          # 5-scenario end-to-end demo
dotnet run -c Release --project bench/StreamFrame.Benchmarks   # benchmarks (select filters using bench/README.en.md)
dotnet test -f net8.0 --collect:"XPlat Code Coverage"  # coverage (CI also collects & summarizes)

NativeAOT validation

The non-required aot CI check separately validates NativeAOT publishing/trimming analysis and execution of the linux-x64 native binary. The TCP loopback smoke verifies two session-bound sends, one ordinary send, and message contents; controlled faults verify nonzero exits. A successful publish alone does not prove runtime success. See the AotSmoke guide for commands, deadlines, fault modes, and future transport coverage under #55.

Project layout

src/StreamFrame/                  # core library (no business dependencies)
src/StreamFrame.Protocols.Xml/    # XML message driver (sample codec)
test/StreamFrame.Tests/           # xUnit tests
samples/StreamFrame.Demo/         # console end-to-end demo
bench/StreamFrame.Benchmarks/     # BenchmarkDotNet benchmarks

License

MIT

Navigation Menu

Repository navigation

README.en.md

README.en.md

StreamFrame

Why StreamFrame

Installation

Quick start

Concepts

Framing (`IFramer`) — slicing frames out of a byte stream

Codec (`ICodec<T>`) — parsing/writing the payload inside a frame

Connection (`IStreamConnection<T>`) — the transport

Buffer ownership and concurrency

Production configuration starting point

Session-aware messaging (advanced)

Diagnostics & debugging

FrameError — frame-level diagnostics

RawBytesReceived / RawBytesSent — raw byte taps

Incomplete-frame protection

Built-in metrics

Logging (optional)

Liveness & heartbeat pattern

Performance

Large-message guide (≥64KB)

Supported frameworks

Dependencies

Tests & samples

NativeAOT validation

Project layout

License

FilesExpand file tree

README.en.md

Latest commit

History

README.en.md

File metadata and controls

StreamFrame

Why StreamFrame

Installation

Quick start

Concepts

Framing (IFramer) — slicing frames out of a byte stream

Codec (ICodec<T>) — parsing/writing the payload inside a frame

Connection (IStreamConnection<T>) — the transport

Buffer ownership and concurrency

Production configuration starting point

Session-aware messaging (advanced)

Diagnostics & debugging

FrameError — frame-level diagnostics

RawBytesReceived / RawBytesSent — raw byte taps

Incomplete-frame protection

Built-in metrics

Logging (optional)

Liveness & heartbeat pattern

Performance

Large-message guide (≥64KB)

Supported frameworks

Dependencies

Tests & samples

NativeAOT validation

Project layout

License

Framing (`IFramer`) — slicing frames out of a byte stream

Codec (`ICodec<T>`) — parsing/writing the payload inside a frame

Connection (`IStreamConnection<T>`) — the transport