acts 0.18.0

a fast, lightweight, extensiable workflow engine
Documentation

Acts workflow engine

Build Test

Acts is a fast, lightweight, extensiable workflow engine that executes workflows defined in YAML format.

Unlike traditional workflow engines (such as BPMN). Acts uses a message-driven architecture to execute and distribute messages.

Acts uses Step, Branch, Act to build the workflow. Step and Branch are the workflow stucture to run in sequence or to step into different branch by condition. Act is responsible for the action execution.

Key Features

Fast

Write in Rust, No virtual machine.

  1. bechmark with memory store
load                    time:   [57.334 µs 61.745 µs 66.755 µs]
deploy                  time:   [21.323 µs 23.811 µs 26.829 µs]
start                   time:   [80.320 µs 82.188 µs 84.336 µs]
act                     time:   [601.40 µs 636.69 µs 674.49 µs]

Lightweight

The lib size is about 4.6mb now.

Extensiable

  • store collection extension support creating external store, please refer to the code under store/sqlite.

  • pakcage extension support creating custom package, please refer to the code under example/pakcage.

Installation

The easiest way to get the latest version of acts is to install it via cargo

cargo add acts

Quickstart

  1. Create and start the workflow engine by engine.new().
  2. Load a yaml model to create a workflow.
  3. Deploy the model in step 2 by engine.executor().model().
  4. Config events by engine.channel().
  5. Start the workflow by engine.executor().model().
use acts::{Engine, Vars, Workflow};

#[tokio::main]
async fn main() {
    let engine = Engine::new().start().unwrap();

    // create yaml workflow model
    let model = r#"
    id: my_model
    name: my model
    steps:
      - name: step 1
        acts:
          - uses: acts.transform.set
            params:
              a: 10
      - name: step 2
        acts:
          - uses: acts.transform.code
            params: |
                return { data: a + 10 };
    "#;
    let workflow = Workflow::from_yml(model).unwrap();

    let executor = engine.executor();
    executor.model().deploy(&workflow).expect("fail to deploy workflow");

    let mut vars = Vars::new();

    // set the input value
    vars.insert("a".into(), 0.into());

    // set the pid or auto generate by engine
    vars.insert("pid".to_string(), "w1".into());

    // start workflow by model id
    executor.proc().start(&workflow.id, vars).expect("fail to start workflow");

    // create channel to receive messages
    let chan = engine.channel();

    chan.on_start(|e| {
        println!("start: {}", e.start_time);
    });

    chan.on_message(|e| {
        println!("message: {:?}", e);
    });

    chan.on_complete(|e| {
        println!("outputs: {:?} end_time: {}", e.outputs, e.end_time);
    });

    chan.on_error(|e| {
        println!("error on proc id: {} model id: {}", e.pid, e.mid);
    });
}

Examples

Please see examples

Model Usage

The model is a yaml format file. where there are different type of node, including [Workflow], [Branch], [Step] and [Act].

name: model name
# workflow default inputs vars
vars:
  - name: value
    value: 0

# schema for inputs and outputs
inputs:
  - name: value
    title: Value
    desc:  Set value when starting workflow
    type: number

outputs:
  - name: data
    title: Output data
    type: object

# the event to start the workflow
on:
  - id: event1
    uses: acts.event.manual
# workflow steps
steps:
  - name: step 1
    acts:
        # init with interrupt request to client
        # and make sure complete the action with 'list' var
      - name: init
        uses: acts.core.irq

  - name: step 2
    # workflow branches to run by condition
    branches:
      - name: branch 1
        if: value > 100
        steps:
          - name: step 3
            acts:
              - name: send a message
                uses: acts.core.msg

      - name: branch 2
        if: value <= 100
        steps:
          - name: step 4
            acts:
              - name: parallel send irq request
                uses: acts.core.parallel
                params:
                  in: '{{ list }}'
                  acts:
                    - uses: acts.core.irq
  - name: final step

Vars

In the [Workflow], you can set the vars to init the workflow vars.

name: model name
vars:
  - name: a
    value: 100

steps:
  - name: step1
    acts:
      - uses: acts.transform.code
        params: |
          // get the a variable
          let v = a + 100;
          // do somthing else

The vars can also be set by starting the workflow.

use acts::{Engine, Vars, Workflow};

#[tokio::main]
async fn main() {
  let engine = Engine::new().start().unwrap();
  let executor = engine.executor();

  let mut vars = Vars::new();
  vars.insert("input".into(), 3.into());
  vars.insert("pid".to_string(), "w2".into());

  executor.proc().start("m1", vars);
}

