Client libraries

Table of contents

• Connecting to the RTI broker
  • Auto-connect
  • Waiting for connection
  • Threading and polling
  • Specifying a custom URL
  • Disconnecting
• Publish/subscribe text
• Publish/subscribe JSON
• Publish/subscribe protobuf
• Publishing entities and other Gensim messages
• Runtime control
• Fast-time control

The client libraries are designed with a compromise: similar naming and conventions, but adhering to the conventions of the respective language/platform. This page includes common use cases and examples for all the client libraries.

The client libraries are open-source and available on GitHub:

Language	Package	Install
Python	PyPI	pip install inhumate-rti
TypeScript/JS	npm	npm install inhumate-rti
C# / .NET	NuGet	dotnet add package Inhumate.RTI
C++	GitHub	See README

For C# in the context of the Unity game engine, see the Unity integration instead.

---

Connecting to the RTI broker

All client libraries connect to the RTI broker over WebSocket. By default, they connect to ws://127.0.0.1:8000, which is the default address of a locally running RTI.

You can override the URL by passing it explicitly, or by setting the RTI_URL environment variable.

Clients try to maintain a persistent connection and should reconnect automatically when losing connection.

Auto-connect

By default, the client connects automatically when constructed. Pass connect: false (or the equivalent for your language) to disable this and connect manually later.

import inhumate_rti as RTI

# Auto-connects immediately (default)
rti = RTI.Client(application="My App")

# Or disable auto-connect
rti = RTI.Client(application="My App", connect=False)
rti.connect()

import * as RTI from "inhumate-rti"

// Auto-connects immediately (default)
const rti = new RTI.Client({ application: "My App" })

// Or disable auto-connect
const rti = new RTI.Client({ application: "My App", connect: false })
rti.connect()

using Inhumate.RTI;

// Auto-connects immediately (default)
var rti = new RTIClient { Application = "My App" };

// Or disable auto-connect
var rti = new RTIClient(connect: false) { Application = "My App" };
rti.Connect();

#include "inhumaterti.hpp"

// Auto-connects immediately (default)
inhumate::rti::RTIClient rti("My App");

// Or disable auto-connect
inhumate::rti::RTIClient rti("My App", false);
rti.Connect();

Waiting for connection

Since connecting is asynchronous, you often need to wait for the connection to be established before proceeding.

import inhumate_rti as RTI

rti = RTI.Client(application="My App")

# Block until connected
rti.wait_until_connected()

# Or use a callback
rti.on("connect", lambda: print("Connected!"))

You can also pass wait=True to the constructor to block until connected:

rti = RTI.Client(application="My App", wait=True)
# Already connected here

import * as RTI from "inhumate-rti"

const rti = new RTI.Client({ application: "My App" })

// Execute a callback once connected
rti.whenConnected(() => {
    console.log("Connected!")
})

// Or listen for the connect event
rti.on("connect", () => {
    console.log("Connected!")
})

using Inhumate.RTI;

var rti = new RTIClient { Application = "My App" };

// Block until connected
rti.WaitUntilConnected();

// Or use an event callback
rti.OnConnected += () => {
    Console.WriteLine("Connected!");
};

#include "inhumaterti.hpp"

inhumate::rti::RTIClient rti("My App");

// Register a callback
rti.OnConnected([]() {
    std::cout << "Connected!" << std::endl;
});

// Poll to process messages (required in C++)
while (true) {
    rti.Poll();
}

Threading and polling

The Python and C# clients are multi-threaded by default — subscribe callbacks are called from a separate receive thread. The C++ client is single-threaded and requires you to call Poll() to process messages. The JavaScript/TypeScript client relies on the JavaScript event loop and requires no special handling.

In the default multi-threaded mode, the client runs the connection on a daemon thread and subscribe callbacks are called from that thread.

For applications that need single-threaded operation (e.g. game loops), pass a main_loop callback instead. The client will call your function repeatedly, with main_loop_idle_time seconds between calls:

import inhumate_rti as RTI

def main_loop():
    if rti.connected:
        rti.publish_text("foo", "bar")

rti = RTI.Client(
    application="My App",
    main_loop=main_loop,
    main_loop_idle_time=1.0
)
rti.connect()  # blocks, runs main_loop repeatedly

The JavaScript client uses the standard event loop — no threading or polling configuration is needed. Subscribe callbacks are called asynchronously as messages arrive.

import * as RTI from "inhumate-rti"

