navactor 0.2.4

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

use crate::actor::Actor;
use crate::actor::Handle;
use crate::message::ActorError;
use crate::message::Envelope;
use crate::message::Message;
use crate::message::Observations;
use async_trait::async_trait;
use time::format_description::well_known::Iso8601;
use time::OffsetDateTime;
use tokio::sync::mpsc;
extern crate serde;
extern crate serde_json;

/// actor accepts numerical json and converts into the internal state data msg
pub struct JsonDecoder {
    pub receiver: mpsc::Receiver<Envelope>,
    pub output: Handle,
}

fn extract_values_from_json(text: &str) -> Result<Observations, String> {
    let observations: Observations = match serde_json::from_str(text) {
        Ok(o) => o,
        Err(e) => return Err(e.to_string()),
    };
    Ok(observations)
}

fn extract_datetime(datetime_str: &str) -> OffsetDateTime {
    match OffsetDateTime::parse(datetime_str, &Iso8601::DEFAULT) {
        Ok(d) => d,
        Err(e) => {
            log::warn!("can not parse datetime {} due to: {}", datetime_str, e);
            OffsetDateTime::now_utc()
        }
    }
}

#[async_trait]
impl Actor for JsonDecoder {
    async fn stop(&self) {}
    async fn handle_envelope(&mut self, envelope: Envelope) {
        let Envelope {
            message,
            respond_to,
            datetime,
            ..
        } = envelope;
        // match the messages we know how to decode and forward them and everything else to the
        // next hop
        match &message {
            Message::PrintOneCmd { text } => match extract_values_from_json(text) {
                Ok(observations) => {
                    log::trace!("json parsed");
                    //forward observations to actor
                    let msg = Message::Update {
                        path: String::from(&observations.path),
                        datetime: extract_datetime(&observations.datetime),
                        values: observations.values,
                    };

                    // forward if output is configured
                    let senv = Envelope {
                        message: msg,
                        respond_to, // delegate responding to an ask to director
                        datetime,
                        ..Default::default()
                    };

                    self.output
                        .send(senv)
                        .await
                        .map_err(|e| {
                            log::error!("cannot send: {e:?}");
                        })
                        .ok();
                }
                Err(error) => {
                    log::warn!("json parse error: {}", error);
                    if let Some(respond_to) = respond_to {
                        let etxt = format!("json parse error: {error}");

                        respond_to
                            .send(Err(ActorError { reason: etxt }))
                            .map_err(|e| {
                                log::error!("cannot send: {e:?}");
                            })
                            .ok();
                    }
                }
            },
            m => {
                // forward everything else
                let senv = Envelope {
                    message: m.clone(),
                    respond_to,
                    ..Default::default()
                };

                self.output
                    .send(senv)
                    .await
                    .map_err(|e| {
                        log::error!("cannot send: {e:?}");
                    })
                    .ok();
            }
        }
    }
}

/// actor private constructor
impl JsonDecoder {
    const fn new(receiver: mpsc::Receiver<Envelope>, output: Handle) -> Self {
        Self { receiver, output }
    }
}

/// actor handle public constructor
#[must_use]
pub fn new(bufsz: usize, output: Handle) -> Handle {
    async fn start(mut actor: JsonDecoder) {
        while let Some(envelope) = actor.receiver.recv().await {
            actor.handle_envelope(envelope).await;
        }
    }

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

    let actor = JsonDecoder::new(receiver, output);

    let actor_handle = Handle::new(sender);

    tokio::spawn(start(actor));

    actor_handle
}