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()                  // Actor authority mesh는 Framework가 선택한다.
    .registerSession(PlayStreamSession.class); // 연결마다 만들 session type을 등록한다.

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.

public final class PlaySession implements ZLinkSession {
    private final ZLinkSessionContext context;
    private final ZLinkSessionPacketDispatcher<ZLinkSessionContext> handlers;

    public PlaySession(
        ZLinkSessionContext context,
        ZLinkSessionPacketDispatcher<ZLinkSessionContext> handlers) {
        this.context = context;
        this.handlers = handlers;
    }

    @Override
    public ZLinkSessionContext context() {
        return context;
    }

    @Override
    public CompletionStage<Void> onConnected() {
        return CompletableFuture.completedFuture(null);
    }

    @Override
    public CompletionStage<Void> onDisconnected() {
        return CompletableFuture.allOf(context.actors().bound().stream()
            .map(actor -> actor.notifyDisconnected().toCompletableFuture())
            .toArray(CompletableFuture[]::new));
    }

    @Override
    public CompletionStage<Void> onError(ZLinkStreamError error) {
        return CompletableFuture.completedFuture(null);
    }

    @Override
    public CompletionStage<Void> onDispatch(
        ZLinkSessionDispatchContext header,
        ZLinkMessage payload) {
        return handlers.tryHandle(context, header, payload).thenCompose(handled ->
            handled
                ? CompletableFuture.completedFuture(null)
                : requireActor(header.packetName()).relay(header, payload).thenApply(ignored -> null));
    }

    private ZLinkSessionActor requireActor(String packetName) {
        return switch (context.actors().bound().size()) {
            case 1 -> context.actors().bound().get(0);
            case 0 -> throw new IllegalStateException(
                "AuthenticateReq is required before play packet '" + packetName + "'");
            default -> throw new IllegalStateException(
                "Exactly one actor must be bound before play packet '" + packetName + "'");
        };
    }
}

A minimal implementation looks like this.

public final class PlaySession implements ZLinkSession {
    private final ZLinkSessionContext context;
    private final Logger logger;

    @Override
    public void configure() {
        // Registers a typed session packet handler.
        context.handlers().addHandler(PingHandler.class);
    }

    @Override
    public CompletionStage<Void> onConnected() {
        logger.info("connected: {}", context.sessionId());
        return CompletableFuture.completedFuture(null);
    }

    @Override
    public CompletionStage<Void> onDispatch(
        ZLinkSessionDispatchContext dispatch, ZLinkMessage payload) {
        return context.handlers().tryHandle(context, dispatch, payload).thenCompose(handled -> handled
            ? CompletableFuture.<Void>completedFuture(null)
            // Closes the connection on a packet outside the application protocol.
            : context.close().toCompletableFuture());
    }

    @Override
    public CompletionStage<Void> onDisconnected() {
        logger.info("disconnected: {}", context.sessionId());
        return CompletableFuture.completedFuture(null);
    }
}

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 final class PingHandler
    implements ZLinkTypedSessionPacketHandler<ZLinkSessionContext, StreamMessages.Ping> {

    @Override
    public Class<StreamMessages.Ping> messageType() {
        return StreamMessages.Ping.class;
    }

    @Override
    public CompletionStage<Void> handle(
        ZLinkSessionContext context,
        ZLinkSessionDispatchContext dispatch,
        StreamMessages.Ping message) {
        if (!dispatch.canReply()) {
            throw new IllegalStateException("Ping must be a request.");
        }

        // 같은 request correlation으로 한 번만 reply한다.
        return context.client().reply(new StreamMessages.Pong(message.sequence())).submit();
    }
}

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.

// local transport queue admission까지 기다린다.
context.client()
    .send(new StreamMessages.ServerNotice("maintenance"))
    .metadata("severity", "info")
    .compress()
    .submit();

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.

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

connector.on(GameStateNotify.class, message -> {
    render(message.payload());
    return CompletableFuture.completedFuture(null);
});

// Finishes connecting and preparing the receive loop.
connector.connect().submit().toCompletableFuture().join();

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

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(new PlayerInput(direction)).submit().toCompletableFuture().join();

// Finds the response by request sequence.
Profile profile = connector
    .request(new GetProfile(playerId))
    .submit(Profile.class)
    .toCompletableFuture().join();

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.

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