TCP Transport¶

Start here: For overall communication architecture, see Communication Overview.

The framework provides a ready-made TCP transport for adapters communicating with devices over TCP/IP. You provide framing, response matching, and protocol logic. The framework handles sockets, buffering, backpressure, disconnect propagation, and the serialized request-response channel.

Architecture¶

+---------------------------------------------------------------------+
|                    Framework TCP Transport                            |
+---------------------------------------------------------------------+
|                                                                       |
|  Transport (singleton, per adapter host)                              |
|  +------------------------------------------------------------------+|
|  | SingleTcpTransportBase<TFrame, TEvent>                           ||
|  |   ConnectChannel(host, port, router, ct, ...) -> channel         ||
|  +------------------------------------------------------------------+|
|       |                                                               |
|       | one channel per device connection                            |
|       v                                                               |
|  +------------------------------------------------------------------+|
|  | ProtocolChannel<TFrame> where TFrame : FrameBase                 ||
|  |                                                                  ||
|  |  Send(request, options, ct)                                      ||
|  |  SendBulk(...)                                                   ||
|  |  OnUnmatchedFrame / OnDisconnected                               ||
|  +---------------------------+--------------------------------------+|
|                              |                                       |
|                              v                                       |
|  +------------------------------------------------------------------+|
|  | TcpConnection<TFrame>                                            ||
|  |                                                                  ||
|  |  Socket ──> PipeReader ──> IFrameParser ──> Channel<TFrame>      ||
|  |                                                                  ||
|  |  Write(bytes, ct)            = raw socket write                  ||
|  |  Frames                      = parsed incoming frames            ||
|  +------------------------------------------------------------------+|
|                                                                       |
+---------------------------------------------------------------------+

Key properties:

ProtocolChannel owns request-response — Send() serializes a frame, waits for matching response
Matching via FrameBase — your frame type defines how requests pair with responses
Unmatched frames dispatched — unsolicited device events go to OnUnmatchedFrame / TransformToEvents()
Zero-copy reads via System.IO.Pipelines (pooled buffers)
Backpressure — if adapter can't process frames fast enough, TCP window closes
Thread-safe writes — transport serializes writes internally
Automatic disconnect signaling — framework handles reconnection via ConnectionStateTracker
Channel poison on terminal timeout/cancel/disconnect — caller must reconnect after session loss

What You Provide vs What Framework Provides¶

Concern	Framework	Adapter
TCP socket lifecycle	Yes	-
Read loop + buffering	Yes (PipeReader)	-
Frame parsing (bytes → frame)	-	Yes (`IFrameParser`)
Frame serialization (frame → bytes)	-	Yes (`FrameBase.Serialize()`)
Request-response matching	Yes (ProtocolChannel)	Yes (`FrameBase.MatchResponse()` / `IsResponseTo()`)
Request-response dispatch	Yes (`ProtocolChannel.Send()`)	-
Unmatched frame routing	Yes	Yes (handler logic)
Write serialization	Yes (SemaphoreSlim)	-
Disconnect detection + reconnect	Yes	-
Keepalive heartbeat	Yes (idle-based)	Yes (frame factory)
Frame interpretation	-	Yes (Protocol)
Command sequencing	-	Yes (Protocol via CommandQueue)
Domain event conversion	-	Yes (Protocol)

Core Concept: FrameBase¶

All TCP frame types must inherit FrameBase. A frame is one message between the adapter and the device. This abstract class has four responsibilities:

Serialize — convert a request frame to bytes for transmission
ExpectsResponse — declare whether a request is request-response or fire-and-forget
MatchResponse / IsResponseTo — decide how an incoming frame relates to the pending request: the response, a transient busy response, a further part of the reply, or a frame that arrives independently of any request
ToString — render the frame as its documented packet name followed by key addressing fields, for readable trace logs

public abstract class FrameBase
{
    public virtual bool ExpectsResponse => true;

    /// <summary>
    /// Serializes this frame to bytes for transmission.
    /// </summary>
    public abstract byte[] Serialize();

    public virtual MatchResult MatchResponse(FrameBase request)
        => IsResponseTo(request) ? MatchResult.Match : MatchResult.NoMatch;

