TCP Request-Reply Protocol

Use when: Device communicates over TCP/IP with a binary or text protocol where you send a command and wait for a response.

Examples: intrusion panels with an ASCII command protocol, access controllers with a binary or AES-encrypted protocol, and most request/response field devices.

Framework provides: TCP socket, PipeReader buffering, request-reply orchestration, retry, timeout, disconnect handling.

You implement: Frame parsing, response matching, protocol logic.

---

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                        Your Code                                 │
├─────────────────────────────────────────────────────────────────┤
│  MyFrame : FrameBase         ← frame structure + matching       │
│  MyFrameParser : IFrameParser ← byte stream → frames            │
│  Protocol : TcpProtocolBase   ← business logic + Send()         │
│  Things (Panel, Door, etc.)   ← device tree                     │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│                     Framework (automatic)                        │
├─────────────────────────────────────────────────────────────────┤
│  TcpTransportBase            ← connection management            │
│  ProtocolChannel             ← request-reply orchestration      │
│  TcpConnection               ← socket + PipeReader              │
└─────────────────────────────────────────────────────────────────┘

---

Step-by-Step Implementation

1. Configure transport in adapter-registration.yaml

transport:
  type: tcp
  protocol:
    address_type: byte                    # device address type (byte, ushort, uint, string)
    frame_type: "MyAdapter.Communication.MyFrame"
    event_type: "MyAdapter.Communication.MyEvent"

2. Define your Frame type

// Communication/MyFrame.cs
using Pq.Adapters.Framework.Communication.Tcp;

public sealed class MyFrame : FrameBase
{
    public required byte Type { get; init; }
    public required byte[] Payload { get; init; }

    // Pre-built bytes for outgoing requests (null for parsed incoming)
    internal byte[]? RawPacket { get; init; }

    // Serialize for transmission
    public override byte[] Serialize() =>
        RawPacket ?? throw new InvalidOperationException("Not a request frame");

    // Match response to request
    public override bool IsResponseTo(FrameBase request) =>
        request is MyFrame req && Type == req.Type;

    // Factory for outgoing requests
    internal static MyFrame CreateSend(byte type, byte[] packet) => new()
    {
        Type = type,
        Payload = [],
        RawPacket = packet,
    };
}

For protocols with explicit busy response codes, override MatchResponse and return MatchResult.Busy for those codes. ProtocolChannel then retries using SendOptions.RetryPolicy.

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

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

3. Implement frame parser

// Communication/MyFrameParser.cs
using System.Buffers;
using Pq.Adapters.Framework.Communication.Tcp;

public sealed class MyFrameParser : IFrameParser<MyFrame>
{
    public bool TryParse(ref ReadOnlySequence<byte> buffer, out MyFrame frame)
    {
        frame = null!;

        // Need at least header
        if (buffer.Length < 3)
            return false;

        // Read length from header
        var reader = new SequenceReader<byte>(buffer);
        reader.TryReadBigEndian(out short length);

        // Need full frame
        if (buffer.Length < length)
            return false;

        // Parse frame
        reader.TryRead(out var type);
        var payload = new byte[length - 3];
        reader.TryCopyTo(payload);
        reader.Advance(payload.Length);

        frame = new MyFrame { Type = type, Payload = payload };
        buffer = buffer.Slice(length);  // consume bytes
        return true;
    }
}

4. Write Protocol using TcpProtocolBase

// Communication/Protocol.cs
using Pq.Adapters.Framework.Communication;
using Pq.Adapters.Framework.Communication.Protocol;
using Pq.Adapters.Framework.Events;

public partial class Protocol : TcpProtocolBase<MyEvent, MyFrame>
{
    private static readonly TimeSpan CommandTimeout = TimeSpan.FromSeconds(5);

    public Protocol(Transport transport, ILogger<Protocol> logger) : base(logger)
    {
        Transport = transport;
    }

    internal Transport Transport { get; }

    // --- Simple command ---
    public async Task<bool> Arm(int partition, CancellationToken ct)
    {
        var request = MyFrame.CreateSend(0x01, BuildPacket(0x01, [(byte)partition]));
        var response = await Send(request, CommandTimeout, ct);
        return response.Payload[0] == 0x00;
    }

    // --- With retry ---
    public async Task<byte[]?> ReadMemory(ushort address, CancellationToken ct)
    {
        var request = MyFrame.CreateSend(0x10, BuildPacket(0x10, BitConverter.GetBytes(address)));
        try
        {
            var response = await Send(request, new SendOptions
            {
                Timeout = TimeSpan.FromSeconds(2),
                RetryPolicy = RetryPolicy.Exponential(3, TimeSpan.FromMilliseconds(100))
            }, ct);
            return response.Payload;
        }
        catch (TimeoutException)
        {
            Logger.LogWarning("ReadMemory timeout");
            return null;
        }
    }

