compio-actor 0.1.0

Actors for Compio
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

use std::{any::Any, future::Future, pin::Pin};

use super::{Actor, ActorExit, Handler, Message};
use crate::{
    Mailbox,
    mailbox::{MailboxEvent, Receiver},
    supervisor::Supervision,
};

type DeliveryFuture<'a, E> = Pin<Box<dyn Future<Output = Result<(), E>> + 'a>>;

trait Deliverable<A: Actor>: Send {
    fn deliver_to<'a>(
        self: Box<Self>,
        actor: &'a A,
        myself: &'a Mailbox<A>,
        state: &'a mut A::State,
    ) -> DeliveryFuture<'a, A::Error>;

    fn into_any(self: Box<Self>) -> Box<dyn Any + Send>;
}

struct Envelope<M: Message>(M);

impl<A, M> Deliverable<A> for Envelope<M>
where
    A: Handler<M>,
    M: Message,
{
    fn deliver_to<'a>(
        self: Box<Self>,
        actor: &'a A,
        myself: &'a Mailbox<A>,
        state: &'a mut A::State,
    ) -> DeliveryFuture<'a, A::Error> {
        let Self(message) = *self;
        Box::pin(async move { Handler::<M>::handle(actor, myself, message, state).await })
    }

    fn into_any(self: Box<Self>) -> Box<dyn Any + Send> {
        self
    }
}

pub(crate) struct Delivering<A: Actor>(Box<dyn Deliverable<A> + Send>);

impl<A: Actor> Delivering<A> {
    pub(crate) fn from_msg<M>(message: M) -> Self
    where
        A: Handler<M>,
        M: Message,
    {
        Self(Box::new(Envelope(message)))
    }

    pub(crate) fn recover<M>(self) -> M
    where
        A: Handler<M>,
        M: Message,
    {
        let Envelope(message) = *self
            .0
            .into_any()
            .downcast::<Envelope<M>>()
            .expect("message envelope type changed before delivery");
        message
    }

    pub(crate) fn deliver_to<'a>(
        self,
        actor: &'a A,
        myself: &'a Mailbox<A>,
        state: &'a mut A::State,
    ) -> DeliveryFuture<'a, A::Error> {
        self.0.deliver_to(actor, myself, state)
    }
}

pub(crate) async fn run<A: Actor>(
    actor: A,
    myself: Mailbox<A>,
    receiver: Receiver<A>,
    mut state: A::State,
    supervision: Option<&Supervision<A>>,
) -> ActorExit<A::Error> {
    let exit = match actor.post_start(&myself, &mut state).await {
        Ok(()) => {
            if let Some(supervision) = supervision {
                supervision.started(&myself);
            }
            loop {
                match receiver.recv().await {
                    MailboxEvent::Message(message) => {
                        if let Err(error) = message.deliver_to(&actor, &myself, &mut state).await {
                            break ActorExit::Failed(error);
                        }
                    }
                    MailboxEvent::Stop => break ActorExit::Stopped,
                }
            }
        }
        Err(error) => ActorExit::Failed(error),
    };

    finish(&actor, &myself, receiver, &mut state, exit).await
}

pub(crate) async fn finish<A: Actor>(
    actor: &A,
    myself: &Mailbox<A>,
    receiver: Receiver<A>,
    state: &mut A::State,
    mut exit: ActorExit<A::Error>,
) -> ActorExit<A::Error> {
    myself.begin_stop();
    if let Err(error) = actor.pre_stop(myself, state).await
        && matches!(exit, ActorExit::Stopped)
    {
        exit = ActorExit::Failed(error);
    }

    drop(receiver);
    if let Err(error) = actor.post_stop(myself, state).await
        && matches!(exit, ActorExit::Stopped)
    {
        exit = ActorExit::Failed(error);
    }
    exit
}