    /// <summary>
    /// Checks if this frame is a response to the given request frame.
    /// </summary>
    public abstract bool IsResponseTo(FrameBase request);

    /// <summary>
    /// Packet name from the protocol documentation plus key fields. Never raw hex only —
    /// the transport already trace-logs the hex bytes.
    /// </summary>
    public abstract override string ToString();
}

How matching works¶

When you call channel.Send(request):

Framework registers a pending entry (before writing, to avoid race conditions)
Calls request.Serialize() → sends bytes over TCP
TcpConnection read loop parses incoming frames
ProtocolChannel matches each frame against the active pending request via MatchResponse()
MatchResponse() returns one verdict per frame (table below)
The verdict decides whether the frame completes the Send() caller, is forwarded to the unmatched-frame handlers, or both

Verdict	Meaning	What the channel does
`Match`	The frame is the final response.	Completes the `Send()` caller with that frame.
`Busy`	The frame is the response, and it says the device is busy.	Applies the configured `RetryPolicy`. No exception is thrown for normal busy control flow.
`Continue`	The frame belongs to the pending request and more frames follow (for example a receipt before the data reply).	Forwards the frame to `OnUnmatchedFrame` at Trace level. The request keeps waiting.
`OutOfBand`	The frame does not belong to the request, but the device may send it at any time (poll, heartbeat, status push).	Forwards the frame to `OnUnmatchedFrame` at Trace level. The request keeps waiting.
`NoMatch`	The frame does not belong to the request and nobody expects it.	Forwards the frame to `OnUnmatchedFrame`. While a request waits, the channel logs a Warning.

Continue and OutOfBand do not restart the request timeout. The whole reply, including all Continue frames, must arrive within the request timeout.

Send(requestFrame)                          ProtocolChannel / TcpConnection
  │                                            │
  ├─ register pending                          ├─ read frame from parser
  ├─ serialize + write bytes                   ├─ frame.MatchResponse(pending)?
  └─ await response                            │    Match / Busy       → complete Send() caller
                                               │    Continue / OutOfBand → OnUnmatchedFrame, keep waiting
                                               │    NoMatch            → OnUnmatchedFrame + Warning

Choose OutOfBand for every frame the protocol defines as device-initiated. Choose NoMatch only for frames that indicate a mistake, such as a reply to a request that already timed out.

Matching strategies¶

Sequential protocol (one command at a time, e.g. Acme):

// All data packets share ctrl=0x09 — safe because CommandQueue serializes
public override bool IsResponseTo(FrameBase request) =>
    request is MyFrame req && req.Control == Control;

Type-based protocol (e.g. Acme — ACK for any command, events are separate):

public override bool IsResponseTo(FrameBase request) =>
    request is AcmeFrame req
    && req.Type != MessageType.EventNotification    // request is a command
    && Type != MessageType.EventNotification;       // response is not an event

Correlation-ID protocol (future — tagged requests):

public override bool IsResponseTo(FrameBase request) =>
    request is TaggedFrame req && req.Tag == Tag;

Busy-aware protocol (device returns explicit busy response code):

public override MatchResult MatchResponse(FrameBase request)
{
    if (!IsResponseTo(request))
        return MatchResult.NoMatch;

    return ResponseCode == 0xDD
        ? MatchResult.Busy
        : MatchResult.Match;
}

Receipt, device polls and data reply (the device confirms the request, polls the adapter on its own, then sends the data):

public override MatchResult MatchResponse(FrameBase request) => Type switch
{
    MessageType.Poll or MessageType.Status => MatchResult.OutOfBand, // device sends these at any time
    MessageType.Receipt => MatchResult.Continue,                     // the data reply follows
    _ => IsResponseTo(request) ? MatchResult.Match : MatchResult.NoMatch,
};

Adapter code answers device polls in the unmatched-frame path (TransformToEvents), not in MatchResponse.

Step-by-Step: Creating a TCP Adapter¶

1. Define transport in adapter.yaml¶

transport:
  type: tcp
  protocol:
    address_type: byte
    frame_type: "MyAdapter.Communication.MyFrame"
    event_type: "MyAdapter.Communication.MyEvent"

2. Define your frame type (inheriting FrameBase)¶

Your frame type has two roles:

