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")
    // Kotlin uses the Java surface as-is.
    .enableActorDispatch()
    // Registers the session type to create per connection.
    .registerSession(PlaySession::class.java)

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.

class PlaySession(
    private val context: ZLinkSessionContext,
    private val handlers: ZLinkSessionPacketDispatcher<ZLinkSessionContext>,) : ZLinkSuspendingSession() {
    override fun context(): ZLinkSessionContext = context

    override suspend fun onDisconnectedSuspending() {
        context.actors().bound().forEach { actor ->
            actor.notifyDisconnected().await()
        }
    }

    override suspend fun onDispatchSuspending(header: ZLinkSessionDispatchContext, payload: ZLinkMessage) {
        if (handlers.tryHandle(context, header, payload).await()) {
            return
        }
        requireActor(header.packetName()).relay(header, payload).await()
    }

    private fun requireActor(packetName: String): ZLinkSessionActor =
        when (context.actors().bound().size) {
            1 -> context.actors().bound()[0]
            0 -> throw IllegalStateException("AuthenticateReq is required before play packet '$packetName'")
            else -> throw IllegalStateException("Exactly one actor must be bound before play packet '$packetName'")
        }
}

A minimal implementation looks like this.

class PlaySession(
    private val context: ZLinkSessionContext,
    private val logger: Logger,
) : ZLinkSession {

    override fun configure() {
        // Registers a typed session packet handler.
        context.handlers().addHandler(PingHandler::class.java)
    }

    override suspend fun onConnected() {
        logger.info("connected: {}", context.sessionId())
    }

    override suspend fun onDispatch(dispatch: ZLinkSessionDispatchContext, payload: ZLinkMessage) {
        if (context.handlers().tryHandle(context, dispatch, payload).await()) return
        // Closes the connection on a packet outside the application protocol.
        context.close().await()
    }

    override suspend fun onDisconnected() {
        logger.info("disconnected: {}", context.sessionId())
    }
}

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.

suspend fun handle(
    context: ZLinkSessionContext, dispatch: ZLinkSessionDispatchContext, message: Ping) {
    check(dispatch.canReply()) { "Ping must be a request." }

    // Replies exactly once, using the same request correlation.
    context.client().reply(Pong(message.sequence)).submit().await()
}

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.

// Waits for admission into the local transport queue.
context.client()
    .send(ServerNotice("maintenance"))
    .metadata("severity", "info")
    .compress()
    .submit()
    .await()

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.

val connector = ZLinkStreamConnectorFactory.create(
    ZLinkStreamConnectorOptions(
        URI.create("tcp://game.example.com:9100"),
        ZLinkStreamDispatchMode.MANUAL))

connector.on(GameStateNotify::class.java) { message ->
    render(message.payload())
    CompletableFuture.completedFuture(null)
}

// Finishes connecting and preparing the receive loop.
connector.connect().submit().await()

while (running) {
    // MANUAL mode runs the callback on this caller.
    connector.dispatch().submit().await()
}

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.

ZLinkStreamConnectorOptions.createDefault(endpoint).withDiagnosticsLevel(ZLinkStreamDiagnosticsLevel.OFF)

6. Client Send and Request

// Waits for admission into the bounded outbound queue.
connector.send(PlayerInput(direction)).submit().await()

// Finds the response by request sequence.
val profile = connector.request(GetProfile(playerId)).submit(Profile::class.java).await()

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