navactor 0.2.0

A cli tool for creating and updating actors from piped input
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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

use crate::actor::Actor;
use crate::actor::ActorHandle;
use crate::genes::DefaultGene;
use crate::genes::Gene;
use crate::message::Message;
use crate::message::MessageEnvelope;
use async_trait::async_trait;
use std::collections::HashMap;
use time::OffsetDateTime;
use tokio::sync::mpsc;

// TODO: reject late reporters based on genes
// TODO: reject late reporters based on genes
// TODO: reject late reporters based on genes
// TODO: reject late reporters based on genes
// TODO: by default, counters are ok for late reports but guages are not

/// the state actor is the heart of the system.  each digital twin has an
/// instance of actor keeping state computed from an arriving stream of
/// observations.
pub struct StateActor {
    pub receiver: mpsc::Receiver<MessageEnvelope>,
    pub output: Option<ActorHandle>,
    pub state: HashMap<i32, f64>,
    pub path: String,
    pub gene: Box<dyn Gene + Send + Sync>,
}

#[async_trait]
impl Actor for StateActor {
    async fn stop(&self) {}
    async fn handle_envelope(&mut self, envelope: MessageEnvelope) {
        let MessageEnvelope {
            message,
            respond_to,
            stream_from,
            ..
        } = envelope;

        match message {
            Message::InitCmd { .. } => {
                log::trace!("{} init started...", self.path);
                // this is an init so read your old events to recalculate your state
                if let Some(mut stream_from) = stream_from {
                    log::debug!("{} init", self.path);
                    let mut count = 0;

                    while let Some(message) = stream_from.recv().await {
                        if !self.update_state(message.clone()) {
                            stream_from.close();
                        } else {
                            count += 1;
                        }
                    }

                    log::debug!("{} finished init from {} events", self.path, count);

                    if let Some(respond_to) = respond_to {
                        respond_to
                            .send(Ok(Message::EndOfStream {}))
                            .expect("can not reply to init");
                    }
                }
            }
            Message::Update { .. } => {
                log::trace!("{} handling update", self.path);
                self.update_state(message.clone());
                if let Some(respond_to) = respond_to {
                    respond_to
                        .send(Ok(self.get_state_rpt()))
                        .expect("can not reply to ask");
                }
            }
            Message::Query { .. } => {
                // respond with a copy of our new state if this is an 'ask'
                if let Some(respond_to) = respond_to {
                    respond_to
                        .send(Ok(self.get_state_rpt()))
                        .expect("can not reply to ask");
                }
            }
            m => {
                log::warn!("unexpected message: {m:?}");
            }
        }

        // report the update to our state to the output actor
        if let Some(output_handle) = &self.output {
            output_handle
                .tell(self.get_state_rpt())
                .await
                .expect("cannot tell");
        }
    }
}

/// actor private constructor
impl StateActor {
    fn update_state(&mut self, message: Message) -> bool {
        let mut updated = false;
        match message {
            Message::Update { values, .. } => {
                updated = true;

                self.state.extend(&values); //update state
            }
            Message::Query { path: _ } => {
                // no impl because all "respond_to" requests get a state_rpt
            }
            Message::EndOfStream {} => {
                log::trace!("{} end of update stream", self.path);
            }
            m => {
                log::warn!("unexpected message in update stream: {:?}", m);
            }
        }
        updated
    }

    fn get_state_rpt(&self) -> Message {
        Message::StateReport {
            path: self.path.clone(),
            values: self.state.clone(),
            datetime: OffsetDateTime::now_utc(),
        }
    }

    /// state will populated from event store before any other processing via
    /// the lifecycle processing coordinated by the director
    fn new(
        path: String,
        receiver: mpsc::Receiver<MessageEnvelope>,
        output: Option<ActorHandle>,
        //gene: &Gene,
    ) -> Self {
        let gene = Box::new(DefaultGene::new());
        let state = HashMap::new();
        StateActor {
            path,
            receiver,
            output,
            state,
            gene,
        }
    }
}

/// actor handle public constructor
pub fn new(path: String, bufsz: usize, output: Option<ActorHandle>) -> ActorHandle {
    async fn start<'a>(mut actor: StateActor) {
        while let Some(envelope) = actor.receiver.recv().await {
            actor.handle_envelope(envelope).await;
        }
    }

    let (sender, receiver) = mpsc::channel(bufsz);

    let actor = StateActor::new(path, receiver, output);

    let actor_handle = ActorHandle::new(sender);

    tokio::spawn(start(actor));

    actor_handle
}