Parsed incoming frame — created by IFrameParser, RawPacket is null
Outgoing request frame — created by CreateSend() factory, RawPacket holds pre-built bytes

using Pq.Adapters.Framework.Communication.Tcp;

namespace MyAdapter.Communication;

/// <summary>
/// Parsed protocol frame. Inherits FrameBase for request-response matching.
/// </summary>
public sealed class MyFrame : FrameBase
{
    /// <summary>
    /// Gets the message type code.
    /// </summary>
    public required byte Type { get; init; }

    /// <summary>
    /// Gets the payload data.
    /// </summary>
    public required byte[] Payload { get; init; }

    /// <summary>
    /// Gets pre-built packet bytes for outgoing requests. Null for parsed incoming frames.
    /// </summary>
    internal byte[]? RawPacket { get; init; }

    /// <inheritdoc/>
    public override byte[] Serialize() =>
        RawPacket ?? throw new InvalidOperationException("Frame was not created as a request");

    /// <inheritdoc/>
    public override bool IsResponseTo(FrameBase request) =>
        request is MyFrame req && req.Type == Type;

    /// <summary>
    /// Creates a request frame for sending.
    /// </summary>
    internal static MyFrame CreateSend(byte type, byte[] rawPacket) => new()
    {
        Type = type,
        Payload = [],
        RawPacket = rawPacket,
    };
}

Key pattern: The parser creates frames from incoming bytes (RawPacket stays null). The protocol creates request frames via CreateSend() with pre-built packet bytes. The same class serves both roles.

3. Implement IFrameParser¶

Tells the framework where one frame ends and the next begins in the TCP byte stream:

using System.Buffers;
using Pq.Adapters.Framework.Communication.Tcp;

namespace MyAdapter.Communication;

/// <summary>
/// Parses protocol frames from the TCP byte stream.
/// </summary>
public sealed class MyFrameParser : IFrameParser<MyFrame>
{
    /// <inheritdoc/>
    public bool TryParse(ref ReadOnlySequence<byte> buffer, out MyFrame frame)
    {
        frame = null!;

        // not enough data for a header?
        if (buffer.Length < 4)
            return false;

        // read length from first 2 bytes (example: big-endian uint16)
        var reader = new SequenceReader<byte>(buffer);
        reader.TryReadBigEndian(out short lengthShort);
        var totalLength = (ushort)lengthShort;

        // not enough data for full frame?
        if (buffer.Length < totalLength)
            return false;

        // parse the frame from buffer
        reader.TryRead(out var type);
        var payloadLength = totalLength - 3;
        var payload = new byte[payloadLength];
        for (var i = 0; i < payloadLength; i++)
            reader.TryRead(out payload[i]);

        frame = new MyFrame
        {
            Type = type,
            Payload = payload,
        };

        buffer = buffer.Slice(totalLength);  // advance past consumed bytes
        return true;
    }
}

Rules:

Return false when buffer has insufficient data (framework reads more bytes automatically)
On success, advance buffer past consumed bytes via buffer.Slice()
Never throw — return false for malformed data and advance past bad bytes
The source generator auto-discovers and registers this class in DI

4. Write Protocol (using ProtocolChannel for request-response)¶

The generated Protocol extends TcpProtocolBase<MyEvent, MyFrame> when frame_type is declared in adapter-registration.yaml. You write the partial class that uses the attached channel:

using Microsoft.Extensions.Logging;
using Pq.Adapters.Framework.Communication.Protocol;
using Pq.Adapters.Framework.Communication.Tcp;
using Pq.Adapters.Framework.Events;

namespace MyAdapter.Communication;

/// <summary>
/// Device protocol implementation.
/// </summary>
public partial class Protocol
{
    /// <summary>
    /// Connects to a device via the framework TCP transport.
    /// </summary>
    internal async Task<bool> Connect(
        MyDevice device,
        IEventRouter<MyEvent> router,
        CancellationToken ct)
    {
        var connected = false;

        // 1. Establish TCP channel (framework manages TCP + request-response orchestration)
        var channel = await Transport.ConnectChannel(
            device.IpAddress,
            device.Port,
            router,
            ct);
        AttachChannel(channel, router);

        try
        {
            // 2. Protocol handshake using Send()
            var helloPacket = BuildPacket(0x01, []);
            var request = MyFrame.CreateSend(0x01, helloPacket);
            var response = await Send(request, TimeSpan.FromSeconds(5), ct);

            if (response.Payload[0] != 0x00)
            {
                Logger.LogWarning("Handshake rejected");
                return false;
            }

            connected = true;
            Logger.LogInformation("Connected and authenticated");
            return true;
        }
        finally
        {
            if (!connected)
                await channel.Disconnect();
        }
    }

