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
135
136
137
138
139
140
141
142
143
144
145
146
147
148

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

// TODO:
// use rust std Path to update and persist petgraph graph Edges and
// lookup/upsert actor for each input record msg send

/// actor graph director creates a graph and instantiates all the actors that
/// it is forwarding commands to.  director also accepts metadata to create
/// and store graph edges to support arbitrary paths
pub struct Director {
    pub receiver: mpsc::Receiver<MessageEnvelope>,
    pub store_actor: Option<ActorHandle>,
    pub output: Option<ActorHandle>,
    pub actors: HashMap<String, ActorHandle>,
    namespace: String,
}

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

        match &message {
            Message::Update { path, .. } | Message::Query { path } => {
                let response = self.update_actor(path, message.clone()).await;

                // the store actor to jrnl this Update for this path here.
                if let Message::Update { path: _, .. } = message {
                    if let Some(store_actor) = &self.store_actor {
                        //store_actor.tell(message).await;
                        _ = store_actor.ask(message).await; // waiting for confirm before
                                                            // triggering pipeline and risking
                                                            // process termination while store
                                                            // actor is still writing
                    }
                }

                // reply with response if this is an 'ask' from the sender
                if let Some(respond_to) = respond_to {
                    respond_to
                        .send(response.clone())
                        .expect("can not reply to ask");
                }

                // forward response if output is configured
                if let Some(o) = &self.output {
                    let senv = MessageEnvelope {
                        message: response,
                        respond_to: None,
                        ..Default::default()
                    };
                    o.send(senv).await;
                }
            }
            Message::EndOfStream {} => {
                log::debug!("complete");

                if let Some(a) = &self.output {
                    let senv = MessageEnvelope {
                        message,
                        respond_to,
                        ..Default::default()
                    };
                    a.send(senv).await
                } else if let Some(respond_to) = respond_to {
                    // else we're the end of the line so reply if this is an ask
                    respond_to.send(message).expect("can not reply to ask");
                }
            }
            m => log::warn!("unexpected message: {:?}", m),
        }
    }
}

/// actor private constructor
impl Director {
    fn new(
        namespace: String,
        receiver: mpsc::Receiver<MessageEnvelope>,
        output: Option<ActorHandle>,
        store_actor: Option<ActorHandle>,
    ) -> Self {
        Director {
            namespace,
            actors: HashMap::new(),
            receiver,
            output,
            store_actor,
        }
    }

    async fn update_actor(&mut self, path: &String, message: Message) -> Message {
        let actor_is_in_init = self.actors.get(path).is_none();
        let actor = self
            .actors
            .entry(path.clone())
            .or_insert_with(|| state_actor::new(path.clone(), 8, None));

        match &self.store_actor {
            Some(store_actor) if actor_is_in_init => {
                log::debug!(
                    "{} handling actor '{}' messsage w/ actor init via store",
                    self.namespace,
                    path
                );
                actor
                    .integrate(message.clone(), String::from(path), store_actor)
                    .await
            }
            _ => {
                log::debug!("{} handling actor '{}' messsage", self.namespace, path);
                actor.ask(message.clone()).await
            }
        }
    }
}

/// actor handle public constructor
pub fn new(
    namespace: String,
    bufsz: usize,
    output: Option<ActorHandle>,
    store_actor: Option<ActorHandle>,
) -> ActorHandle {
    async fn start(mut actor: Director) {
        while let Some(envelope) = actor.receiver.recv().await {
            actor.handle_envelope(envelope).await;
        }
    }
    let (sender, receiver) = mpsc::channel(bufsz);
    let actor = Director::new(namespace.clone(), receiver, output, store_actor);
    let actor_handle = ActorHandle::new(sender);
    tokio::spawn(start(actor));
    log::debug!("{} started", namespace);
    actor_handle
}