hydra 0.1.40

A framework for writing fault tolerant, highly scalable applications with the Rust programming language.
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

use std::sync::Arc;

use serde::Deserialize;
use serde::Serialize;

use tokio::net::TcpListener;

use crate::Local;
use crate::Message;
use crate::NodeOptions;
use crate::NodeRemoteConnectorMessage;
use crate::Pid;
use crate::Process;
use crate::ProcessFlags;
use crate::SystemMessage;
use crate::node_local_panic;
use crate::node_remote_accepter;

#[derive(Serialize, Deserialize)]
pub enum NodeLocalSupervisorMessage {
    /// Occurs when a connector process requests the supervisor.
    RequestLocalSupervisor(Pid),
}

pub struct NodeLocalSupervisor {
    pub name: String,
    pub options: NodeOptions,
    #[allow(dead_code)]
    pub process: Pid,
}

impl Drop for NodeLocalSupervisor {
    fn drop(&mut self) {
        node_local_panic();
    }
}

async fn node_local_listener(supervisor: Arc<NodeLocalSupervisor>) {
    let listener = TcpListener::bind(supervisor.options.listen_address)
        .await
        .expect("Failed to bind socket for local node listener!");

    loop {
        let Ok((socket, _)) = listener.accept().await else {
            continue;
        };

        Process::spawn(node_remote_accepter(socket, supervisor.clone()));
    }
}

pub async fn node_local_supervisor(name: String, options: NodeOptions) {
    Process::set_flags(ProcessFlags::TRAP_EXIT);

    let supervisor = Arc::new(NodeLocalSupervisor {
        name,
        options,
        process: Process::current(),
    });

    let listener = Process::spawn_link(node_local_listener(supervisor.clone()));

    loop {
        let message = Process::receive::<NodeLocalSupervisorMessage>().await;

        match message {
            Message::User(NodeLocalSupervisorMessage::RequestLocalSupervisor(process)) => {
                let supervisor = Local::new(supervisor.clone());

                Process::send(
                    process,
                    NodeRemoteConnectorMessage::LocalNodeSupervisor(supervisor),
                );
            }
            Message::System(SystemMessage::Exit(pid, exit_reason)) => {
                if pid == listener {
                    panic!("Lost the local node listener: {:?}!", exit_reason);
                }
            }
            _ => unreachable!(),
        }
    }
}