acts-package-state 0.18.0

acts package for state store
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
94

//! Acts postgres store

#![allow(rustdoc::bare_urls)]
// #![doc = include_str!("../README.md")]

mod config;
mod package;

#[cfg(test)]
mod tests;

use acts::{ActError, ActPackage, ActPlugin, ChannelOptions, Result};
use package::StatePackage;

const CONFIG_NAME: &str = "state";
#[derive(Clone)]
pub struct StatePackagePlugin;

#[async_trait::async_trait]
impl ActPlugin for StatePackagePlugin {
    fn on_init(&self, engine: &acts::Engine) -> Result<()> {
        if !engine.config().has(CONFIG_NAME) {
            println!(
                "skip the initialization of StatePackagePlugin for no 'state' secion in config file"
            );
            return Ok(());
        }
        let config = engine
            .config()
            .get::<config::StateConfig>(CONFIG_NAME)
            .map_err(|err| acts::ActError::Config(format!("get state config error: {err}")))?;

        let mut client = redis::Client::open(config.database_uri.as_str())
            .map_err(|err| acts::ActError::Config(format!("create redis client error: {err}")))?;

        redis::cmd("PING")
            .exec(&mut client)
            .map_err(|err| acts::ActError::Config(format!("ping redis error: {err}")))?;

        let meta = package::StatePackage::meta();
        engine.extender().register_package(&meta)?;

        let executor = engine.executor();
        let chan = engine.channel_with_options(&ChannelOptions {
            id: meta.id.to_string(),
            ack: true,
            r#type: "act".to_string(),
            state: "created".to_string(),
            uses: meta.id.to_string(),
            ..Default::default()
        });
        chan.on_message(move |e| {
            // check the params in inputs
            let Some(params) = e.inputs.get::<serde_json::Value>("params") else {
                let executor = executor.clone();
                let pid = e.pid.clone();
                let tid = e.tid.clone();
                executor
                    .act()
                    .fail(
                        &pid,
                        &tid,
                        ActError::Package("missing 'params' in inputs".to_string()).into(),
                    )
                    .unwrap();
                return;
            };

            // convert the params to StatePackage
            let pakage: StatePackage = serde_json::from_value(params).unwrap();
            match pakage.run(&client, &e.pid) {
                Ok(ref vars) => {
                    let executor = executor.clone();
                    let pid = e.pid.clone();
                    let tid = e.tid.clone();
                    let vars = vars.clone();
                    tokio::spawn(async move {
                        executor.act().complete(&pid, &tid, vars).unwrap();
                    });
                }
                Err(err) => {
                    let executor = executor.clone();
                    let pid = e.pid.clone();
                    let tid = e.tid.clone();
                    let err: acts::Vars = err.into();
                    tokio::spawn(async move {
                        executor.act().fail(&pid, &tid, err).unwrap();
                    });
                }
            }
        });
        Ok(())
    }
}