websocat 4.0.0-alpha2

Command-line client for web sockets, like netcat/curl/socat for ws://.
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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

use rhai::{Engine, EvalAltResult, FnPtr, FuncArgs, NativeCallContext, Variant, AST};
use std::sync::{Arc, Mutex, Weak};
use tokio::time::Instant;
use tracing::error;

use crate::scenario_executor::{
    types::{Handle, Task},
    utils1::run_task,
};

use super::{
    types::{DiagnosticOutput, RandomnessSource, Registry},
    utils1::RhResult,
};

pub struct Scenario {
    pub ast: AST,
    pub engine: Engine,
    pub diagnostic_output: Mutex<DiagnosticOutput>,
    pub time_base: Instant,
    pub prng: Mutex<RandomnessSource>,
    pub registry: Registry,
}

pub trait ScenarioAccess {
    fn callback<T: Variant + Clone, A: FuncArgs>(&self, f: FnPtr, args: A) -> anyhow::Result<T>;
    fn get_scenario(&self) -> RhResult<Arc<Scenario>>;
}

pub fn load_scenario(
    s: &str,
    diagnostic_output: DiagnosticOutput,
    time_base: Instant,
    prng: RandomnessSource,
    registry: Registry,
) -> anyhow::Result<Arc<Scenario>> {
    let mut engine = Engine::RAW;

    crate::scenario_executor::all_functions::register_types(&mut engine);
    crate::scenario_executor::all_functions::register_functions(&mut engine);

    engine.set_max_expr_depths(1000, 1000);

    let ast = engine.compile(s)?;
    let mut scenario = Scenario {
        ast,
        engine,
        diagnostic_output: Mutex::new(diagnostic_output),
        time_base,
        prng: Mutex::new(prng),
        registry,
    };

    let scenario_arc: Arc<Scenario> = Arc::new_cyclic(move |weak_scenario_arc| {
        let weak_scenario_arc: Weak<Scenario> = weak_scenario_arc.clone();
        scenario
            .engine
            .set_default_tag(rhai::Dynamic::from(weak_scenario_arc));
        scenario
    });

    Ok(scenario_arc)
}

impl Scenario {
    pub fn execute(&self) -> anyhow::Result<Handle<Task>> {
        let task: Handle<Task> = self.engine.eval_ast(&self.ast)?;
        Ok(task)
    }
}

impl ScenarioAccess for NativeCallContext<'_> {
    fn callback<T: Variant + Clone, A: FuncArgs>(&self, f: FnPtr, args: A) -> anyhow::Result<T> {
        let scenario: Weak<Scenario> = self.tag().unwrap().clone().try_cast().unwrap();

        if let Some(s) = scenario.upgrade() {
            s.callback(f, args)
        } else {
            anyhow::bail!("Scenario is already terminated")
        }
    }

    fn get_scenario(&self) -> RhResult<Arc<Scenario>> {
        let scenario: Weak<Scenario> = self.tag().unwrap().clone().try_cast().unwrap();

        if let Some(s) = scenario.upgrade() {
            Ok(s)
        } else {
            Err(Box::new(EvalAltResult::ErrorRuntime(
                rhai::Dynamic::from(
                    "Scenario is already terminating, cannot make a callback into it",
                ),
                rhai::Position::NONE,
            )))
        }
    }
}

impl ScenarioAccess for Arc<Scenario> {
    fn callback<T: Variant + Clone, A: FuncArgs>(&self, f: FnPtr, args: A) -> anyhow::Result<T> {
        let scenario = self;

        let ret = f.call(&self.engine, &scenario.ast, args);
        if let Err(ref e) = ret {
            error!("Error from scenario task: {e}");
        };
        Ok(ret?)
    }

    fn get_scenario(&self) -> RhResult<Arc<Scenario>> {
        Ok(self.clone())
    }
}

pub async fn callback_and_continue<A: FuncArgs>(ctx: Arc<Scenario>, f: FnPtr, args: A) {
    match ctx.callback::<Handle<Task>, A>(f, args) {
        Ok(h) => run_task(h).await,
        Err(e) => error!("Error from scenario task: {e}"),
    };
}