9. STREAM

Guide Home | Previous: 8. Session and Actor Binding | Next: 10. Location — Auto-Connect and Object Location

View in another language — C#/.NET · C++ · Java · Kotlin · Node/TypeScript

The documents that own this chapter's contract — STREAM server session owns the behavior, and the per-language STREAM session public contract owns the server's exact signatures. The client package follows the Stream Connector guide and the per-language public contract.

STREAM is a connection-oriented, bidirectional message channel between an external client and the Framework server. The server implements session lifecycle and packet dispatch. The client uses the independent package Zlink.Stream.Connector.

1. Registering a Server Node

Register one session type on a Stream node. If you use Actor dispatch, enable it explicitly.

options.AddStreamNode("client-stream")
    .Bind("tcp://0.0.0.0:9100")
    .EnableActorDispatch()
    // Registers the session type to create per connection.
    .AddSession<PlaySession>();

Session handlers and Actor/Spot handlers use the Framework's default typed JSON serialization. The application doesn't register a codec per message type or parse a raw frame itself.

Registration is explicit. There's no surface that implicitly registers a stream node via an attribute, annotation, or decorator. There are only three axes — node name, bind endpoint, session type. Of these, the bind endpoint must always be specified.

The following eight conditions are rejected as configuration errors before host startup, not deferred to the first connection.

Condition
The node name is empty
The same node name is registered twice
There's no bind endpoint
The same session type is registered more than once
More than one session is registered on one node
TLS is on but the certificate path is empty
TLS is on but the key path is empty
A client certificate is required without a TLS server configured

If you enable TLS, specify both the certificate and key paths together. Requiring a client certificate is disabled by default; turning it on rejects a connection that fails verification before a session is ever created.

2. Session Lifecycle

A session implements callbacks for connection, packet dispatch, errors, and disconnection. A given session's callbacks run serially.

See it in a sample — TicTacToe. This is the session that represents one client connection. It filters out the authentication packet first and relays everything else to the Actor. This is actual code from the repository.

internal sealed class PlaySession(
    IZLinkSessionContext context,
    ILogger<PlaySession> logger)
    : IZLinkSession
{
    public IZLinkSessionContext Context { get; } = context;

    public void Configure()
    {
        // request: authenticates the STREAM session before actor packet relay starts.
        Context.Handlers.AddHandler<AuthenticatePlaySessionHandler>(nameof(AuthenticateReq));
    }

    public ValueTask OnConnectedAsync(CancellationToken cancellationToken)
    {
        logger.LogInformation(
            "client -> play stream: connected. sessionId={SessionId}",
            Context.SessionId);
        return ValueTask.CompletedTask;
    }

    public async ValueTask OnDisconnectedAsync(CancellationToken cancellationToken)
    {
        var boundActors = Context.Actors.Bound.ToArray();
        logger.LogInformation(
            "client -> play stream: disconnected. sessionId={SessionId}, actors={ActorCount}",
            Context.SessionId,
            boundActors.Length);

        foreach (var actor in boundActors) await actor.NotifyDisconnectedAsync(cancellationToken);
    }

    public ValueTask OnErrorAsync(
        ZLinkStreamError error,
        CancellationToken cancellationToken)
    {
        logger.LogWarning(
            "play stream: error. code={Code}, message={Message}, sessionId={SessionId}",
            error.Error,
            error.Message,
            Context.SessionId);
        return ValueTask.CompletedTask;
    }

    public async ValueTask OnDispatchAsync(
        ZLinkSessionDispatchContext dispatch,
        ZLinkMessage payload,
        CancellationToken cancellationToken)
    {
        logger.LogInformation(
            "client -> play stream: message received. name={MessageName}, kind={Kind}, sessionId={SessionId}",
            dispatch.PacketName,
            dispatch.CanReply ? "Request" : "Send",
            Context.SessionId);

        if (await Context.Handlers.TryHandleAsync(dispatch, payload, cancellationToken))
            return;

        var actor = RequireSingleBoundActor($"relaying packet '{dispatch.PacketName}'");
        await actor.RelayAsync(payload, cancellationToken);
    }

    private IZLinkSessionActor RequireSingleBoundActor(string action)
    {
        var actors = Context.Actors.Bound;
        return actors.Count switch
        {
            1 => actors.Single(),
            0 => throw new InvalidOperationException($"Client must authenticate before {action}."),
            _ => throw new InvalidOperationException($"Exactly one actor must be bound before {action}.")
        };
    }
}

A minimal implementation looks like this.

public sealed class PlaySession(
    IZLinkSessionContext context,
    ILogger<PlaySession> logger) : IZLinkSession
{
    public IZLinkSessionContext Context { get; } = context;

    public void Configure()
    {
        // Registers a typed session packet handler.
        Context.Handlers.AddHandler<PingHandler>();
    }

    public ValueTask OnConnectedAsync(CancellationToken cancellationToken)
    {
        logger.LogInformation("connected: {SessionId}", Context.SessionId);
        return ValueTask.CompletedTask;
    }

    public async ValueTask OnDispatchAsync(
        ZLinkSessionDispatchContext dispatch,
        ZLinkMessage payload,
        CancellationToken cancellationToken)
    {
        if (!await Context.Handlers.TryHandleAsync(
                dispatch,
                payload,
                cancellationToken))
        {
            // Closes the connection on a packet outside the application protocol.
            await Context.CloseAsync();
        }
    }

    public ValueTask OnErrorAsync(
        ZLinkStreamError error,
        CancellationToken cancellationToken)
    {
        logger.LogWarning(
            "session error: {Error} {Message}",
            error.Error,
            error.Message);
        return ValueTask.CompletedTask;
    }

    public ValueTask OnDisconnectedAsync(CancellationToken cancellationToken)
    {
        logger.LogInformation("disconnected: {SessionId}", Context.SessionId);
        return ValueTask.CompletedTask;
    }
}