Options

In the [Workflow], you can set the options.exposes to filter the outputs.

name: model name
options:
  exposes:
    - name: output_key
steps:
  - name: step1
    acts:
      - uses: acts.transform.set
        params:
          output_key: 100

Steps

Use steps to add step to the workflow

name: model name
steps:
  - id: step1
    name: step 1
  - id: step2
    name: step 2

For more acts example, please see examples

step.catches

Use the catches to capture the step error.

name: a catches example
id: catches
steps:
  - name: prepare
    id: prepare
    acts:
      - uses: acts.core.irq
        key: init
  - name: step1
    id: step1
    acts:
      - uses: acts.core.irq
        key: act1
    # catch the step errors
    catches:
      - id: catch1
        on: err1
        steps:
          - name: catch step 1
            acts:
              - uses: acts.core.irq
                key: act2

      - id: catch2
        on: err2
        steps:
          - name: catch step 2
            acts:
              - uses: acts.core.irq
                key: act3

  - name: final
    id: final

step.timeouts

Use the timeouts to check the task time.

name: a timeout example
id: timeout
steps:
  - name: prepare
    id: prepare
    acts:
      - uses: acts.core.irq
        key: init
  - name: step1
    id: step1
    acts:
      - uses: acts.core.irq
        key: act1
    # check timeout rules
    timeouts:
      # 1d means one day
      # triggers act2 when timeout
      - uses: acts.core.irq
        id: act2
        if: $cost_in('1d')

      # 2h means two hours
      # triggers act3 when timeout
      - uses: acts.core.irq
        id: act2
        if: $cost_in('2h')

  - name: final
    id: final

Branches

Use branches to add branch to the step

name: model name
steps:
  - id: step1
    name: step 1
    branches:
      - id: b1
        if: v > 0
        steps:
          - name: step a
          - name: step b
      - id: b2
        else: true
        steps:
          - name: step c
          - name: step d
  - id: step2
    name: step 2

Acts

Use acts to create act to interact with client， or finish a special function through several act type.

name: model name
options:
  exposes:
    - name: output_key
steps:
  - name: step1
    acts:
      # send message to client
      - uses: acts.core.msg
        key: msg1
        params:
          a: 1

      # irq is an act to send a request from acts server
      # the client can complete the act and pass data to serever
      - uses: acts.core.irq
        key: init
        name: my act init

        # passes data to the act
        params:
          a: 6

For more acts example, please see examples

Store

Current, the supported store features: store-sqlite, store-postgres, store-sled, store-redis, store-nats

You can add custom store support as follow:

use acts::{Engine, Result, KvStore, ScanOptions, data};
use std::sync::Arc;

pub struct MyStore;
impl KvStore for MyStore {
    fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
        Ok(None)
    }

    fn put(&self, key: &str, value: Vec<u8>) -> Result<()> {
        Ok(())
    }

    fn delete(&self, key: &str) -> Result<()> {
        Ok(())
    }

    fn scan_prefix(&self, key: &str, options: ScanOptions) -> Result<Vec<(String, Vec<u8>)>> {
        Ok(vec![])
    }

}

#[tokio::main]
async fn main() {
    let engine = acts::Engine::new().start().unwrap();
    let store: Arc<dyn KvStore + 'static> = Arc::new(MyStore);
    engine.extender().register_store(store);
}

Package

Please see the example example/pakcage.

Acts-Server

Create a acts-server to interact with clients based on grpc. please see more from acts-server

Client channels

Roadmap

acts:

  • runtime

    • model (Workflow, Branch, Step, Act)
    • scheduler (Config, Builder, Node, Process, Task, Queue, Event)
    • javascript runner
    • cache
    • plugin register
    • package register
    • message channel

  • store

    • memory
    • sqlite
    • postgres
    • nats
    • redis
    • sled

  • packages

    • core

      • irq
      • msg
      • block
      • action
      • parallel
      • sequence
      • subflow

    • event

      • manual
      • hook
      • chat
      • schedule

    • transform

      • set
      • code

  • doc (doc/)

  • store extension

    • sqlite
    • postgres

  • package extension

    • form (plugins/form)
    • ai (plugins/ai)
    • state (plugins/state)
    • http (plugins/http)
    • shell (plugins/shell) support nushell, bash and powershell
    • pubsub (plugins/pubsub)
    • observability (plugins/obs)
    • database (plugins/database)
    • mail (plugins/mail)