const rti = new RTI.Client({ application: "My App" })
rti.subscribeText("hello", (channel, message) => {
    console.log(`Received: ${message}`)
})

In the default multi-threaded mode, subscribe callbacks are called from a separate receive thread.

For applications that need single-threaded operation, pass polling: true to the constructor and call Poll() to process messages on your own schedule:

using Inhumate.RTI;

var rti = new RTIClient(polling: true) { Application = "My App" };
rti.WaitUntilConnected();

var done = false;
rti.Subscribe("hello", (string channel, object message) => {
    Console.WriteLine($"Received: {message}");
    done = true;
});

while (!done) {
    rti.Poll(100);
    Thread.Sleep(10);
}

The C++ client is single-threaded. You must call Poll() regularly to process incoming messages and trigger callbacks. Use PollForever() for a simple blocking loop:

#include "inhumaterti.hpp"

inhumate::rti::RTIClient rti("My App");

rti.OnConnected([]() {
    std::cout << "Connected!" << std::endl;
});

/* Process messages in your application loop */
while (running) {
    rti.Poll();
}

/* Or block forever */
rti.PollForever();

Specifying a custom URL

To connect to an RTI broker running on a different host or port, pass the URL explicitly or set the RTI_URL environment variable.

rti = RTI.Client(application="My App", url="ws://192.168.1.100:8000")

const rti = new RTI.Client({ application: "My App", url: "ws://192.168.1.100:8000" })

var rti = new RTIClient(url: "ws://192.168.1.100:8000") { Application = "My App" };

inhumate::rti::RTIClient rti("My App", true, "ws://192.168.1.100:8000");

Disconnecting

rti.disconnect()

rti.disconnect()

rti.Disconnect();

rti.Disconnect();

Publish/subscribe text

The simplest way to send and receive data is as plain text strings on named channels.

import inhumate_rti as RTI

rti = RTI.Client(application="My App")
rti.wait_until_connected()

rti.subscribe_text("hello", lambda content: print(f"Received: {content}"))
rti.publish_text("hello", "Hello World!")

The handler can optionally receive the channel name as the first argument:

rti.subscribe_text("hello", lambda channel, content: print(f"{channel}: {content}"))

import * as RTI from "inhumate-rti"

const rti = new RTI.Client({ application: "My App" })

rti.subscribeText("hello", (channel, message) => {
    console.log(`Received on ${channel}: ${message}`)
})

rti.whenConnected(() => {
    rti.publishText("hello", "Hello World!")
})

using Inhumate.RTI;

var rti = new RTIClient { Application = "My App" };
rti.WaitUntilConnected();

rti.Subscribe("hello", (string channel, object message) => {
    Console.WriteLine($"Received on {channel}: {message}");
});
rti.Publish("hello", "Hello World!");

#include "inhumaterti.hpp"

inhumate::rti::RTIClient rti("My App");

rti.Subscribe("hello", [](const std::string& channel, const std::string& message) {
    std::cout << "Received on " << channel << ": " << message << std::endl;
});

rti.PollForever();

Publish/subscribe JSON

For structured data without a predefined schema, you can publish and subscribe JSON.

rti.subscribe_json("my/data", lambda data: print(data["temperature"]))
rti.publish_json("my/data", {"temperature": 22.5, "unit": "celsius"})

rti.subscribeJSON("my/data", (channel, data) => {
    console.log(data.temperature)
})
rti.publishJSON("my/data", { temperature: 22.5, unit: "celsius" })

rti.SubscribeJson<Dictionary<string, object>>("my/data", (channel, data) => {
    Console.WriteLine(data["temperature"]);
});
rti.PublishJson("my/data", new { temperature = 22.5, unit = "celsius" });

rti.Subscribe("my/data", [](const std::string& channel, const std::string& message) {
    auto data = nlohmann::json::parse(message);
    std::cout << data["temperature"] << std::endl;
});
rti.Publish("my/data", "{\"temperature\": 22.5, \"unit\": \"celsius\"}");

Publish/subscribe protobuf

For typed, efficient messaging, use Protocol Buffer messages. The client libraries ship with the message types of the RTI contract — clients, channels, commands, measures and measurements, logs, runtime/launch/recorder control and fast-time — and corresponding pre-defined channel names.

The simulation information model — entities, positions, geometry, events and injects — is not part of the client libraries. It lives in the separate Gensim contract, generated into your own project. See publishing entities and other Gensim messages below.

import inhumate_rti as RTI

rti = RTI.Client(application="My App")
rti.wait_until_connected()

