acts-next 0.14.5

a fast, tiny, extensiable workflow engine
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

use crate::event::EventAction;
use crate::{utils, Act, Action, Engine, MessageState, TaskState, Vars, Workflow};
use serde_json::json;

#[tokio::test]
async fn sch_scher_next() {
    let engine = Engine::new();

    let rt = engine.runtime();
    let store = rt.cache().store();
    let workflow = Workflow::new().with_id(&utils::longid());

    let s = rt.clone();
    store.deploy(&workflow).unwrap();
    tokio::spawn(async move {
        let mut options = Vars::new();
        options.insert("pid".to_string(), json!(utils::longid()));
        s.start(&workflow, &options).unwrap();
    });

    let ret = rt.scher().next().await;
    assert!(ret);
}

#[tokio::test]
async fn sch_scher_task() {
    let engine = Engine::new();
    let rt = engine.runtime();
    let workflow = Workflow::new();
    let pid = utils::longid();
    let s = rt.clone();
    tokio::spawn(async move {
        let proc = s.create_proc(&pid, &workflow);
        proc.set_state(TaskState::Pending);
        s.launch(&proc)
    });

    let ret = rt.scher().next().await;
    assert!(ret);
}

#[tokio::test]
async fn sch_scher_start_default() {
    let engine = Engine::new();
    let rt = engine.runtime();
    let workflow = Workflow::new();
    let result = rt.start(&workflow, &Vars::new());
    assert!(result.is_ok());
}

#[tokio::test]
async fn sch_scher_start_with_vars() {
    let engine = Engine::new();
    let rt = engine.runtime();
    let workflow = Workflow::new();
    let mut vars = Vars::new();
    vars.insert("a".to_string(), json!(100));
    vars.insert("b".to_string(), json!("string"));

    let proc = rt.start(&workflow, &vars).unwrap();
    rt.scher().next().await;

    assert_eq!(proc.inputs().get::<i64>("a").unwrap(), 100);
    assert_eq!(proc.inputs().get::<String>("b").unwrap(), "string");
}

#[tokio::test]
async fn sch_scher_do_action() {
    let engine = Engine::new();
    let rt = engine.runtime();
    let sig = engine.signal(());
    let rx = sig.clone();
    let workflow = Workflow::new().with_step(|step| {
        step.with_name("step1").with_act(Act::irq(|act| {
            act.with_key("act1").with_ret("uid", json!("u1"))
        }))
    });
    let s = rt.clone();
    engine.channel().on_complete(move |_| rx.close());
    engine.channel().on_message(move |e| {
        if e.is_key("act1") && e.is_state(MessageState::Created) {
            let mut options = Vars::new();
            options.insert("uid".to_string(), json!("u1"));
            let action = Action::new(&e.pid, &e.tid, EventAction::Next, &options);
            s.do_action(&action).unwrap();
        }
    });
    let proc = rt.create_proc(&utils::longid(), &workflow);
    rt.launch(&proc);
    sig.recv().await;

    assert!(proc.state().is_success());
}