STREAM¶

Guide Home | Previous: Actor | Next: Session and Actor

What you get from this chapter

You can let a program outside the mesh attach over one connection, send requests and receive answers. The code comes from the tutorial README in the examples repository; follow its Download, Build, and Run sections to reproduce the results below.

A client that reaches a STREAM node uses its language's stream connector. Client stream connector covers installation for Unity, Unreal, browser, Node, .NET, Java, and C++ clients, and the stream connector guide for that language covers connection and packet handling.

Every call so far ran between nodes inside the mesh. A game client or an app is outside it and references no Framework at all. STREAM is where such a program attaches, and while the one connection is open both sides send to each other. This chapter goes as far as accepting a connection and answering one packet; binding that connection to an Actor is covered by Session and Actor.

1. The Role of STREAM¶

STREAM is the boundary where a client outside the mesh exchanges packets with it over one connection. Channels, Spots and Actors are all paths called from inside the mesh. The caller has to be a member of the mesh and to reference the Framework. A client running on a player's device meets neither condition.

STREAM narrows that boundary to a single connection. The client needs one address and the packet names, and it never learns how many nodes the mesh has.

↗ View larger

One session object stands for one connection. Packets arriving on that connection are handled one at a time in that session, so a session's fields are the place to keep the state of that connection.

2. Deciding the Packets¶

These are no different from the contracts of a mesh call. They travel to a client outside the mesh, though, so the names have to be the ones that client uses.

// Exchanged over the external TCP connection, not between mesh nodes. The
// stream codec wants 64-bit integers as decimal strings, so the timestamp is
// carried as text rather than as an integer.
struct ping_t
{
    static constexpr const char *packet_name = "Ping";
    std::string sent_at_unix_ms;
};

struct pong_t
{
    static constexpr const char *packet_name = "Pong";
    std::string sent_at_unix_ms;
};

3. The Receiving Side — the Node That Accepts Connections¶

The receiving process

The code in this section goes into the process that accepts the connection.

3.1 Writing the Session¶

A session stands for one connection. Connect, disconnect, error and packet arrival all come as callbacks, and the callbacks for one connection run in order.

// One connected game client. Callbacks for the same connection run in order.
// C++ has no session handler registry: every packet arrives at on_packet and
// this class decides what to do with it by name.
class game_session_t final : public fw::packet_stream_session_t
{
  public:
    fw::task_t<void> on_connected (fw::stream_t &stream) override
    {
        std::cout << "client connected: " << stream.session_id () << std::endl;
        co_return;
    }

    fw::task_t<void> on_disconnected (fw::stream_t &stream) override
    {
        std::cout << "client disconnected: " << stream.session_id () << std::endl;
        co_return;
    }

    fw::task_t<void> on_error (fw::stream_t &stream, const fw::stream_error_t &) override
    {
        std::cout << "stream error on " << stream.session_id () << std::endl;
        co_return;
    }

    fw::task_t<void> on_packet (fw::stream_t &stream,
                                const fw::session_message_context_t &dispatch,
                                const zlink::message_t &payload) override
    {
        const auto packet = std::string (dispatch.packet_name);

        if (packet == ping_t::packet_name) {
            const auto request = payload.parse_json<ping_t> ();

            // reply_packet answers a request. To push to a client that is not
            // waiting for one, use write_packet instead.
            stream.reply_packet (zlink::message_t::from_json (pong_t{request.sent_at_unix_ms}))
              .async ();
            co_return;
        }

        // Each authenticate packet binds its player to this connection.
        if (packet == authenticate_t::packet_name) {
            const auto request = payload.parse_json<authenticate_t> ();

            // A returning client finds its existing player rather than a new one.
            auto &actors = stream.actors ();
            auto located = actors.get_or_create (
              "player", request.player_id, create_player_t{request.player_id});
            if (!located)
                throw fw::framework_exception_t (located.error_kind (),
                                                 "Player creation was rejected.");

            auto bound = co_await actors.bind_or_get (located.value ().ref ()).async ();
            stream
              .reply_packet (
                zlink::message_t::from_json (authenticated_t{std::string (bound.actor_id ())}))
              .async ();
            co_return;
        }

        // The packet's Actor slot selects one binding on this connection.
        if (dispatch.actor) {
            co_await dispatch.actor->relay (packet, payload);
            co_return;
        }

        auto bound = stream.actors ().bound ();
        if (bound.size () != 1)
            throw fw::framework_exception_t (
              fw::framework_error_kind_t::invalid_operation,
              "Specify an Actor when the session has no unique binding.");

        co_await bound.front ().relay (packet, payload);
    }
};

