navactor 0.1.3

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

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

/// 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>,
    path: String,
}

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

        if let Message::InitCmd {} = message {
            // 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);

                // once old state is recalculated, you can now process the new event
                if let Some(nm) = next_message {
                    self.update_state(nm);

                    // respond with a copy of our new state if this is an 'ask'
                    if let Some(r) = next_message_respond_to {
                        r.send(self.get_state_rpt()).expect("can not reply to ask");
                    }
                }
            }
        } else {
            self.update_state(message.clone());
        }

        // 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;
        }

        // respond with a copy of our new state if this is an 'ask'
        if let Some(respond_to) = respond_to {
            respond_to
                .send(self.get_state_rpt())
                .expect("can not reply to ask");
        }
    }
}

/// actor private constructor
impl StateActor {
    fn update_state(&mut self, message: Message) -> bool {
        let mut updated = false;
        match message {
            Message::Update { values, .. } => {
                log::trace!("{} update", self.path);
                updated = true;
                self.state.extend(&values); //update state
            }
            Message::Query { path: _ } => {
                log::trace!("{} inspect", self.path);
                // no impl because all "respond_to" requests get a state_rpt
            }
            Message::EndOfStream {} => {
                log::trace!("{} end of stream", self.path);
            }
            m => {
                log::warn!("unexpected message {:?}", m);
            }
        }
        updated
    }

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

    fn new(
        path: String,
        receiver: mpsc::Receiver<MessageEnvelope>,
        output: Option<ActorHandle>,
    ) -> Self {
        let state = HashMap::new(); // TODO: load from event store
        StateActor {
            path,
            receiver,
            output,
            state,
        }
    }
}

/// 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
}