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

use crate::actor::Actor;
use crate::actor::ActorHandle;
use crate::message::ActorError;
use crate::message::Message;
use crate::message::MessageEnvelope;
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<MessageEnvelope>,
    pub output: ActorHandle,
}

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: MessageEnvelope) {
        let MessageEnvelope {
            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 = MessageEnvelope {
                        message: msg,
                        respond_to, // delegate responding to an ask to director
                        datetime,
                        ..Default::default()
                    };

                    self.output.send(senv).await.expect("cannot send");
                }
                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 }))
                            .expect("can not return error");
                    }
                }
            },
            m => {
                // forward everything else
                let senv = MessageEnvelope {
                    message: m.clone(),
                    respond_to,
                    ..Default::default()
                };

                self.output.send(senv).await.expect("cannot send");
            }
        }
    }
}

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

/// actor handle public constructor
pub fn new(bufsz: usize, output: ActorHandle) -> ActorHandle {
    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 = ActorHandle::new(sender);

    tokio::spawn(start(actor));

    actor_handle
}