@spec eventually(
  (-> boolean()),
  keyword()
) :: :ok

Poll predicate every 50 ms until it returns truthy or timeout elapses.

eventually(fn -> length(messages("/sensor/data")) >= 5 end)
eventually(fn -> length(messages("/sensor/data")) >= 5 end, timeout: 10_000)

@spec expect(String.t(), (struct() -> boolean())) :: struct()

Assert that a message matching predicate arrives on topic.

Polls the internal buffer until the predicate returns truthy or the default timeout (5000 ms) elapses.

expect "/robot/state", &(&1.state == :running)

@spec expect(String.t(), module(), (struct() -> boolean())) :: struct()

@spec expect(String.t(), (struct() -> boolean()), keyword()) :: struct()

expect/2 with either an explicit message type or options.

# explicit type — registers the type and starts watching in one call
expect "/robot/state", MyPkg.Msg.RobotState, &(&1.state == :running)

# options
expect "/robot/state", &(&1.state == :running), timeout: 10_000, qos: sensor_qos

opts

• :timeout how long to wait, in milliseconds. Defaults to 5000.
• :namespace the namespace of the scenario node. Defaults to the namespace given to scenario/3, or "/".
• :qos a Rclex.QoS profile. If not specified, Rclex.QoS.profile_default/0 is used. Only applied when the subscription is started.

@spec expect(String.t(), module(), (struct() -> boolean()), keyword()) :: struct()

expect/3 with both an explicit message type and options.

expect "/robot/state", MyPkg.Msg.RobotState, &(&1.state == :running), timeout: 10_000

@spec expect_count(String.t(), pos_integer(), keyword()) :: [struct()]

Assert that count messages arrive on topic.

Returns the list of matched messages (next count in arrival order from cursor).

messages = expect_count "/sensor/data", 3

opts

• :timeout how long to wait, in milliseconds. Defaults to 5000.
• :namespace the namespace of the scenario node. Defaults to the namespace given to scenario/3, or "/".
• :qos a Rclex.QoS profile. If not specified, Rclex.QoS.profile_default/0 is used. Only applied when the subscription is started.

@spec expect_sequence(String.t(), [(struct() -> boolean())], keyword()) :: [struct()]

Assert a sequence of matchers in order on topic, using cursor semantics.

Each matcher in matchers is matched against the next unconsumed message. Returns the list of matched messages. opts apply to every step.

expect_sequence "/robot/state", [
  matches(%{mode: :idle}),
  matches(%{mode: :starting}),
  matches(%{mode: :running})
]

@spec messages(String.t()) :: [struct()]

Return all messages buffered for topic since the scenario began (or since watch/1,2 was called), in arrival order.

@spec publish(String.t(), struct(), keyword()) :: :ok

Publish msg on topic.

The message type is inferred from the struct module. A publisher is started lazily and reused across calls within the same scenario.

publish "/mission/start", %MissionStart{id: 42}
publish "/mission/start", %MissionStart{id: 42}, qos: Rclex.QoS.profile_sensor_data()

opts

• :namespace the namespace of the scenario node. Defaults to the namespace given to scenario/3, or "/".
• :qos a Rclex.QoS profile. If not specified, Rclex.QoS.profile_default/0 is used. Only applied when the publisher is started.

@spec refute_message(String.t()) :: :ok

Assert that no message arrives on topic within timeout ms (default 200 ms).

The message type must have been registered by an earlier watch/2, publish/2, or expect/3 call.

refute_message "/robot/error"

@spec refute_message(String.t(), module() | (struct() -> boolean()) | keyword()) ::
  :ok

refute_message/1 with a message type, a predicate, or options.

# explicit type — registers the type and starts watching in one call
refute_message "/robot/error", MyPkg.Msg.Error

# predicate against the already-registered type
refute_message "/robot/error", &(&1.code == :fatal)

# options
refute_message "/robot/error", timeout: 500

opts

• :timeout how long to wait, in milliseconds. Defaults to 200.
• :namespace the namespace of the scenario node. Defaults to the namespace given to scenario/3, or "/".
• :qos a Rclex.QoS profile. If not specified, Rclex.QoS.profile_default/0 is used. Only applied when the subscription is started.

@spec refute_message(
  String.t(),
  module() | (struct() -> boolean()),
  (struct() -> boolean()) | keyword()
) :: :ok

refute_message/2 with any two of a message type, a predicate, and options.

refute_message "/robot/error", MyPkg.Msg.Error, &(&1.code == :fatal)
refute_message "/robot/error", MyPkg.Msg.Error, timeout: 500
refute_message "/robot/error", &(&1.code == :fatal), timeout: 500

@spec refute_message(String.t(), module(), (struct() -> boolean()), keyword()) :: :ok

refute_message/3 with a message type, a predicate, and options.

refute_message "/robot/error", MyPkg.Msg.Error, &(&1.code == :fatal), timeout: 500

@spec watch(String.t()) :: :ok

Start buffering all messages arriving on topic.

The message type must have been registered via a prior watch/2 or publish/2 call. For the first subscription on a topic, prefer watch/2.

@spec watch(String.t(), module() | keyword()) :: :ok

Subscribe to topic and buffer all arriving messages.

Safe to call multiple times for the same topic; subsequent calls are no-ops. When msg_type is omitted, the type registered by an earlier watch/2 or publish/2 is reused.

watch "/robot/state", MyPkg.Msg.RobotState
watch "/robot/state", qos: Rclex.QoS.profile_sensor_data()

opts

• :namespace the namespace of the scenario node. Defaults to the namespace given to scenario/3, or "/".
• :qos a Rclex.QoS profile. If not specified, Rclex.QoS.profile_default/0 is used.

@spec watch(String.t(), module(), keyword()) :: :ok

watch/2 with both an explicit message type and options.

watch "/robot/state", MyPkg.Msg.RobotState, namespace: "/robot1"