Callbacks and Subscriptions

The Broadcaster Model

Operations you start return a Task, but plenty happens that you did not initiate: players join, properties change, events arrive. These server-initiated signals are delivered through public callback members on RealtimeClient, OnPlayerJoined, OnRoomJoined, OnDisconnected and the rest, each of which is a RealtimeCore::Common::Broadcaster.

A Broadcaster<Signature> is a multicast callback list. Subscribe(handler) registers any callable matching the signature and returns a Subscription handle. When the client broadcasts, every live, non-blocked subscriber is invoked in turn.

All callbacks fire inside Service(), on the thread that calls it. There is no cross-thread delivery, and nothing fires between two Service() calls; see the threading model for the wider rules.

Subscribing to Client Callbacks

Subscribing takes a lambda, or any callable, matching the callback's signature.

Custom events are the one exception to the member-per-callback pattern: they are registered with SubscribeEvent(), which filters by event code and returns the same Subscription type, and are covered on the Custom Events page.

RealtimeCore::Common::Subscription joinSub = client.OnPlayerJoined.Subscribe(
    [](const PlayerView& player) {
        std::printf("player %d joined\n", player.Number);
    });

RealtimeCore::Common::Subscription eventSub = client.SubscribeEvent(
    [](uint8_t eventCode, int senderId, std::span<const uint8_t> data) {
        std::printf("event %d from player %d (%zu bytes)\n", eventCode, senderId, data.size());
    });

/* Nothing is subscribed: realtime_create attaches every callback to the queue,
   and dispatch is a switch over the event type inside the polled batch. */
switch (e->type)
{
    case RT_PlayerJoined:
        /* `origin` is the player number; the blob is the full player record. */
        printf("player %u joined\n", e->origin);
        break;

    case RT_Event:
        /* `origin` is the sender, `counter` the event code. */
        printf("event %u from player %u (%d bytes)\n",
               e->counter, e->origin, e->blobLength);
        break;

    default:
        break;
}

/* Over the C API there is no Broadcaster and nothing to subscribe to.
   Every callback in the table at the bottom of this page is pre-subscribed
   by realtime_create and arrives as one of the broadcaster events 100 to
   120; the C API page maps each event to its callback and lists what
   origin, counter and the blob carry. */

Subscription Lifetime

A raw Subscription does not auto-unsubscribe when it is destroyed; the handler stays registered until you call Unsubscribe(). IsSubscribed() and operator bool report whether the handle is still attached.

Anything the handler captures must stay valid for as long as the subscription is live. A lambda capturing this on an object that is destroyed while still subscribed is the classic dangling-callback bug, tie the subscription's lifetime to the captured object's lifetime, which is exactly what the RAII helpers are for.

ScopedSubscription

ScopedSubscription is the RAII wrapper: it unsubscribes automatically when destroyed and is the recommended default for holding a subscription. It is move-only, constructs implicitly from a Subscription, and Release() hands the raw handle back if you need to manage it manually after all.

Held as a class member, it guarantees the handler never outlives the object it captures.

class ScoreBoard
{
public:
    explicit ScoreBoard(RealtimeClient& client)
        : _playerJoined(client.OnPlayerJoined.Subscribe(
              [this](const PlayerView& player) { _names.push_back(player.Name); }))
    {
    }

private:
    std::vector<RealtimeCore::Common::StringType> _names;
    RealtimeCore::Common::ScopedSubscription      _playerJoined;
};

/* There is no subscription to scope, so a system is fed by the dispatcher
   instead of registering with the client. */
typedef struct
{
    char    names[32][64];
    int32_t count;
} ScoreBoard;

static void ScoreBoardOnEvent(ScoreBoard* board, const RealtimeEvent* e, const uint8_t* blob)
{
    if (e->type != RT_PlayerJoined || e->blobOffset < 0 || board->count >= 32)
    {
        return;
    }

    /* Player blob: [int32 number][string name][string userId][map][u8][u8] */
    rt_reader r = rt_read(blob + e->blobOffset, e->blobLength);
    (void)rt_get_i32(&r); /* number, also available as e->origin */

    int32_t     nameLen = 0;
    const char* name    = rt_get_str(&r, &nameLen);

    if (nameLen > (int32_t)sizeof(board->names[0]) - 1)
    {
        nameLen = (int32_t)sizeof(board->names[0]) - 1;
    }

    /* The blob is only valid until the next flush, so copy the name out. */
    memcpy(board->names[board->count], name, (size_t)nameLen);
    board->names[board->count][nameLen] = '\0';
    ++board->count;
}

/* Lifetime problems move with the model: instead of a dangling handler,
   the risk is a dangling pointer into the blob. Copy anything a system
   retains out of the blob before the frame's realtime_event_queue_flush. */

SubscriptionBag

SubscriptionBag collects any number of subscriptions for bulk lifetime management. Add with +=, tear everything down with UnsubscribeAll() (or by destroying the bag) and inspect with Count() and IsEmpty().

One bag typically covers one gameplay system.

RealtimeCore::Common::SubscriptionBag subscriptions;

subscriptions += client.OnRoomJoined.Subscribe([] { std::puts("room joined"); });
subscriptions += client.OnRoomLeft.Subscribe([] { std::puts("room left"); });
subscriptions += client.OnPlayerLeft.Subscribe(
    [](int playerNumber, bool isInactive) {
        std::printf("player %d left (inactive: %d)\n", playerNumber, isInactive);
    });

// later, when the system shuts down:
subscriptions.UnsubscribeAll();

/* One dispatch function per system takes the place of one bag per system. */
static void RoomSystemOnEvent(const RealtimeEvent* e)
{
    switch (e->type)
    {
        case RT_RoomJoined:
            puts("room joined");
            break;

        case RT_RoomLeft:
            puts("room left");
            break;

        case RT_PlayerLeft:
            /* `counter` is 1 when the player stays reserved as inactive. */
            printf("player %u left (inactive: %u)\n", e->origin, e->counter);
            break;

        default:
            break;
    }
}

/* Shutting the system down means no longer calling it: */
if (roomSystemActive)
{
    RoomSystemOnEvent(e);
}

/* There is nothing to unsubscribe, so bulk teardown becomes a flag on your
   own dispatcher. Events keep arriving on the queue either way, and
   unhandled ones are simply dropped when the batch is flushed. */

Blocking a Subscription

Block() temporarily silences a handler without unsubscribing it; Unblock() reactivates it and IsBlocked() queries the state. This is handy for ignoring reactions to changes you are about to cause yourself, or for muting UI updates during a cutscene or loading screen without losing the registration.

Subscribing and Unsubscribing During Dispatch

Dispatch is re-entrancy safe: a handler may subscribe new handlers or unsubscribe any handler, including itself, while a broadcast is running. The change is deferred and applied once the current broadcast finishes, so the in-flight broadcast still runs against the subscriber list it started with.

Accessing Realtime from within a Callback

Calling back into the client from inside a handler is safe, you can send events, change properties or start operations such as JoinRoom() directly from a callback. The one exception is the pump itself: never call Service(true) or the DispatchIncomingCommands() method from within a callback, because they are what is currently executing your callback and must not re-enter.

Callback Overview

Custom events are absent from this table because they are not a Broadcaster member, register them with SubscribeEvent() instead, see Custom Events.

The full signatures are listed in the client callbacks reference.