    /// <summary>
    /// Disconnects from the device.
    /// </summary>
    internal async Task Disconnect(CancellationToken ct)
    {
        if (Channel != null)
            await Channel.Disconnect();
        DetachChannel();
    }

    /// <summary>
    /// Sends a command and waits for the matching response.
    /// </summary>
    internal async Task<byte[]?> SendCommand(byte type, byte[] payload, CancellationToken ct)
    {
        if (Channel == null)
            return null;

        var packet = BuildPacket(type, payload);
        var request = MyFrame.CreateSend(type, packet);

        try
        {
            var response = await Send(request, TimeSpan.FromSeconds(2), ct);
            return response.Payload;
        }
        catch (OperationCanceledException)
        {
            Logger.LogWarning("Command timeout for type {Type}", type);
            return null;
        }
    }

    /// <summary>
    /// Handles frames that don't match any pending Send() — device events.
    /// </summary>
    protected override IEnumerable<MyEvent> TransformToEvents(MyFrame frame)
    {
        Logger.LogDebug("Unsolicited frame: type={Type}", frame.Type);
        return [];
    }

    /// <summary>
    /// Builds a protocol packet from type and payload.
    /// </summary>
    private byte[] BuildPacket(byte type, byte[] payload)
    {
        // your protocol's packet framing logic
        var packet = new byte[3 + payload.Length];
        var length = (ushort)packet.Length;
        packet[0] = (byte)(length >> 8);
        packet[1] = (byte)(length & 0xFF);
        packet[2] = type;
        Array.Copy(payload, 0, packet, 3, payload.Length);
        return packet;
    }
}

5. Implement Device Thing¶

using Pq.Adapters.Framework.Communication;

namespace MyAdapter;

/// <summary>
/// Device thing implementing framework connection interface.
/// </summary>
public partial class MyDevice : IDeviceConnection
{
    /// <inheritdoc/>
    public async Task<bool> Connect(ConnectionContext context, CancellationToken ct) =>
        await Connection.Protocol.Connect(this, Connection, ct);

    /// <inheritdoc/>
    public async Task Disconnect(CancellationToken ct) =>
        await Connection.Protocol.Disconnect(ct);
}

Connection Flow Summary¶

1. Framework calls IDeviceConnection.Connect()
      │
2. Protocol calls Transport.ConnectChannel(host, port, router, ct)
      │  → creates TcpConnection and ProtocolChannel (identified by the channel instance)
      │  → carries the event router, wires OnDisconnected for retry
      │
3. Protocol calls AttachChannel(channel, router)
      │  → unmatched frames can flow through TransformToEvents / OnUnmatchedFrame
      │
4. Protocol does handshake using Send(frame, timeout, ct)
      │  → framework serializes, sends, waits for matching response
      │
5. Connection is live — adapter sends commands via Send()
      │  → unsolicited events flow through TransformToEvents / OnUnmatchedFrame
      │
6. On disconnect:
      │  → read loop detects EOF/error
      │  → OnDisconnected fires → framework retry via ConnectionStateTracker
      │  → IDeviceConnection.Disconnect() called, then Connect() again

Keepalive¶

The framework no longer injects transport-level keepalive packets automatically. If your protocol requires keepalive, implement it explicitly in the protocol layer as a normal command or no-reply send. This keeps heartbeat semantics adapter-specific and avoids hidden traffic during reconnect and timeout scenarios.

API Reference¶

FrameBase (abstract, your frame inherits this)¶

Member	Description
`Serialize()`	Convert frame to bytes for transmission
`ExpectsResponse`	Declare whether the request expects a reply
`MatchResponse(FrameBase request)`	Rich matching result (`NoMatch`, `Match`, `Busy`, `Continue`, `OutOfBand`)
`IsResponseTo(FrameBase request)`	Boolean matching used by the default `MatchResponse`
`ToString()`	Documented packet name plus key fields, for trace logs

