aspartam 0.1.0

Minimalistic actor framework based on tokio, inspired by actix
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

//! Internal message queue implementation

use crate::{actor::*, error::*};
use tokio::sync::{mpsc, oneshot};

mod envelope;
use envelope::*;

/// The type used for wrapping enqueued messages
pub(crate) type QueuePayload<T> = Box<dyn EnvelopeProxy<T> + Send>;

/// Message queue wraps a sender for [QueuePayload]
#[derive(Debug)]
pub(crate) struct MessageQueue<T: Actor> {
    tx: mpsc::UnboundedSender<QueuePayload<T>>,
}

impl<T: Actor> Clone for MessageQueue<T> {
    fn clone(&self) -> Self {
        Self {
            tx: self.tx.clone(),
        }
    }
}

impl<T: Actor> MessageQueue<T> {
    /// New message queue with its' corresponding receiver
    pub fn new() -> (Self, mpsc::UnboundedReceiver<QueuePayload<T>>) {
        let (tx, rx) = mpsc::unbounded_channel();
        (Self { tx }, rx)
    }
    pub fn send<M>(
        &self,
        msg: M,
    ) -> Result<oneshot::Receiver<<T as Handler<M>>::Response>, ActorError>
    where
        T: Handler<M>,
        M: 'static + Send,
    {
        let (tx, rx) = oneshot::channel();
        let envelope = Envelope::new(msg, tx).pack();
        self.tx.send(envelope)?;
        Ok(rx)
    }
    pub fn try_send<M>(&self, msg: M) -> Result<(), ActorError>
    where
        T: Handler<M>,
        M: 'static + Send,
    {
        let envelope = Envelope::new_no_sender(msg).pack();
        Ok(self.tx.send(envelope)?)
    }
    pub fn do_send<M>(&self, msg: M)
    where
        T: Handler<M>,
        M: 'static + Send,
    {
        let envelope = Envelope::new_no_sender(msg).pack();
        // do send just ignores errors
        let _ = self.tx.send(envelope);
    }
}