rti.subscribe(RTI.channel.measurement, RTI.proto.Measurement, lambda measurement:
    print(f"Measurement: {measurement.measure_id} = {measurement.value}")
)

measurement = RTI.proto.Measurement()
measurement.measure_id = "speed"
measurement.value = 42.0
rti.publish(RTI.channel.measurement, measurement)

Standard channel names are in RTI.channel (e.g. RTI.channel.measurement, RTI.channel.runtime_control, RTI.channel.commands). Standard protobuf message classes are in RTI.proto (e.g. RTI.proto.Measurement, RTI.proto.Clients, RTI.proto.RuntimeControl).

import * as RTI from "inhumate-rti"

const rti = new RTI.Client({ application: "My App" })

rti.subscribe(RTI.channel.measurement, RTI.proto.Measurement, (measurement) => {
    console.log(`Measurement: ${measurement.measureId} = ${measurement.value}`)
})

rti.whenConnected(() => {
    rti.publish(RTI.channel.measurement, RTI.proto.Measurement, { measureId: "speed", value: 42 })
})

Standard channel names are in RTI.channel (e.g. RTI.channel.measurement, RTI.channel.runtimeControl, RTI.channel.commands). Standard protobuf message classes are in RTI.proto (e.g. RTI.proto.Measurement, RTI.proto.Clients, RTI.proto.RuntimeControl).

using Inhumate.RTI;
using Inhumate.RTI.Proto;

var rti = new RTIClient { Application = "My App" };
rti.WaitUntilConnected();

rti.Subscribe<Measurement>(RTIChannel.Measurement, (channel, measurement) => {
    Console.WriteLine($"Measurement: {measurement.MeasureId} = {measurement.Value}");
});

rti.Publish(RTIChannel.Measurement, new Measurement { MeasureId = "speed", Value = 42 });

Standard channel names are in RTIChannel (e.g. RTIChannel.Measurement, RTIChannel.RuntimeControl, RTIChannel.Commands). Standard protobuf message classes are in the Inhumate.RTI.Proto namespace (e.g. Measurement, Clients, RuntimeControl).

#include "inhumaterti.hpp"

inhumate::rti::RTIClient rti("My App");

rti.Subscribe<inhumate::rti::proto::Measurement>(
    inhumate::rti::MEASUREMENT_CHANNEL,
    [](const std::string& channel, const inhumate::rti::proto::Measurement& measurement) {
        std::cout << "Measurement: " << measurement.measure_id()
                  << " = " << measurement.value() << std::endl;
    }
);

rti.PollForever();

Standard channel names are defined as constants in the inhumate::rti namespace (e.g. MEASUREMENT_CHANNEL, RUNTIME_CONTROL_CHANNEL, COMMANDS_CHANNEL). Standard protobuf message classes are in the inhumate::rti::proto namespace.

Publishing entities and other Gensim messages

Entities, positions, geometry, events and injects are the Gensim simulation information model. They used to be part of the RTI client libraries, but now live in their own contract, from which bindings are generated straight into your project — there is no Gensim package to install.

Channel names are unchanged on the wire (rti/entity, rti/geometry, …), so a client using generated Gensim bindings interoperates with clients that still use the older, all-in-one RTI client.

Generate the bindings with the contract tool, by listing what you need in an inhumate-contract.yml at your project root and running inhumate-contract with no arguments as part of your build. Then use the generated code alongside the RTI client, which stays responsible for the connection.

generate:
    - source: inhumate:gensim@1.0.0
      language: python
      output: myapp/gensim

The output directory is the package — the generated __init__.py makes it importable as one:

import inhumate_rti as RTI
from myapp import gensim as Gensim

rti = RTI.Client(application="My App")
rti.wait_until_connected()

rti.subscribe(Gensim.channel.entity, Gensim.proto.Entity, lambda entity:
    print(f"Entity: {entity.id}")
)

entity = Gensim.proto.Entity()
entity.id = "player-1"
entity.category = Gensim.proto.VEHICLE
rti.publish(Gensim.channel.entity, entity)

generate:
    - source: inhumate:gensim@1.0.0
      language: typescript
      output: src/gensim

import * as RTI from "inhumate-rti"
import * as Gensim from "./gensim/index.js"

const rti = new RTI.Client({ application: "My App" })

rti.subscribe(Gensim.channel.entity, Gensim.proto.Entity, (entity) => {
    console.log(`Entity: ${entity.id}`)
})

rti.whenConnected(() => {
    rti.publish(Gensim.channel.entity, Gensim.proto.Entity, { id: "player-1" })
})