    // --- Bulk upload ---
    public async Task<int> UploadCredentials(IReadOnlyList<Credential> creds, CancellationToken ct)
    {
        var frames = creds.Select(c => MyFrame.CreateSend(0x20, EncodeCredential(c))).ToList();
        var result = await SendBulk(frames, new BulkOptions
        {
            RetryPolicy = RetryPolicy.Fixed(2, TimeSpan.FromMilliseconds(50))
        }, ct);
        return result.SuccessCount;
    }

    // --- Multi-packet logical operation ---
    internal Task<TResult> RunSession<TResult>(Func<CancellationToken, Task<TResult>> operation, CancellationToken ct)
        => ExclusiveSession(operation, ct);

    // --- Unsolicited events (alarms, status) ---
    protected override IEnumerable<MyEvent> TransformToEvents(MyFrame frame)
    {
        if (frame.Type == 0xE0)
            yield return MyEvent.Parse(frame.Payload);
    }

    // --- Connect/Disconnect ---
    internal async Task<bool> Connect(Panel panel, IEventRouter<MyEvent> router, CancellationToken ct)
    {
        var channel = await Transport.ConnectChannel(
            panel.Address,
            panel.IpAddress,
            panel.Port,
            router,
            ct);
        AttachChannel(channel, router);

        // Handshake, login, etc.
        return await Authenticate(panel.Password, ct);
    }

    internal async Task Disconnect(byte address)
    {
        DetachChannel();
        await Transport.Disconnect(address);
    }

    private byte[] BuildPacket(byte type, byte[] payload)
    {
        var length = (short)(3 + payload.Length);
        return [
            (byte)(length >> 8), (byte)(length & 0xFF),
            type,
            ..payload
        ];
    }
}

5. Implement Thing with IDeviceConnection

// Things/Panel.cs
public partial class Panel : IDeviceConnection
{
    public Task<bool> Connect(ConnectionContext context, CancellationToken ct)
        => Protocol.Connect(this, EventRouter, ct);

    public async Task Disconnect(CancellationToken ct)
    {
        await Protocol.Disconnect(Address);
    }
}

---

Retry Design Checklist

Before finishing protocol implementation, fill this checklist for every Send/SendBulk path:

1. Operation map - Split operations into read/poll and mutating (write/enroll/delete/action).
2. Busy semantics - Document what vendor busy codes mean (not processed, processing, unknown) per operation class.
3. Policy declaration - RetryPolicy.None is the compliant one-attempt default. Every enabled or vendor-required retry has an operation-specific RetryPolicy.*, rationale, and evidence.
4. Bounds and cancellation - Retries are bounded (maxAttempts and optional cap) and every delay/send honors ct.
5. Starvation guard - On shared channels, retry timing does not block history/status polling cadence.

Use this lightweight table in PR notes or adapter docs:

Operation	Mutating	RetryPolicy	Why safe	Evidence	Notes
ReadMemory	No	Exponential(4,100ms,500ms)	read-only	vendor SDK sec. 4.2	busy tolerated
UnlockDoor	Yes	None	duplicate unsafe	vendor SDK sec. 7.1	manual retry only

---

Key APIs

What you need	How to do it
Simple send	await Send(request, timeout, ct)
Send with retry	await Send(request, new SendOptions { RetryPolicy = ... }, ct)
Fire-and-forget	await Send(request, new SendOptions { NoReply = true }, ct)
Bulk upload	await SendBulk(frames, options, ct)
Multi-packet logical operation	await ExclusiveSession(operation, ct) or an adapter-local wrapper
High priority	new SendOptions { Priority = Priority.High }
Handle events	Override TransformToEvents(frame)
Connect	Transport.ConnectChannel() then AttachChannel()
Disconnect	DetachChannel() then Transport.Disconnect()

---

When to use this vs other archetypes

Archetype	Use when
tcp-request-reply (this)	Binary/text TCP protocol, request-response pattern
native-sdk-shared-transport	Vendor SDK with callbacks (C#/.NET wrapper)
gateway-grpc	gRPC streaming API
http-streaming	REST API with polling or webhooks
generated-only	Pure cloud integration, no device protocol

---

Full reference

See concepts/tcp-transport.md for complete API reference, matching strategies, exclusive sessions, and common mistakes.