murmer 0.3.0

A distributed actor framework for Rust
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

//! ReadyHandle — deferred actor startup.
//!
//! [`ReadyHandle`] is returned by [`Receptionist::prepare()`](crate::Receptionist::prepare)
//! and [`System::prepare()`](crate::System::prepare). It represents an actor that is
//! registered in the receptionist (and therefore discoverable / message-queueable)
//! but whose supervisor has not yet been spawned.
//!
//! Messages sent to the actor's endpoint will queue in the unbounded mailbox
//! channel until [`ReadyHandle::start()`] is called to spawn the supervisor.
//!
//! If the `ReadyHandle` is dropped without calling `start()`, the internal
//! `DeregisterGuard` fires and the actor is automatically removed from the
//! receptionist.

use std::sync::{Arc, Mutex};

use tokio::sync::{mpsc, oneshot};

use crate::actor::{Actor, ActorContext, RemoteDispatch};
use crate::lifecycle::{SystemSignal, TerminationReason};
use crate::receptionist::DeregisterGuard;
use crate::supervisor::run_supervisor;
use crate::wire::{DispatchRequest, EnvelopeProxy};

/// A handle to a prepared (but not yet running) actor.
///
/// The actor is already registered in the receptionist and its endpoint is
/// functional — messages sent to the endpoint queue in the unbounded mailbox.
/// Calling [`start()`](ReadyHandle::start) spawns the supervisor which begins
/// draining the mailbox and processing messages.
///
/// Dropping without calling `start()` automatically deregisters the actor.
///
/// # Examples
///
/// ```rust,ignore
/// // Prepare an actor — it's discoverable but not processing yet
/// let (endpoint, ready) = system.prepare("cache/0", Cache, CacheState::new());
///
/// // Messages sent now will queue in the mailbox
/// endpoint.send(Warmup { key: "users".into() }).await.ok();
///
/// // Start processing — queued messages drain in order
/// ready.start();
/// ```
pub struct ReadyHandle<A: Actor + RemoteDispatch + 'static> {
    actor: Option<A>,
    state: Option<A::State>,
    ctx: Option<ActorContext<A>>,
    mailbox_rx: Option<mpsc::UnboundedReceiver<Box<dyn EnvelopeProxy<A>>>>,
    dispatch_rx: Option<mpsc::UnboundedReceiver<DispatchRequest>>,
    shutdown_rx: Option<oneshot::Receiver<()>>,
    system_rx: Option<mpsc::UnboundedReceiver<SystemSignal>>,
    exit_reason: Arc<Mutex<TerminationReason>>,
    _deregister_guard: Option<DeregisterGuard>,
}

impl<A: Actor + RemoteDispatch + 'static> ReadyHandle<A> {
    /// Create a new ReadyHandle. Called internally by `Receptionist::prepare()`.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        actor: A,
        state: A::State,
        ctx: ActorContext<A>,
        mailbox_rx: mpsc::UnboundedReceiver<Box<dyn EnvelopeProxy<A>>>,
        dispatch_rx: mpsc::UnboundedReceiver<DispatchRequest>,
        shutdown_rx: oneshot::Receiver<()>,
        system_rx: mpsc::UnboundedReceiver<SystemSignal>,
        exit_reason: Arc<Mutex<TerminationReason>>,
        deregister_guard: DeregisterGuard,
    ) -> Self {
        Self {
            actor: Some(actor),
            state: Some(state),
            ctx: Some(ctx),
            mailbox_rx: Some(mailbox_rx),
            dispatch_rx: Some(dispatch_rx),
            shutdown_rx: Some(shutdown_rx),
            system_rx: Some(system_rx),
            exit_reason,
            _deregister_guard: Some(deregister_guard),
        }
    }

    /// Spawn the supervisor, beginning message processing.
    ///
    /// All messages that were queued in the mailbox while the actor was in
    /// the prepared state will be processed in order.
    ///
    /// This consumes the `ReadyHandle`. The `DeregisterGuard` ownership
    /// transfers to the supervisor task — when the supervisor exits, the
    /// guard fires and deregisters the actor.
    pub fn start(mut self) {
        let actor = self.actor.take().expect("ReadyHandle: actor already taken");
        let state = self.state.take().expect("ReadyHandle: state already taken");
        let ctx = self.ctx.take().expect("ReadyHandle: ctx already taken");
        let mailbox_rx = self
            .mailbox_rx
            .take()
            .expect("ReadyHandle: mailbox_rx already taken");
        let dispatch_rx = self
            .dispatch_rx
            .take()
            .expect("ReadyHandle: dispatch_rx already taken");
        let shutdown_rx = self
            .shutdown_rx
            .take()
            .expect("ReadyHandle: shutdown_rx already taken");
        let system_rx = self
            .system_rx
            .take()
            .expect("ReadyHandle: system_rx already taken");
        let exit_reason = self.exit_reason.clone();
        let guard = self
            ._deregister_guard
            .take()
            .expect("ReadyHandle: deregister_guard already taken");

        tokio::spawn(async move {
            let _ = run_supervisor(
                actor,
                state,
                ctx,
                mailbox_rx,
                dispatch_rx,
                shutdown_rx,
                system_rx,
                exit_reason,
                guard,
            )
            .await;
        });
    }
}