generate:
    - source: inhumate:gensim@1.0.0
      language: csharp
      output: src/Generated

C# has no index file — the namespaces are the index:

using Inhumate.RTI;
using Inhumate.Gensim;           // GensimChannel
using Inhumate.Gensim.Proto;     // Entity, Geometry, ...

var rti = new RTIClient { Application = "My App" };
rti.WaitUntilConnected();

rti.Subscribe<Entity>(GensimChannel.Entity, (channel, entity) => {
    Console.WriteLine($"Entity: {entity.Id}");
});

rti.Publish(GensimChannel.Entity, new Entity { Id = "player-1", Category = EntityCategory.Vehicle });

Parameter is imported from the RTI contract rather than redefined, so don’t also generate the RTI contract — the Inhumate.RTI package already provides it.

Generated C++ only links against the libprotobuf it was generated for, so pin the same protobuf version the RTI C++ client is built with:

generate:
    - source: inhumate:gensim@1.0.0
      language: cpp
      output: generated
      protobufVersion: 3.11.2 # match the RTI C++ client

Compile the generated .pb.cc files into your target, and put both the output directory and the RTI client’s generated/ on the include path. Everything is behind one header:

#include "inhumaterti.hpp"
#include "inhumategensim.hpp"

inhumate::rti::RTIClient rti("My App");

rti.Subscribe<inhumate::gensim::proto::Entity>(
    inhumate::gensim::ENTITY_CHANNEL,
    [](const std::string& channel, const inhumate::gensim::proto::Entity& entity) {
        std::cout << "Entity: " << entity.id() << std::endl;
    }
);

rti.PollForever();

Runtime control

Most applications should react to load scenario, start, pause, end, stop and reset — see runtime control for the concept and the expected state handling.

Rather than subscribing to the runtime control channel and picking the messages apart yourself, use the runtime control helper shipped with each client library. It subscribes for you, keeps track of the current state and scenario, declares the runtimecontrol, scenario and timescale capabilities, and calls one method per message.

The same object also publishes runtime control — useful for test harnesses, scenario drivers and operator interfaces.

import inhumate_rti as RTI

class MyRuntime(RTI.RuntimeControl):

    def on_load_scenario(self, scenario, playback):
        print(f"Load scenario: {scenario.name}")
        return True  # return False to fail loading

    def on_start(self): print("Start")
    def on_pause(self): print("Pause")
    def on_end(self): print("End")
    def on_stop(self): print("Stop")
    def on_reset(self): print("Reset")
    def on_time_scale(self, time_scale): print(f"Time scale: {time_scale}")

rti = RTI.Client(application="My Sim")
runtime = MyRuntime(rti)
rti.wait_until_connected()

# The client state is maintained by the helper
if rti.state == RTI.proto.RuntimeState.RUNNING:
    ...

To drive other applications instead, publish runtime control and wait for state:

runtime = RTI.RuntimeControl(rti)
runtime.load_scenario("demo")
runtime.wait_for_application_state("My Sim", RTI.proto.RuntimeState.READY)
runtime.start()
runtime.set_time_scale(2.0)
runtime.stop()

import * as RTI from "inhumate-rti"

class MyRuntime extends RTI.RuntimeControl {

    onLoadScenario(scenario: any, playback: boolean): boolean {
        console.log(`Load scenario: ${scenario.name}`)
        return true // return false to fail loading
    }

    onStart() { console.log("Start") }
    onPause() { console.log("Pause") }
    onEnd() { console.log("End") }
    onStop() { console.log("Stop") }
    onReset() { console.log("Reset") }
    onTimeScale(timeScale: number) { console.log(`Time scale: ${timeScale}`) }
}

const rti = new RTI.Client({ application: "My Sim" })
const runtime = new MyRuntime(rti)

// The client state is maintained by the helper
if (rti.state === RTI.proto.RuntimeState.RUNNING) {
    // ...
}

To drive other applications instead, publish runtime control and wait for state:

const runtime = new RTI.RuntimeControl(rti)
runtime.loadScenario("demo")
await runtime.waitForApplicationState("My Sim", RTI.proto.RuntimeState.READY)
runtime.start()
runtime.setTimeScale(2.0)
runtime.stop()

using Inhumate.RTI;
using Inhumate.RTI.Proto;

class MyRuntime : RTIRuntimeControl {
    public MyRuntime(RTIClient rti) : base(rti) {}

