logic-mesh 1.0.0

Control logic engine using event based and reactive blocks written in Rust.
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

// Copyright (c) 2022-2023, Radu Racariu.

use libhaystack::val::Value;

use crate::{
    base::{
        Status,
        input::InputProps,
        link::BaseLink,
        output::{BaseOutput, Output},
    },
    tokio_impl::WriterImpl,
};

use super::input::InputImpl;

pub type LinkImpl = BaseLink<WriterImpl>;
pub type OutputImpl = BaseOutput<LinkImpl>;

impl Output for OutputImpl {
    type Writer = <InputImpl as InputProps>::Writer;

    fn add_link(&mut self, link: BaseLink<Self::Writer>) {
        self.links.push(link);
    }

    fn set(&mut self, value: Value) {
        // Watch channels coalesce: each downstream receives the latest value;
        // there is no queue and no drop-on-full failure mode. We use
        // `send_if_modified` so writes that don't change the value don't wake
        // downstream blocks — convergent feedback loops quiesce at their
        // fixed point instead of busy-cycling forever. Status is part of the
        // payload comparison so an externally-managed status flip (set via
        // [`Output::set_effective_status`] from the actor task) wakes
        // downstream even when the value didn't change.
        let status = self.effective_status;
        for link in &mut self.links {
            if let Some(tx) = &link.tx {
                tx.send_if_modified(|current| {
                    if current.0 != value || current.1 != status {
                        current.0 = value.clone();
                        current.1 = status;
                        true
                    } else {
                        false
                    }
                });
            }
        }
        self.value = value;
    }

    fn emit_status(&mut self, status: Status) {
        // Re-emit the current value with the new status. Updates effective
        // status so subsequent `set(value)` calls keep using it.
        self.effective_status = status;
        let value = self.value.clone();
        for link in &mut self.links {
            if let Some(tx) = &link.tx {
                tx.send_if_modified(|current| {
                    if current.0 != value || current.1 != status {
                        current.0 = value.clone();
                        current.1 = status;
                        true
                    } else {
                        false
                    }
                });
            }
        }
    }
}