@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 timeout options.

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

# custom timeout (ms)
expect "/robot/state", &(&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

@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.

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()
) :: :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}

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

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

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

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

Assert that no message matching predicate arrives on topic within timeout ms.

refute_message "/robot/error", MyPkg.Msg.Error, &(&1.code == :fatal)
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()) :: :ok

Subscribe to topic with msg_type and buffer all arriving messages.

Safe to call multiple times for the same topic; subsequent calls are no-ops.

watch "/robot/state", MyPkg.Msg.RobotState