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.
Write in Rust, No virtual machine.
- 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]
The lib size is about 4.6mb now.
-
store collection extension support creating external store, please refer to the code under
crates/src/store/postgres. -
pakcage extension support creating custom package, please refer to the code under
example/pakcage.
The easiest way to get the latest version of acts is to install it via cargo
cargo add acts- Create and start the workflow engine by
engine.new(). - Load a yaml model to create a
workflow. - Deploy the model in step 2 by
engine.executor().model(). - Config events by
engine.channel(). - 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
uses: acts.transform.set
params:
a: 10
- name: step 2
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.set("a", 0);
// set the pid or auto generate by engine
vars.set("pid", "w1");
// 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);
});
}Please see examples
The model is a yaml format file. where there are different type of node, including [Workflow], [Branch], [Step].
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
# init with interrupt request to client
# and make sure complete the action with 'list' var
uses: acts.core.irq
- name: step 2
# workflow branches to run by condition
branches:
- name: branch 1
if: value > 100
steps:
- name: step 3
uses: acts.core.msg
- name: branch 2
if: value <= 100
steps:
- name: step 4
uses: acts.core.parallel
params:
in: '{{ list }}'
acts:
- uses: acts.core.irq
- name: final step
In the [Workflow], you can set the vars to init the workflow vars.
name: model name
vars:
- name: a
value: 100
steps:
- name: step1
uses: acts.transform.code
params: |
// get the a variable
let v = a + 100;
// do somthing elseThe 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.set("input", 3);
vars.set("pid", "w2");
executor.proc().start("m1", vars);
}In the [Workflow], you can set the options.exposes to filter the outputs.
name: model name
options:
exposes:
- name: output_key
steps:
- name: step1
uses: acts.transform.set
params:
output_key: 100Use steps to add step to the workflow
name: model name
steps:
- id: step1
name: step 1
- id: step2
name: step 2For more acts example, please see examples
Use the catches to capture the step error.
name: a catches example
id: catches
steps:
- name: prepare
id: prepare
uses: acts.core.irq
- name: step1
id: step1
uses: acts.core.irq
# catch the step errors
catches:
- id: catch1
on: err1
steps:
- name: catch step 1
uses: acts.core.irq
- id: catch2
on: err2
steps:
- name: catch step 2
uses: acts.core.irq
- name: final
id: finalUse the timeouts to check the task time.
name: a timeout example
id: timeout
steps:
- name: prepare
id: prepare
uses: acts.core.irq
- name: step1
id: step1
uses: acts.core.irq
# 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: finalUse 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 2For more acts example, please see examples
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);
}Please see the example example/pakcage.
Create a acts-server to interact with clients based on grpc.
please see more from acts-server
- rust https://github.com/yaojianpin/acts-channel
- python https://github.com/yaojianpin/acts-channel-py
- go https://github.com/yaojianpin/acts-channel-go
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/)
-
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)