mm1-sup 0.7.23

An Erlang-style actor runtime 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

use std::fmt;
use std::time::Duration;

use mm1_address::address::Address;
use mm1_common::errors::chain::StdErrorDisplayChainExt;
use mm1_common::errors::error_of::ErrorOf;
use mm1_common::log;
use mm1_core::context::{
    Fork, Linking, Messaging, Quit, ShutdownErrorKind, Start, Stop, Tell, Watching,
};
use mm1_proto::{Message, message};
use mm1_proto_sup::common as sup_common;
use mm1_proto_system::StartErrorKind;

use crate::common::child_spec::InitType;
use crate::mixed::ChildType;
type ChildSpec<F> = crate::common::child_spec::ChildSpec<F, ChildType>;

#[message(base_path = ::mm1_proto)]
pub(crate) struct Started<K> {
    pub child_id: K,
    pub address:  Address,
}

#[message(base_path = ::mm1_proto)]
pub(crate) struct StartFailed<K> {
    pub child_id: K,
}
#[message(base_path = ::mm1_proto)]
pub(crate) struct StopFailed {
    pub address: Address,
    pub reason:  ErrorOf<ShutdownErrorKind>,
}

pub(crate) async fn shutdown<Ctx>(
    ctx: &mut Ctx,
    sup_address: Address,
    address: Address,
    stop_timeout: Duration,
) where
    Ctx: Messaging + Fork + Stop + Watching,
{
    if let Err(reason) = ctx.shutdown(address, stop_timeout).await {
        send_report(ctx, sup_address, StopFailed { address, reason }).await;
    }
}

pub(crate) async fn run<K, Runnable, Ctx>(
    ctx: &mut Ctx,
    sup_address: Address,
    child_id: K,
    child_spec: ChildSpec<Runnable>,
) where
    Ctx: Linking + Start<Runnable> + Quit + Messaging,
    K: fmt::Display,
    Started<K>: Message,
    StartFailed<K>: Message,
{
    let ChildSpec {
        launcher: factory,
        init_type,
        child_type: _,
        stop_timeout: _,
        announce_parent,
    } = child_spec;
    let runnable = factory;

    match do_start(ctx, runnable, init_type).await {
        Ok(address) => {
            log::info!(sup_address = %sup_address, child_id = %child_id, address = %address, "started");
            if announce_parent {
                log::debug!(sup_address = %sup_address, child_id = %child_id, "announcing parent");
                ctx.tell(
                    address,
                    sup_common::SetParent {
                        parent: sup_address,
                    },
                )
                .await
                .ok();
            }
            send_report(ctx, sup_address, Started { child_id, address }).await;
        },
        Err(reason) => {
            log::info!(sup_address = %sup_address, child_id = %child_id, reason = %reason.as_display_chain(), "failed to start");
            send_report(ctx, sup_address, StartFailed { child_id }).await;
        },
    };
}

async fn send_report<Ctx, M>(ctx: &mut Ctx, to: Address, report: M)
where
    Ctx: Messaging,
    M: Message,
{
    ctx.tell(to, report).await.expect("failed to send report");
}

async fn do_start<Runnable, Ctx>(
    ctx: &mut Ctx,
    runnable: Runnable,
    init_type: InitType,
) -> Result<Address, ErrorOf<StartErrorKind>>
where
    Ctx: Linking + Start<Runnable>,
{
    match init_type {
        InitType::NoAck => {
            ctx.spawn(runnable, true)
                .await
                .map_err(|e| e.map_kind(StartErrorKind::Spawn))
        },
        InitType::WithAck { start_timeout } => ctx.start(runnable, true, start_timeout).await,
    }
}