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
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

//! Basic definitions for actors

use crate::{
    addr::*,
    context::ActorContext,
    message_queue::{MessageQueue, QueuePayload},
    runner::*,
};
use async_trait::async_trait;
use std::sync::Arc;
use tokio::sync::mpsc::UnboundedReceiver;

/// Represents the current lifecycle state of the actor
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ActorState {
    /// About to begin processing messages
    Starting,
    /// Normal operation
    Running,
    /// State indicating that the actor might stop, giving it a chance to react 
    Stopping,
    /// The actor is dead
    Stopped,
}

/// Value indicating whether a stopping actor should stop or continue running
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Stopping {
    Continue,
    Stop,
}

/// Inner implementation of actor creation logic
pub(crate) fn actor_create_impl<A: Actor, F: FnOnce(&mut ActorContext<A>) -> A + Send>(
    f: F,
) -> (
    A,
    Addr<A>,
    ActorContext<A>,
    UnboundedReceiver<QueuePayload<A>>,
) {
    let (msg_queue, msg_rx) = MessageQueue::new();
    let ret = Addr::<A> {
        msg_queue: Arc::from(msg_queue),
    };
    let weakaddr = ret.downgrade();
    let mut ctx = ActorContext::new(weakaddr);
    let actor = f(&mut ctx);
    (actor, ret, ctx, msg_rx)
}

/// The actor trait
#[async_trait]
pub trait Actor: 'static + Sized + Send {
    /// Starts the actor, consuming the underlying structure and returning an address to it.
    fn start(self) -> Addr<Self> {
        let (msg_queue, msg_rx) = MessageQueue::new();
        let ret = Addr::<Self> {
            msg_queue: Arc::from(msg_queue),
        };
        let weakaddr = ret.downgrade();
        tokio::spawn(actor_runner_loop(self, ActorContext::new(weakaddr), msg_rx));
        ret
    }
    /// Uses the given closure to build and start the actor, returning its' address.
    /// 
    /// This method should be used in case when access to [ActorContext] is required at the time of initialization.
    fn create<F: FnOnce(&mut ActorContext<Self>) -> Self + Send>(f: F) -> Addr<Self> {
        let (actor, ret, ctx, msg_rx) = actor_create_impl(f);
        tokio::spawn(actor_runner_loop(actor, ctx, msg_rx));
        ret
    }
    /// Called when the actor is about to begin processing messages.
    async fn started(&mut self, _ctx: &mut ActorContext<Self>) {}
    /// Called when the actor is in stopping state.
    /// 
    /// It can be overriden to react to this condition and possibly go back to normal operation.
    async fn stopping(&mut self, _ctx: &mut ActorContext<Self>) -> Stopping {
        Stopping::Stop
    }
    /// Called when the actors stops.
    async fn stopped(&mut self, _ctx: &mut ActorContext<Self>) {}
}

/// Trait implemented on [Actor]s to enable them to process messages of a given type
#[async_trait]
pub trait Handler<T: Send>: Actor {
    /// Type used to respond to incoming messages
    type Response: Send + 'static;
    /// The method used to handle incoming messages
    async fn handle(&mut self, msg: T, ctx: &mut ActorContext<Self>) -> Self::Response;
}