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.add_stream_node ("client-stream")
  .bind ("tcp://0.0.0.0:9100")
  .enable_actor_dispatch ()
  // Registers the session type to create per connection.
  .register_session<play_session_t> ();

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 play_session_t final : public packet_stream_session_t
{
  public:

    play_session_t (session_actor_manager_t &actors,
                    authenticate_play_session_handler_t &authenticate) :
        _actors (actors), _authenticate (authenticate)
    {
    }

    task_t<void> on_connected (stream_t &) override
    {
        return task_t<void> (result_t<void>::success ());
    }

    task_t<void> on_disconnected (stream_t &) override
    {
        if (_bound_actor_id) {
            if (auto actor = _actors.find (*_bound_actor_id)) {
                co_await actor->notify_disconnected ();
            }
            _bound_actor_id.reset ();
        }
        co_return;
    }

    task_t<void> on_error (stream_t &, const stream_error_t &) override
    {
        return task_t<void> (result_t<void>::success ());
    }

    task_t<void> on_packet (stream_t &stream,
                            const session_message_context_t &dispatch,
                            const zlink::message_t &payload) override
    {
        if (_authenticate.can_handle (dispatch)) {
            auto authenticated = co_await _authenticate.handle (_actors, stream, payload);
            _bound_actor_id = std::string (authenticated.actor_id ());
            co_return;
        }

        auto actor = require_bound_actor (std::string ("dispatching packet '")
                                          + std::string (dispatch.packet_name) + "'");
        if (!actor) {
            co_return;
        }
        if (dispatch.can_reply) {
            auto reply =
              co_await actor.value ().relay_request (std::string (dispatch.packet_name), payload)
                .async ();
            stream.reply_packet (reply).async ();
            co_return;
        }
        co_await actor.value ().relay (std::string (dispatch.packet_name), payload);
        co_return;
    }

  private:
    result_t<session_actor_t> require_bound_actor (const std::string &action) const
    {
        if (!_bound_actor_id) {
            return result_t<session_actor_t>::failure (framework_error_kind_t::internal_failure,
                                                       "Client must authenticate before " + action
                                                         + ".");
        }
        auto actor = _actors.find (*_bound_actor_id);
        if (!actor) {
            return result_t<session_actor_t>::failure (
              framework_error_kind_t::not_found,
              "Exactly one actor must be bound before " + action + ".");
        }
        return result_t<session_actor_t>::success (std::move (*actor));
    }

    session_actor_manager_t &_actors;
    authenticate_play_session_handler_t &_authenticate;
    std::optional<std::string> _bound_actor_id;
};

A minimal implementation looks like this.

// A C++ session inherits packet_stream_session_t and overrides the callbacks.
class play_session_t : public packet_stream_session_t
{
  public:
    task_t<void> on_connected (stream_t &stream) override
    {
        _logger.info ("connected");
        co_return;
    }

    task_t<void> on_packet (stream_t &stream,
                            const stream_dispatch_context_t &dispatch,
                            const zlink::message_t &payload) override
    {
        if (!_ping.can_handle (dispatch)) {
            // Closes the connection on a packet outside the application protocol.
            co_await stream.close ();
            co_return;
        }
        co_await _ping.handle (stream, payload);
    }

    task_t<void> on_error (stream_t &, const stream_error_t &error) override
    {
        _logger.warn (std::string ("session error: ") + error.message);
        co_return;
    }

    task_t<void> on_disconnected (stream_t &) override
    {
        _logger.info ("disconnected");
        co_return;
    }
};

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.

// A C++ typed handler receives only the stream and the decoded payload. The dispatch
// context is passed only to the raw on_packet path, so can_reply isn't checked here.
task_t<void> handle (stream_t &stream, const ping_t &message)
{
    // Replies exactly once, using the same request correlation. Ends in failure if it isn't a request.
    co_await stream.reply_packet (zlink::message_t::from_json (pong_t{message.sequence})).async ();
}

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.
co_await stream.send (server_notice_t{"maintenance"})
  .metadata ("severity", "info")
  .compress ()
  .async ();

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.

zlink::stream_connector::connector_options_t connector_options;
connector_options.endpoint = "tcp://game.example.com:9100";
connector_options.dispatch_mode = zlink::stream_connector::dispatch_mode_t::manual;
auto connector = zlink::stream_connector::connector_factory_t::create (connector_options);

connector.on<game_state_notify_t> ("GameStateNotify",
                                   [] (const auto &message) { render (message.payload ()); });

// Finishes connecting and preparing the receive loop.
co_await connector.connect ().async ();

while (running) {
    // manual mode runs the callback on this caller.
    co_await connector.dispatch ().async ();
}

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.

connector_options.diagnostics_level = zlink::stream_connector::diagnostics_level_t::off;

6. Client Send and Request¶

// Waits for admission into the bounded outbound queue.
co_await connector.send (player_input_t{direction}).async ();

// Finds the response by request sequence.
auto profile = co_await connector.request (get_profile_t{player_id}).async<profile_t> ();

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