navactor 0.3.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

//!This module implements the `StdinActor`, which is responsible for reading commands from the
//!standard input stream in command-line interface (`CLI`) mode. When a `ReadAllCmd` message is
//!received, the actor reads all incoming lines from the standard input stream until the stream is
//!closed, sending a `TextMsg` message with the received text to the output handle for each line.
//!Once the end of the stream is reached, a `EndOfStream` message is sent to the next hop to
//!trigger any necessary cleanup and shutdown. This actor is only used in `CLI` mode and is used to
//!interact with the command-line interface by reading input commands from the user.

use crate::actor::Actor;
use crate::actor::Handle;
use crate::message::Envelope;
use crate::message::Message;
use crate::message::MtHint;
use async_trait::async_trait;
use tokio::io::stdin;
use tokio::io::AsyncBufReadExt;
use tokio::io::BufReader;
use tokio::sync::mpsc;

/// the stdin actor is only used in CLI mode.  it gets a single command to
/// read from stdin and it reads until the EOF.  once it sees EOF, it sends
/// a `EndOfStream` msg to the next hop to trigger any cleanup and shutdown.
pub struct StdinActor {
    pub receiver: mpsc::Receiver<Envelope<f64>>,
    pub output: Handle,
}

#[async_trait]
impl Actor for StdinActor {
    async fn handle_envelope(&mut self, envelope: Envelope<f64>) {
        let Envelope {
            message,
            respond_to,
            ..
        } = envelope;

        if matches!(message, Message::ReadAllCmd {}) {
            let mut lines = BufReader::new(stdin()).lines();

            while let Some(text) = lines.next_line().await.unwrap_or_else(|e| {
                log::error!("failed to read stream: {e:?}");
                None
            }) {
                let hint = if text.contains("gene_type") {
                    MtHint::GeneMapping
                } else {
                    MtHint::Update
                };
                let msg = Message::Content {
                    text,
                    hint,
                    path: None,
                };
                match self.output.tell(msg).await {
                    Ok(()) => {}
                    Err(e) => {
                        log::error!("cannot send message: {e:?}");
                        return;
                    }
                }
            }

            let complete_msg = Message::EndOfStream {};

            let senv = Envelope {
                message: complete_msg,
                respond_to,
                ..Default::default()
            };

            match self.output.send(senv).await {
                Ok(()) => {}
                Err(e) => {
                    log::error!("cannot send end-of-stream message: {e:?}");
                }
            }
        } else {
            log::warn!("unexpected: {message}");
        }
    }
    async fn stop(&self) {}
    async fn start(&mut self) {}
}

/// actor private constructor
impl StdinActor {
    const fn new(receiver: mpsc::Receiver<Envelope<f64>>, 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: StdinActor) {
        while let Some(envelope) = actor.receiver.recv().await {
            actor.handle_envelope(envelope).await;
        }
    }

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

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

    let actor_handle = Handle::new(sender);

    tokio::spawn(start(actor));

    actor_handle
}