SingleTcpTransportBase¶

Singleton per adapter host. Every transport: type: tcp adapter's generated Transport class inherits this. A TCP connection is one device per socket; the transport identifies each connection by its own ProtocolChannel instance, so several device connections coexist on the one shared host transport without colliding on a key.

Method	Description
`ConnectChannel(host, port, router, ct, tcpOptions?, channelOptions?)`	Opens the connection and returns its `ProtocolChannel<TFrame>`; the caller owns and drives that channel

Close a connection through its own channel (channel.Disconnect()), which the transport observes and drops. There is no address-keyed Disconnect or GetChannel — a caller never reaches another device's connection.

ProtocolChannel where TFrame : FrameBase¶

One per device connection. Created and returned by Transport.ConnectChannel().

Member	Description
`Send(request, options?, ct)`	Request-response or no-reply send
`SendBulk(...)`	Sequential bulk send with stop-on-first-error semantics
`ExclusiveSession(operation, ct)`	Caller-driven exclusive lease for multi-packet logical operations
`Disconnect()`	Closes this connection; the owning transport drops exactly this channel
`OnUnmatchedFrame`	Hook for unsolicited frames not consumed by matching
`OnDisconnected`	Fires when the channel becomes terminally unusable
`TcpConnection`	Access to raw TCP transport when a protocol truly needs it
`IsConnected`	Whether the channel is still usable

TcpConnection where TFrame : FrameBase¶

Raw framed transport under ProtocolChannel.

Member	Description
`Write(data, ct)`	Fire-and-forget: send raw bytes (thread-safe)
`Frames`	`ChannelReader<TFrame>` with parsed incoming frames
`IsConnected`	Whether the connection is active
`OnDisconnected`	Callback (set by Transport, triggers framework retry)

IFrameParser where TFrame : FrameBase¶

Adapter implements this. Auto-discovered and registered by source generator.

Method	Description
`TryParse(ref buffer, out frame)`	Parse one frame from `ReadOnlySequence<byte>`, advance buffer on success

TcpConnectionOptions¶

Optional configuration passed to Transport.ConnectChannel() as tcpOptions.

Property	Default	Description
`FrameQueueCapacity`	256	Max buffered frames before backpressure
`PipeReaderOptions`	null	PipeReader configuration (buffer sizes, pool)
`ConnectTimeout`	10s	TCP connect timeout

ProtocolChannelOptions¶

Optional configuration passed to Transport.ConnectChannel() as channelOptions.

Property	Default	Description
`DefaultTimeout`	5s	Default response timeout when no per-send timeout is provided
`MaxQueueSize`	256	Buffered normal-priority requests
`QueueFullMode`	`Wait`	Behavior when the normal queue is full
`HighPriorityBurstLimit`	10	Fairness limit before one normal item is forced
`FollowUpQuietTime`	`Zero`	Mandatory gap after a no-reply follow-up/acknowledge frame before the next request is dispatched

For multi-step mandatory exchanges (e.g. a history/log drain that must run to completion on the wire), use the frame-typed subtype ProtocolChannelOptions<TFrame>, which adds a SequencePolicies list. See Sequence Policies below and compliance rule COMM-007.

ProtocolChannelOptions<TFrame> — Sequence Policies¶

Some packet types carry a protocol obligation: their response must be acknowledged and the request repeated until the device signals the end (GetLog → AckLog → GetLog → … → NoData). That obligation is a property of the protocol, not the call site, so it is declared once per channel instead of looped per call:

channelOptions: new ProtocolChannelOptions<XFrame>
{
    FollowUpQuietTime = interFrameGap,
    SequencePolicies = [GetLogSequence.Create(BuildFrame)],
}

A SequencePolicy<TFrame> is (request, response) → next step or null, consulted on the dispatch loop after every matched response. It must be pure decision logic — no blocking, no await, no I/O, no throwing. Returning an action makes the channel write the no-reply Acknowledge, honor FollowUpQuietTime, then send Continue; a null Continue (or a null action) ends the sequence.

