Client connections

Most realtime needs don't need a connection. A deferred response to one request is request-driven streaming; server→client facts are client events; and producer-owned ordered outputs are streams. A connection is a stateful, bidirectional carrier for the cases those deliberately don't cover — when the conversation itself is stateful or latency-interactive. It does not add replay, ordering, or delivery guarantees to the contract it carries:

1. Interactive rate — client→server messages are frequent and per-message latency is part of the UX (collaboration ops, live input, telemetry ingest where batching's flush window is itself too slow).
2. The connection is the state — device links where connect/disconnect are business events, or the link fronts a stateful device protocol (the IoT gateway shape).
3. Server→client RPC — the server must address one specific connected client and await a result.

If a list-carrying batch command flushed every N items or M milliseconds would do, use that instead: one authorization, one validation pass, one transaction per batch, visible in the schema.

The kernel

Elarion.Connections (contracts in Elarion.Abstractions.Connections) is transport-neutral — the same seams serve a WebSocket adapter today and a raw TCP or datagram adapter tomorrow, and nothing in them assumes a socket exists at all:

Register the kernel once:

builder.Services.AddElarionConnections();   // registry + client-events bridge

The WebSocket endpoint

Elarion.Connections.AspNetCore ships the first adapter, shaped for device gateways: the endpoint owns accept, message reassembly (size-capped), registry lifecycle, and teardown; the app supplies an authenticator and a codec by subclassing WebSocketConnectionHandler. For device links, the provisioning chain behind the authenticator — pairing codes, per-device keys, the HMAC challenge/response — is device identity (Elarion.Devices).

// HmacChallengeVerifier + DevicePrincipal come from Elarion.Devices — pairing codes, key storage,
// and the constant-time verification are covered on the device identity page.
public sealed class GatewaySocketHandler(IActorSystem actors, HmacChallengeVerifier verifier)
    : WebSocketConnectionHandler {

    // In-socket challenge/response (HTTP-level token auth works too — the HttpContext is available).
    public override async ValueTask<ClientConnectionTicket?> AuthenticateAsync(
        WebSocketHandshakeContext handshake, CancellationToken ct) {
        var nonce = HmacChallengeVerifier.CreateNonce();
        await handshake.SendTextAsync($"challenge:{Convert.ToBase64String(nonce)}", ct);
        var answer = await handshake.ReceiveTextAsync(ct);         // "deviceId:base64(mac)"
        if (answer?.Split(':', 2) is not [var deviceId, var mac]) {
            return null;                                   // → PolicyViolation close, nothing registered
        }

        var principal = await verifier.VerifyAsync(deviceId, nonce, Convert.FromBase64String(mac), ct);
        if (principal is null) {
            return null;
        }

        return new ClientConnectionTicket {
            Principal = principal,
            PrincipalId = deviceId,                        // the registry indexes by this
            Metadata = new Dictionary<string, string> { ["channel"] = "telemetry" },
        };
    }

    public override IClientConnectionProtocol CreateProtocol(WebSocketClientConnection connection) =>
        new GatewayCodec(connection, actors);
}

// The codec: parse the device's frames, route into the digital-twin actor, answer over raw sends.
internal sealed class GatewayCodec(WebSocketClientConnection connection, IActorSystem actors)
    : IClientConnectionProtocol {
    public async ValueTask OnTextAsync(string message, CancellationToken ct) {
        var frame = GatewayFrame.Parse(message);           // your wire format
        var twin = actors.Get<IDeviceTwin>(connection.Connection.PrincipalId!);
        await twin.ApplyAsync(frame);                      // the actor serializes all channels + UI actions
    }
}

builder.Services.AddSingleton<GatewaySocketHandler>();

var app = builder.Build();
app.UseWebSockets();
app.MapElarionConnectionSocket<GatewaySocketHandler>("/gateway/ws",
    o => o.MaxMessageBytes = 64 * 1024);

Everything after the handshake is framework: the connection is minted (v7 id, Transport = "websocket"), registered, observed, pumped, and unregistered on every exit path — client close, abrupt death, oversized message (MessageTooBig), codec exception (logged, closed), host shutdown.

Required opening work

IClientConnectionObserver.OnConnectedAsync remains best-effort: observers are deliberately failure-isolated and suit presence/client-event projections. A codec that cannot process frames until it has completed required work (for example, attaching the authenticated link to a keyed device actor) implements IClientConnectionProtocol.OnOpenedAsync. It runs exactly once after registry registration and observer visibility, before the first inbound frame. It is awaited; a failure or cancellation closes the link, calls OnClosedAsync, and unregisters it without delivering a frame.

For tests, choose the lowest faithful tier: SimulatedClientConnection for kernel/observer work, InMemoryTcpLink (or loopback TCP) for a framed TCP lifecycle, and the separate Elarion.Connections.AspNetCore.Simulation package's WebSocketTestHost for a real Kestrel upgrade, handshake, codec, and close path.

Settings can vary per connection on one route: override ConfigureConnectionAsync(HttpContext, …) and return WebSocketConnectionSettings (size cap, idle window, keep-alive interval, transport tag — nulls inherit the endpoint options), resolved from the upgrade request (route values, query, headers) before the socket is accepted. And where the TCP adapter needs a runtime endpoint manager (ports are OS resources), WebSocket bindings-as-data are just routing: map one wildcard route (/gateway/{binding}/ws) and let the authenticator + ConfigureConnectionAsync consult the binding row per connection — an unknown or disabled binding returns null from the authenticator and the socket closes with nothing registered; a changed binding takes effect the next time the device connects (close its current connections via the registry to force it).

Talking back to one client

Look a connection up and use its sink — from a handler, an observer, or an actor turn:

// All channels of one device (or all tabs of one user):
foreach (var sink in registry.GetForPrincipal(deviceId)) {
    await ((WebSocketClientConnection)sink).SendTextAsync(frame, ct);   // raw leg, codec-owned format
}

// Request/reply into a client whose codec implements InvokeAsync:
var status = await sink.InvokeAsync<StatusQuery, StatusReply>(
    "status.get", new StatusQuery(), new ClientInvokeOptions { Timeout = TimeSpan.FromSeconds(5) }, ct);

InvokeAsync is deliberately the simple tier — one request, one reply, bounded by a timeout. Multi-message protocol conversations (sequence numbers, re-synchronisation, wait-for-any flows) are codec/actor state; the codec seam is their mounting point, not their replacement.

Every invoke is bounded by default: a call that carries no ClientInvokeOptions.Timeout gets the kernel's DefaultInvokeTimeout (30 s out of the box; AddElarionConnections(o => o.DefaultInvokeTimeout = …)), resolved by the sink before the codec sees the options — a client that never answers surfaces as a TimeoutException, never a silently hung await. The layering is per-call Timeout > DefaultInvokeTimeout > unbounded: a per-call value always wins (pass Timeout.InfiniteTimeSpan to make a single call unbounded), and only a default explicitly configured to null applies none.

The TCP endpoints

Elarion.Connections.Tcp runs the same handler/codec seams over raw sockets (BCL only, no ASP.NET) — the shape for devices that speak a proprietary protocol and can't do WebSockets. Because TCP has no message boundaries, the adapter owns a framing seam — and framing is boundaries only: pick LengthPrefixedTcpFramer, DelimitedTcpFramer (line/telegram framing — an optional start delimiter skips serial-bridge line noise), or implement TcpMessageFramer for the vendor's format. Bytes are bytes on TCP, so every inbound message reaches the codec's OnBinaryAsync as a raw slice; a text protocol's codec decodes with one Encoding.UTF8.GetString — the string is paid for only where it is wanted. The pipeline is benchmarked at raw-socket parity with zero allocations per message in both directions — receive (framer slice → codec) and send (SendBinaryAsync reuses a per-connection frame buffer, so a proxy that forwards between links allocates nothing) — and an armed-but-quiet idle window costs nothing on the hot path. Sockets default to NoDelay (Nagle stalls serial request/reply telegrams); override per endpoint or per connection for bulk one-way streams.

// Devices dial in:
builder.Services.AddElarionTcpConnectionListener<GatewaySocketHandler2>(o => {
    o.ListenEndPoint = new IPEndPoint(IPAddress.Any, 7010);
    o.Framer = new DelimitedTcpFramer(end: (byte)'>', start: (byte)'<');
    o.IdleTimeout = TimeSpan.FromSeconds(60);
});

// Or the gateway dials the device and keeps the link alive (jittered exponential reconnect):
builder.Services.AddElarionTcpConnectionDialer<DeviceLinkHandler>(o => {
    o.Host = "10.0.40.17";
    o.Port = 2101;
    o.Framer = new LengthPrefixedTcpFramer();
});

Each Add… call is one endpoint — listen on several ports (one per device channel type) or dial several devices, each with its own handler, framer, and options. The handler subclasses TcpConnectionHandler; device protocols with no credential exchange ticket straight from handshake.RemoteEndPoint/binding configuration.

Bindings that live as data (an admin configures which device is dialed or listened for, on which port, with which framing) use the runtime manager instead of composition-time registrations: add AddElarionTcpConnectionEndpoints() and apply named endpoints whenever configuration changes — Apply… is an upsert that tears the old endpoint down first, so changing a binding means a reconnect under the new settings, including flipping its direction:

var endpoints = provider.GetRequiredService<TcpConnectionEndpoints>();

// Startup: apply every binding row; on admin change: re-apply just that row.
await endpoints.ApplyDialerAsync<DeviceLinkHandler>("device-7:mgmt", o => {
    o.Host = row.Host; o.Port = row.Port;
    o.Framer = row.Protocol == "telegram" ? telegramFramer : new LengthPrefixedTcpFramer();
}, ct);

// The same binding later becomes server-based (the device now dials us):
await endpoints.ApplyListenerAsync<GatewayListenHandler>("device-7:mgmt", o => {
    o.ListenEndPoint = new IPEndPoint(IPAddress.Any, row.Port);
    o.Framer = telegramFramer;
}, ct);

await endpoints.RemoveAsync("device-7:mgmt", ct);   // unbind: connections unregister

Every managed endpoint advertises its binding health: endpoints.Statuses / GetStatus(name) answer "which bindings are serving, which failed to bind, and why" — a listener whose port couldn't be bound is Faulted with the reason; a dialer between attempts is Dialing carrying the last failure — and endpoints.StatusChanged fires on every transition, ready to be projected onto a client event so an admin UI shows binding state live.

Settings can also vary per connection on one endpoint: override ConfigureConnectionAsync(TcpConnectionPeer, …) and return TcpConnectionSettings (framer, size cap, idle window, transport tag — nulls inherit the endpoint options). It runs before any byte is exchanged, so a binding-configuration lookup keyed on the peer can pick the wire framing that governs the handshake itself — the shape gateways need when differently-speaking device families share one ingress port.

Keepalives and conversations

Two optional pieces cover what every device codec otherwise hand-rolls:

• The idle hook. Set IdleTimeout on any adapter and the codec's OnIdleAsync fires per elapsed window without inbound traffic — send the protocol's poll/heartbeat there, or throw to declare the link dead. Default is off; the hook never abandons the pending read.
• Conversation helpers (in the kernel, use only if they fit): ConnectionPendingRequests<TKey, TResponse> is the sequence-number → completion map behind a codec's InvokeAsync — register the key, send, await the correlated reply with a timeout, FailAll on teardown. ConnectionInbox<TMessage> serves multi-message flows: the receive path Posts every parsed message, flow code awaits the next message matching a predicate ("the ready frame or the abort frame"), buffered so a fast reply beats its waiter safely, and Complete faults everything when the connection ends.

internal sealed class DeviceCodec(TcpClientConnection connection) : IClientConnectionProtocol {
    private readonly ConnectionPendingRequests<ushort, DeviceFrame> _pending = new();

    public ValueTask OnBinaryAsync(ReadOnlyMemory<byte> message, CancellationToken ct) {
        var frame = DeviceFrame.Parse(message.Span);
        return _pending.TryComplete(frame.Sequence, frame)
            ? ValueTask.CompletedTask
            : HandleUnsolicitedAsync(frame, ct);              // telemetry, events, …
    }

    public async ValueTask<TResponse> InvokeAsync<TRequest, TResponse>(
        string name, TRequest request, ClientInvokeOptions? options, CancellationToken ct)
        where TRequest : class {
        var sequence = NextSequence();
        // The sink already resolved the default invoke timeout into options — honor it as-is
        // (a null Timeout here means the invoke is deliberately unbounded).
        var reply = _pending.WaitAsync(sequence, options?.Timeout, ct);
        await connection.SendBinaryAsync(DeviceFrame.Encode(sequence, name, request), ct);
        return Decode<TResponse>(await reply);
    }

    public ValueTask OnIdleAsync(CancellationToken ct) =>
        connection.SendBinaryAsync(DeviceFrame.Poll(), ct);   // the 60 s keepalive
}

Client events over a connection

A connection can serve the same topics the SSE endpoint serves — same catalog, same fail-closed authorization, so a second transport can never fork the rules. The adapter parses its subscribe frame and hands the requests to the bridge; delivery framing is the adapter's choice:

var result = await bridge.SubscribeAsync(connection.Connection, requests,
    (envelope, ct) => connection.SendTextAsync($"event:{envelope.Topic}:{envelope.Payload}", ct), ct);

Delivery starts with the elarion.connected greeting (the same re-query contract an SSE stream opens with), and the subscription dies automatically when its connection unregisters. Statuses mirror the SSE endpoint's HTTP mapping (unauthenticated / invalid / not-found — unknown and denied topics stay indistinguishable).

Simulation and testing

Elarion.Connections.Simulation ships the pieces every gateway test suite — and every runnable dev/demo device simulator — otherwise hand-rolls. SimulatedClientConnection is an in-memory sink double — register it with the real registry and drive twins, observers, and presence logic deterministically: await its Sent channel for outbound messages, answer InvokeAsync via InvokeResponder, and Close() it to assert your code survives ClientConnectionClosedException. AwaitableConnectionObserver turns lifecycle edges into awaitable completions (with Reset() for reconnect scenarios). TcpSimulatorClient is the framed client for simulated devices/peers: connect it to a listener (or wrap the accepted side of a fake device an Elarion dialer reached) and speak whole messages through the endpoint's own framer, with text helpers for challenge/response handshakes.

The default for codec/handler-level work is fully in-memory — no OS sockets, ports, or firewall prompts: InMemoryTcpLink.Start(handler, registry, o => o.Framer = …) runs the complete connection lifecycle (per-connection settings, handshake, framing, registry, codec, idle hook) over an in-memory duplex pair, hands you the simulator-side Client, a deterministic ServerConnection await (the registered sink — no polling, no handshake races), and a ServerCompletion for teardown assertions. Reserve real sockets (loopback TCP, Kestrel + ClientWebSocket) for integration tests of the adapters themselves. InMemoryDuplexStream.CreatePair() is public for anything else that wants a socket pair without the socket.

Putting it together — the device-gateway loop

Connections are deliberately not a fourth invocation model — they're a carrier with identity. Actors stay the concurrency gate for device state, handlers stay the authorization gate for anything command-shaped, client events stay the one fan-out. The whole loop, end to end:

Connect/disconnect → the twin. An observer (the one piece of hand-written glue) routes lifecycle into the digital twin's public facade; the twin marks presence and captures the sink — an actor owning a stateful connection is exactly actor use-case #1, and the mailbox serializes a device's parallel channels against user-triggered commands for free:

internal sealed class DeviceTwinAttachment(IActorSystem actors) : IClientConnectionObserver {
    public ValueTask OnConnectedAsync(IClientConnectionSink connection, CancellationToken ct = default) =>
        new(actors.Get<IDeviceTwin>(connection.Connection.PrincipalId!).ChannelUp(connection));

    public ValueTask OnDisconnectedAsync(ClientConnection connection, CancellationToken ct = default) =>
        new(actors.Get<IDeviceTwin>(connection.PrincipalId!).ChannelDown(connection.ConnectionId));
}

// services.TryAddEnumerable(ServiceDescriptor.Singleton<IClientConnectionObserver, DeviceTwinAttachment>());

Inbound frames → the twin, pipeline-free. Telemetry isn't a command; the codec parses and hands the frame to the mailbox (see the codec examples above). The codec's one routing decision is which frames are commands — those go through the dispatch rail instead:

// Inside a codec, for command-shaped messages: the same per-call scope rail JSON-RPC and MCP use.
// Authorization, validation, [Idempotent], transaction, and audit all run per message, evaluated
// against the connection's principal (seeded as ICurrentUser by the standard initializer).
var boundary = new DispatchScopeContext();
boundary.Set(connection.Connection.Principal);
await using var scope = services.CreateDispatchScope(boundary);
if (dispatcher.TryGetRoute(name, HandlerTransports.Connection, out var route)) {
    var result = await route.InvokeAsync(request, scope.ServiceProvider, ct);
    // encode the result back over the sink — your framing
}

User → device. A UI action arrives as a normal [Handler] (over HTTP/JSON-RPC — [RequirePermission], ownership check, [Auditable]) and calls the twin facade; the twin talks to the device through its captured sink, serialized with everything else in its mailbox:

[Handler("robots.start")]
internal sealed class StartRobot(IActorSystem actors) : IHandler<StartRobot.Command, Result<Unit>> {
    public sealed record Command : ICommand { public required Guid RobotId { get; init; } }

    public async ValueTask<Result<Unit>> HandleAsync(Command request, CancellationToken ct) =>
        await actors.Get<IDeviceTwin>(request.RobotId.ToString("N")).StartAsync(ct);
}

// inside the twin: one request/reply into the device, bounded by a timeout
var ack = await _channel.InvokeAsync<StartCommand, StartAck>(
    "start", new StartCommand(), new ClientInvokeOptions { Timeout = TimeSpan.FromSeconds(5) }, ct);

Live state → every UI. The twin publishes client events from its hot state (the ephemeral tier), or projects committed facts via a method-form [ConsumeEvent]; SSE browsers and connection-carried subscribers receive identically through the bridge. Durable facts flow as integration events — which can come back into actors via [ConsumeEvent] on actor methods.

The hand-written glue per app is the observer (~10 lines) and the codec's routing decision; the handshake, registry, lifecycle, per-message scopes, mailbox, and fan-out are all framework.

Rules of the road

• Facts still travel as client events, even over a connection — the sink is for conversation-shaped traffic (replies, control frames, RPC), not a side channel for state fan-out.
• Guarantees are unchanged: events are at-most-once hints healed by re-query; sends are at-most-once; InvokeAsync faults explicitly. Commands a device may retransmit belong behind [Idempotent].
• One device, many connections: give every channel the same PrincipalId and let a digital-twin actor keyed by device id serialize the shared state — per-connection ordering comes from the receive loop, cross-channel consistency from the actor.
• The registry is node-local by design. Multi-node deployments co-locate device ingress with the single-homed twin (the role-holder proxy's prefix list); a replicated connection directory is the point where you adopt a clustered runtime instead.
• Telemetry: register the Elarion.Connections meter — connection.active, connection.opened/connection.closed (tagged elarion.connection.transport), and connection.event_subscriptions.active.