The packet-arrival callback is the gate. Registering handlers is not enough on its own; nothing is handled unless this callback routes the packet to them.

Where handlers are registered differs by language

.NET and Node register from the session; Java and Kotlin register alongside the stream node. C++ has no registration surface at all — every packet arrives at one callback, and telling them apart by name is the application's own code.

3.2 Writing a Packet Handler¶

Instead of returning a value, a handler writes the answer itself. A packet that arrived as a request is answered with reply; pushing to a client that is waiting for none uses send.

if (packet == ping_t::packet_name) {
    const auto request = payload.parse_json<ping_t> ();

    // reply_packet answers a request. To push to a client that is not
    // waiting for one, use write_packet instead.
    stream.reply_packet (zlink::message_t::from_json (pong_t{request.sent_at_unix_ms}))
      .async ();
    co_return;
}

3.3 Registration¶

A stream node opens a port and takes one session type. It is a registration of its own, separate from a mesh node.

// The port game clients connect to. One session type per stream node,
// and actor dispatch must be on for a session to relay to its player.
options.add_stream_node ("client-stream")
  .bind ("tcp://127.0.0.1:7421")
  .enable_actor_dispatch ()
  .register_session<game_session_t> ();

4. The Attaching Side — a Client Outside the Mesh¶

The attaching process

The code in this section goes into the process that attaches from outside the mesh. It pairs with the section above.

This process references no Framework. It references one separately published connector and nothing else.

// A game client outside the mesh. It references the connector only, never
// the Framework, and speaks to the port the stream node opened. Every call
// here is the connector's blocking form.
sc::connector_options_t options;
options.typed_codec = sc::json_typed_codec ();
options.endpoint = "tcp://127.0.0.1:7421";
options.connect_timeout = std::chrono::seconds (5);
options.request_timeout = std::chrono::seconds (5);
options.wait_timeout = std::chrono::seconds (5);
options.dispatch_mode = sc::dispatch_mode_t::immediate;

auto connector = sc::connector_factory_t::create (options);

require (connector.connect (), "connect");
std::cout << "connected: " << std::boolalpha << connector.is_connected () << std::endl;

// A request waits for its reply. Use send for one-way traffic; the server
// then answers with write_packet rather than reply_packet.
const auto sent_at = now_unix_ms ();
const auto pong = value_of (
  connector.request (ping_t{std::to_string (sent_at)}).submit<pong_t> (), "ping");

std::cout << "round trip: " << (now_unix_ms () - std::stoll (pong.sent_at_unix_ms)) << "ms"
          << std::endl;

The Node connector attaches over a WebSocket

The connector in the other four languages speaks TCP and names a tcp:// endpoint. The Node connector is published for the browser and speaks WebSocket, so both the stream node and the client name a ws:// endpoint.

5. What You See When You Run It¶

With the tutorial Server running as the README's Run section specifies, start StreamClient with the command below: the request it sends and response it receives appear on the StreamClient process's stdout, while connection and handler records appear on the Server process's stdout or in server.log.

dotnet run --project StreamClient/StreamClient.csproj
# connected: True
# round trip: 56ms

connected turning true means the connection was established, and the round trip is measured by the server echoing back the instant the client sent. Those two lines are the outside and the inside joined.

6. The Difference from a Mesh Call — the Connection Is the Target¶

In a mesh call the caller is a member of the mesh and the target is a name or an id. In STREAM the connection itself is the target. Nothing chooses among clients, and the server can send first.

When the connection drops that session ends. To carry state beyond one connection, bind the connection to an Actor — covered by Session and Actor.

Binding a connection to an entity — Session and Actor
State objects called by id — Spot · Actor
The whole session lifecycle and its options — STREAM
A running version of this chapter's code — the tutorial README in the examples repository