While a sequence runs it occupies a single dispatch-queue slot — nothing interleaves between an ACK and the next request. Externally it is a plain Send that returns the final response; intermediate responses are forwarded, in order, to the unmatched-frame → event path after the sequence completes. SendOptions.Timeout applies per step. Define the policy as a static factory next to the packet definition, parameterized by the protocol's frame builder (which owns the packet-number counter). See Pattern 18: Multi-Step Protocol Sequences for the full behavior — lifecycle walkthrough, event delivery timing, per-step timeouts, failure modes — and compliance rule COMM-007 for the reviewer contract.

Exclusive Sessions¶

Use ExclusiveSession(...) when a caller-driven logical operation issues many request-reply packets and must not be interleaved with status polls or commands. Typical examples are device discovery, bulk configuration reads, and panel restore flows. A plain loop of await Send(...) is not enough for these operations: every awaited send releases the dispatch slot, so queued traffic can slip between two packets of the same logical operation.

ExclusiveSession takes the dispatch slot as one queued item, then holds it until the delegate completes. Inner Send(...) calls are unchanged; an ambient lease token lets the holder send on the held slot while all foreign sends wait. Foreign background sends wait indefinitely; a send belonging to an interactive command waits only for its send deadline, after which it stays unsent and the operator is answered Busy. A session shorter than that deadline is therefore invisible to operator commands.

Inside a TcpProtocolBase subclass, expose a narrow adapter-local wrapper when discovery or another capability needs to start a session from outside the protocol class:

internal Task<TResult> RunSession<TResult>(Func<CancellationToken, Task<TResult>> operation, CancellationToken ct)
    => ExclusiveSession(operation, ct);

Then wrap the whole logical operation once:

public static Task<AdapterDeviceCollection> Discover(Panel panel, Protocol protocol, CancellationToken ct)
    => protocol.RunSession(async leaseCt =>
    {
        var readers = await protocol.EnumerateReaders(leaseCt);
        var doors = await protocol.EnumerateDoors(leaseCt);
        return BuildTree(readers, doors);
    }, ct);

Do not use SendBulk inside an exclusive session. SendBulk is deliberately interruptible and fills idle gaps; the held dispatch loop does not pull bulk items. Single request-reply operations do not need a session.

SendOptions¶

Per-request configuration passed to Send(request, options, ct).

Property	Default	Description
`Timeout`	null (uses channel default)	Per-request timeout override
`RetryPolicy`	`RetryPolicy.None`	Retry behavior on transient failures
`NoReply`	false	Fire-and-forget mode (no response expected)
`Priority`	`Normal`	Queue priority (`Normal` or `High`)

RetryPolicy¶

Configures automatic retry on transient failures, including matched MatchResult.Busy responses. Not retried: OperationCanceledException, TimeoutException, disconnect.

// No retry (default)
RetryPolicy.None

// Fixed delay between attempts
RetryPolicy.Fixed(maxAttempts: 3, delay: TimeSpan.FromMilliseconds(100))

// Exponential backoff: 50ms → 100ms → 200ms → 400ms
RetryPolicy.Exponential(maxAttempts: 5, initialDelay: TimeSpan.FromMilliseconds(50))

// Capped exponential backoff: 100ms → 200ms → 400ms → 500ms (cap)
RetryPolicy.Exponential(
    maxAttempts: 5,
    initialDelay: TimeSpan.FromMilliseconds(100),
    maxDelay: TimeSpan.FromMilliseconds(500))

// Custom delay factory
new RetryPolicy(maxAttempts: 4, attempt => TimeSpan.FromMilliseconds(attempt * 50))

Use capped exponential when short retries are useful but unbounded growth would hurt command latency (for example, user-facing commands on temporarily busy controllers).

Priority Queues¶

ProtocolChannel maintains two queues:

High priority (unbounded) — urgent commands, status polls
Normal priority (bounded, backpressure) — bulk operations, background work

Fairness rule: after HighPriorityBurstLimit (default 10) consecutive high-priority items, one normal item is forced. This prevents starvation during access sync.

// High priority (e.g., user-initiated command)
await Send(request, new SendOptions { Priority = Priority.High }, ct);

// Normal priority (e.g., bulk credential upload)
await SendBulk(frames, new BulkOptions { Priority = Priority.Normal }, ct);