Errors are routed in four ways. The session error callback receives only a transport error that belongs to that session.

Error	Where it goes
That session's transport error	The session error callback
A handshake failure	Runtime monitoring. There's no target to call — the session hasn't been created yet
A socket/node-level error	Runtime monitoring. It can't be attributed to a single session
An application handler exception	The handler exception path. Not the session error callback

A handler filter doesn't apply to session dispatch. Even a filter attached to a different dispatch never runs ahead of a session callback. Anything that needs filtering on the session path, like authentication, is handled through the session's own handler registration.

There's no API for running a receive loop directly. The Framework enqueues the packet and then runs the session callback, applying dispatch, DI, and logging consistently at that boundary. This is by design, so the application never has to manage the loop, cancellation, or backpressure itself.

3. Typed Packet Handler

The handler registry decodes the received message into a typed message. When replying to a request, use the current dispatch's one-shot reply token.

public sealed class PingHandler
    : IZLinkSessionPacketHandler<IZLinkSessionContext, Ping>
{
    public async ValueTask HandleAsync(
        IZLinkSessionContext context,
        ZLinkSessionDispatchContext dispatch,
        Ping message,
        CancellationToken cancellationToken)
    {
        if (!dispatch.CanReply)
        {
            throw new InvalidOperationException("Ping must be a request.");
        }

        await context.Client
            .Reply(new Pong(message.Sequence))
            // Replies exactly once, using the same request correlation.
            .Async(cancellationToken);
    }
}

Reply is valid only for the current request and can be submitted once. Even if the send fails on a timeout or cancellation, the same reply token can't be reused.

Responses do not carry packet names. The client finds the pending request purely by request sequence, and the type specified at the call site decides what type to read the response as. Because it's not selected by name, there's no API for attaching a packet name to the response side either. An error response also comes back on the same sequence.

Use Send for server-initiated pushes.

await Context.Client
    .Send(new ServerNotice("maintenance"))
    .Metadata("severity", "info")
    .Compress()
    // Waits for admission into the local transport queue.
    .Async(cancellationToken);

4. Actor Dispatch

After authentication, bind an Actor to the session, and a message not handled by a session-only handler can be handed off through the session actor's relay call. The detailed flow follows Session and Actor Binding.

The application doesn't query the session route from the Location Store directly. Once Actor relocation completes, the Framework updates the binding route.

5. Client Connection

The client uses the Stream Connector package, not the server Framework package.

await using var connector = ZlinkStreamConnectorFactory.Create(
    new ZlinkStreamConnectorOptions
    {
        Endpoint = new Uri("tcp://game.example.com:9100"),
        DispatchMode = ZlinkStreamDispatchMode.Manual
    });

connector.On<GameStateNotify>("GameStateNotify", (message, cancellationToken) =>
{
    Render(message.Payload);
    return ValueTask.CompletedTask;
});

// Finishes connecting and preparing the receive loop.
await connector.Connect.Async(cancellationToken);

while (running)
{
    // Manual mode runs the callback on this caller.
    await connector.Dispatch.Async(cancellationToken);
}

Use Manual when the callback needs to run on a game loop or UI thread. Immediate runs the callback on the connector's own worker, so it isn't suitable for clients that need thread affinity.

5.1 Diagnostics Level

The connector takes a diagnostics level option with the same four values as the server runtime (Off/Errors/Normal/Detailed). The default is Errors, which keeps the existing behavior; lowering it to Off stops the connector from creating or attaching flow identifiers on outbound frames, removing the observation-only cost (Stream Connector common spec §13). The correlation data used for request/response matching is part of the protocol and keeps working at Off.

new ZlinkStreamConnectorOptions
{
    Endpoint = new Uri("tcp://game.example.com:9100"),
    DiagnosticsLevel = ZlinkStreamDiagnosticsLevel.Off
};

6. Client Send and Request

await connector
    .Send(new PlayerInput(direction))
    // Waits for admission into the bounded outbound queue.
    .Async(cancellationToken);

Profile profile = await connector
    .Request(new GetProfile(playerId))
    // Finds the response by request sequence.
    .Async<Profile>(cancellationToken);

The Connector's default typed codec is JSON. Packet name override, push waiting, reconnect, heartbeat, and bounded queue settings are explained in the Stream Connector guide.

7. Related Documents

• Runnable verification examples for this chapter's contract: 13. Interface Catalog chapter §5 — the verification class StreamContracts
• Session and Actor binding: Session Actor Dispatch
• Full client connector usage: the Stream Connector guide
• Location Store and auto-connect: Location