Self-loops

A direct self-loop transitions from a state back to itself, such as running → running. A state can also be part of an indirect cycle, such as running → paused → running. Both forms allow states on the cycle to be entered repeatedly, so their generated events have multi cardinality.

YAML model

quent: alpha
model: fsm_self_loop

fsms:
  Task:
    states:
      running:
        initial: true
        attributes:
          items_processed: u64
        to: [running, paused, completed]
      paused:
        to: [running]
      completed: {}

Instrumentation API

The task first repeats running through its direct self-loop. It then follows the indirect cycle through paused and back to running before entering completed once.

use instrumentation::{Context, FsmSelfLoop, Noop, Task};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let context = Context::<FsmSelfLoop>::try_new(Noop)?;
    let task = context.observer::<Task>().handle().running(0);

    // Direct self-loop: running -> running.
    let task = task.running(64);
    // Indirect cycle: running -> paused -> running.
    let task = task.paused().running(128);
    let _task = task.completed();

    Ok(())
}

#include "quent-tutorial-fsm-self-loop-cpp-bridge/gen/quent.hpp"

int main() {
  auto context = quent::Context::none();
  auto task = context.task_observer()->handle().running(
      quent::task::Running{.items_processed = 0});
  task = std::move(task).running(
      quent::task::Running{.items_processed = 64});
  auto paused = std::move(task).paused();
  task = std::move(paused).running(
      quent::task::Running{.items_processed = 128});
  auto completed = std::move(task).completed();
  return 0;
}

import quent_tutorial_fsm_self_loop as quent


def main() -> None:
    with quent.Context() as context:
        task = context.task_observer().handle()
        running = task.running(items_processed=0)
        running = running.running(items_processed=64)
        paused = running.paused()
        running = paused.running(items_processed=128)
        completed = running.completed()


if __name__ == "__main__":
    main()

Full code

• Rust source
• C++ source
• Python source