BulkOptions and SendBulk¶

For bulk operations (access sync, mass configuration):

var result = await SendBulk(frames, new BulkOptions
{
    Timeout = TimeSpan.FromSeconds(2),
    RetryPolicy = RetryPolicy.Fixed(2, TimeSpan.FromMilliseconds(50)),
    Priority = Priority.Normal
}, ct);

// Check results
if (result.IsSuccess)
    Logger.LogInformation("All {Count} items succeeded", result.SuccessCount);
else
    Logger.LogWarning("Bulk stopped at index {Index}: {Error}", 
        result.StoppedAtIndex, result.Items.Last().Exception?.Message);

Semantics:

Items sent sequentially (protocol constraint)
First failure stops the bulk (no partial success hidden)
BulkResult<TFrame> contains per-item outcomes

Practical Examples¶

Simple command (most common)¶

public async Task<bool> Arm(int partition, CancellationToken ct)
{
    var request = MyFrame.CreateSend(0x01, BuildPacket(partition));
    var response = await Send(request, TimeSpan.FromSeconds(5), ct);
    return response.Payload[0] == 0x00; // ACK
}

Command with capped exponential retry (unreliable device)¶

public async Task<byte[]?> ReadMemory(ushort address, CancellationToken ct)
{
    var request = MyFrame.CreateSend(0x10, BitConverter.GetBytes(address));
    try
    {
        var response = await Send(request, new SendOptions
        {
            Timeout = TimeSpan.FromSeconds(2),
            RetryPolicy = RetryPolicy.Exponential(
                maxAttempts: 3,
                initialDelay: TimeSpan.FromMilliseconds(100),
                maxDelay: TimeSpan.FromMilliseconds(300))
        }, ct);
        return response.Payload;
    }
    catch (TimeoutException)
    {
        Logger.LogWarning("ReadMemory timeout at 0x{Address:X4}", address);
        return null;
    }
}

Fire-and-forget (no response expected)¶

public async Task SendKeepalive(CancellationToken ct)
{
    var request = MyFrame.CreateSend(0xFF, []);
    await Send(request, new SendOptions { NoReply = true }, ct);
}

Bulk upload (access sync)¶

public async Task<int> UploadCredentials(IReadOnlyList<Credential> credentials, CancellationToken ct)
{
    var frames = credentials
        .Select(c => MyFrame.CreateSend(0x20, EncodeCredential(c)))
        .ToList();

    var result = await SendBulk(frames, new BulkOptions
    {
        Timeout = TimeSpan.FromSeconds(2),
        RetryPolicy = RetryPolicy.Fixed(2, TimeSpan.FromMilliseconds(50))
    }, ct);

    if (!result.IsSuccess)
        Logger.LogWarning("Upload stopped at {Index}/{Total}", result.StoppedAtIndex, credentials.Count);

    return result.SuccessCount;
}

High-priority command (user action)¶

public async Task<bool> EmergencyUnlock(int door, CancellationToken ct)
{
    var request = MyFrame.CreateSend(0x99, [(byte)door]);
    var response = await Send(request, new SendOptions
    {
        Timeout = TimeSpan.FromSeconds(1),
        Priority = Priority.High  // jumps queue
    }, ct);
    return response.Payload[0] == 0x00;
}

Common Mistakes¶

Mistake	Fix
Manually reading from `TcpConnection.Frames` in adapter protocol code	Use `ProtocolChannel` and `TransformToEvents()` / `OnUnmatchedFrame` instead
Using `TcpConnection` directly for request-response	Use `Send()` on `TcpProtocolBase` / `ProtocolChannel`
`IsResponseTo()` matches unsolicited events	Exclude event types in your matching logic (e.g. `Type != "nq"`)
`Serialize()` called on parsed frame (no RawPacket)	Use a factory method like `CreateSend()` for outgoing frames
Multiple `Send()` with same matching key without CommandQueue	Serialize commands through `ProtocolBase.Enqueue()` to avoid collisions
Multi-packet discovery/config read implemented as a plain `Send()` loop	Wrap the whole logical operation in one `ExclusiveSession(...)`
Returning `false` from `Connect()` after opening a channel without cleanup	Always disconnect partial sessions in `finally` so reconnect starts clean