    public override bool OnLoadScenario(RuntimeControl.Types.ScenarioSpecification scenario, bool playback) {
        Console.WriteLine($"Load scenario: {scenario.Name}");
        return true; // return false to fail loading
    }

    public override void OnStart() => Console.WriteLine("Start");
    public override void OnPause() => Console.WriteLine("Pause");
    public override void OnEnd() => Console.WriteLine("End");
    public override void OnStop() => Console.WriteLine("Stop");
    public override void OnReset() => Console.WriteLine("Reset");
    public override void OnTimeScale(double timeScale) => Console.WriteLine($"Time scale: {timeScale}");
}

var rti = new RTIClient { Application = "My Sim" };
var runtime = new MyRuntime(rti);
rti.WaitUntilConnected();

// The client state is maintained by the helper
if (rti.State == RuntimeState.Running) {
    // ...
}

To drive other applications instead, publish runtime control and wait for state:

var runtime = new RTIRuntimeControl(rti);
runtime.LoadScenario("demo");
runtime.WaitForApplicationState("My Sim", RuntimeState.Ready);
runtime.Start();
runtime.SetTimeScale(2.0);
runtime.Stop();

#include "inhumaterti.hpp"
#include "rtiruntimecontrol.hpp"

using namespace inhumate::rti;

class MyRuntime : public RTIRuntimeControl {
    public:
    using RTIRuntimeControl::RTIRuntimeControl;

    bool OnLoadScenario(const proto::RuntimeControl_ScenarioSpecification &scenario,
                        bool playback) override
    {
        std::cout << "Load scenario: " << scenario.name() << std::endl;
        return true; // return false to fail loading
    }

    void OnStart() override { std::cout << "Start" << std::endl; }
    void OnPause() override { std::cout << "Pause" << std::endl; }
    void OnEnd() override { std::cout << "End" << std::endl; }
    void OnStop() override { std::cout << "Stop" << std::endl; }
    void OnReset() override { std::cout << "Reset" << std::endl; }
    void OnTimeScale(double timeScale) override
    {
        std::cout << "Time scale: " << timeScale << std::endl;
    }
};

RTIClient rti("My Sim");
MyRuntime runtime(rti);

while (running) {
    rti.Poll();
    if (rti.state() == proto::RuntimeState::RUNNING) {
        double scale = runtime.has_time_scale() ? runtime.time_scale() : 1.0;
        /* advance simulation */
    }
}

To drive other applications instead, publish runtime control and wait for state:

RTIRuntimeControl runtime(rti);
runtime.LoadScenario("demo");
runtime.WaitForApplicationState("My Sim", proto::RuntimeState::READY);
runtime.Start();
runtime.SetTimeScale(2.0);
runtime.Stop();

The helper always subscribes to runtime control with immediate dispatch, so that stop, end and reset get through even while the client is buffering messages during a fast-time step.

Fast-time control

In fast-time, a controller — usually the CLI (rti fast) — owns simulation time and grants one step at a time, waiting for every worker to complete a step before granting the next. The same runtime control helper turns your application into a fast-time worker.

Enable it by passing fast_time (or a step function) when constructing the helper. That adds the fasttimeworker capability, subscribes to the fast-time control channel, acknowledges the run configuration automatically, and switches the client to buffered dispatch so that incoming messages are queued and delivered at the start of each step instead of in the middle of one.

There are two patterns:

• Step function — you hand the helper a function; it is called once per granted step and the helper reports the step complete when it returns (or failed if it raises).
• Get step grant — your own main loop asks the helper for the next grant, does the work, and reports the step complete itself. Use this when you already have a game/simulation loop that also needs to handle real time.

A grant carries the step number and the logical time interval — start_time, end_time and the time_step between them. Advance your simulation by exactly that interval, publish the results on the normal channels, and complete the step. Never advance on wall-clock time while in fast time.

Step function pattern:

import inhumate_rti as RTI

sim_time = 0.0

def step(grant):
    global sim_time
    sim_time = grant.start_time
    update(grant.time_step)   # your simulation work
    sim_time = grant.end_time
    # complete_step() is called automatically when this returns

rti = RTI.Client(application="My Sim")
runtime = RTI.RuntimeControl(rti, step_fn=step)

Get step grant pattern, in a main loop that also handles real time:

def main_loop():
    global sim_time
    if runtime.is_fast_time:
        # timeout=0 (default) — with main_loop the socket and the loop share a
        # thread, so blocking here would delay the incoming step grant
        grant = runtime.get_step_grant()
        if grant is not None:
            sim_time = grant.start_time
            update(grant.time_step)
            sim_time = grant.end_time
            runtime.complete_step(grant)
    elif rti.state == RTI.proto.RuntimeState.RUNNING:
        ...  # real-time update using wall clock and runtime.time_scale

rti = RTI.Client(application="My Sim", main_loop=main_loop)
runtime = RTI.RuntimeControl(rti, fast_time=True)
rti.connect()  # blocks, runs main_loop repeatedly

In the default multi-threaded mode (no main_loop), a blocking timeout is fine: runtime.get_step_grant(timeout=5).

If a step cannot be completed, report it: runtime.complete_step(grant, failed=True, reason="...").

Step function pattern:

import * as RTI from "inhumate-rti"

let simTime = 0.0

function step(grant: RTI.StepGrant) {
    simTime = grant.startTime
    update(grant.timeStep)   // your simulation work
    simTime = grant.endTime
    // completeStep() is called automatically when this returns
}

const rti = new RTI.Client({ application: "My Sim" })
const runtime = new RTI.RuntimeControl(rti, true, false, step)

Get step grant pattern, in an async loop that also handles real time:

const rti = new RTI.Client({ application: "My Sim" })
const runtime = new RTI.RuntimeControl(rti, true, true)

async function simLoop() {
    while (true) {
        if (runtime.isFastTime) {
            // the event loop keeps receiving messages while awaiting, so a
            // blocking timeout is safe here
            const grant = await runtime.getStepGrant(1000)
            if (grant !== null) {
                simTime = grant.startTime
                update(grant.timeStep)
                simTime = grant.endTime
                runtime.completeStep(grant)
            }
        } else if (rti.state === RTI.proto.RuntimeState.RUNNING) {
            // real-time update using wall clock and runtime.timeScale
        }
    }
}
simLoop()

If a step cannot be completed, report it: runtime.completeStep(grant, true, "...").

Step function pattern:

using Inhumate.RTI;

double simTime = 0;

var rti = new RTIClient { Application = "My Sim" };
var runtime = new RTIRuntimeControl(rti, stepFn: grant => {
    simTime = grant.StartTime;
    Update(grant.TimeStep);   // your simulation work
    simTime = grant.EndTime;
    // CompleteStep() is called automatically when this returns
});

Get step grant pattern, in a loop that also handles real time:

var rti = new RTIClient { Application = "My Sim" };
var runtime = new RTIRuntimeControl(rti, fastTime: true);

while (running) {
    if (runtime.IsFastTime) {
        var grant = runtime.GetStepGrant();
        if (grant != null) {
            simTime = grant.StartTime;
            Update(grant.TimeStep);
            simTime = grant.EndTime;
            runtime.CompleteStep(grant);
        }
    } else if (rti.State == RuntimeState.Running) {
        // real-time update using wall clock and runtime.TimeScale
    }
}

In polling mode (new RTIClient(polling: true)), use GetStepGrant(0) so the loop can call Poll() instead of blocking. In the default multi-threaded mode a blocking timeout is fine.

If a step cannot be completed, report it: runtime.CompleteStep(grant, true, "...").

Step function pattern:

#include "inhumaterti.hpp"
#include "rtiruntimecontrol.hpp"

using namespace inhumate::rti;

RTIClient rti("My Sim");

RTIRuntimeControl runtime(rti, /*subscribe=*/true, /*fastTime=*/false,
                          [](const StepGrant &grant) {
    simTime = grant.start_time;
    Update(grant.time_step);   /* your simulation work */
    simTime = grant.end_time;
    /* CompleteStep() is called automatically — or with failed=true if this throws */
});

Get step grant pattern — GetStepGrant() is non-blocking, since the C++ client is driven by Poll():

RTIClient rti("My Sim");
RTIRuntimeControl runtime(rti, /*subscribe=*/true, /*fastTime=*/true);

while (running) {
    rti.Poll();

    if (runtime.is_fast_time()) {
        if (auto grant = runtime.GetStepGrant()) {
            simTime = grant->start_time;
            Update(grant->time_step);
            simTime = grant->end_time;
            runtime.CompleteStep(*grant);
        }
    } else if (rti.state() == proto::RuntimeState::RUNNING) {
        /* real-time update using wall clock and runtime.time_scale() */
    }
}

If a step cannot be completed, report it: runtime.CompleteStep(*grant, true, "...").

Applications that only observe — viewers, recorders, dashboards — need no fast-time support at all. They just see the ordinary runtime states and data streams, at whatever rate the run produces them.