processmanager 0.6.0

manage process lifecycles, graceful shutdown and process faults
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

//! Dynamic `ProcessManager` demo.
//!
//! Showcases
//! ---------
//! 1. starting a manager with one child,
//! 2. adding new `Runnable`s *after* the manager is already running,
//! 3. shutting the whole tree down gracefully.
//!
//! Build & run
//! -----------
//! ```bash
//! cargo run --example dynamic_add
//! ```
use processmanager::*;
use std::{sync::Arc, time::Duration};
use tokio::time::{interval, sleep};

/// A lightweight long-running component that emits a heartbeat.
struct Worker {
    id: usize,
    guard: Arc<RuntimeGuard>,
}

impl Worker {
    fn new(id: usize) -> Self {
        Self {
            id,
            guard: Arc::new(RuntimeGuard::default()),
        }
    }
}

impl Runnable for Worker {
    fn process_start(&self) -> ProcFuture<'_> {
        let id = self.id;
        let guard = self.guard.clone();

        Box::pin(async move {
            let ticker = guard.runtime_ticker().await;
            let mut beat = interval(Duration::from_secs(1));

            loop {
                match ticker.tick(beat.tick()).await {
                    ProcessOperation::Next(_) => println!("worker-{id}: heartbeat"),
                    ProcessOperation::Control(RuntimeControlMessage::Reload) => {
                        println!("worker-{id}: received *reload*")
                    }
                    ProcessOperation::Control(RuntimeControlMessage::Shutdown) => {
                        println!("worker-{id}: shutting down");
                        break;
                    }
                    // absorb any future control messages we don't explicitly handle
                    ProcessOperation::Control(_) => continue,
                }
            }
            Ok(())
        })
    }

    fn process_handle(&self) -> Arc<dyn ProcessControlHandler> {
        self.guard.handle()
    }
}

#[tokio::main]
async fn main() {
    // ------------------------------------------------------------------
    // 1. Manager with a single initial worker
    // ------------------------------------------------------------------
    let mgr = ProcessManagerBuilder::default()
        .pre_insert(Worker::new(0))
        .build();

    // We need to keep access to the manager after starting it, therefore
    // wrap it in an `Arc` and clone it for the spawning task.
    let mgr: Arc<ProcessManager> = Arc::new(mgr);
    let mgr_clone = Arc::clone(&mgr);

    tokio::spawn(async move {
        mgr_clone
            .process_start()
            .await
            .expect("manager encountered an error");
    });

    let handle = mgr.process_handle();

    // ------------------------------------------------------------------
    // 2. Dynamically add workers
    // ------------------------------------------------------------------
    println!("==> main: sleeping 3 s before adding worker-1");
    sleep(Duration::from_secs(3)).await;

    println!("==> main: adding worker-1");
    mgr.add(Worker::new(1));

    println!("==> main: sleeping 2 s");
    sleep(Duration::from_secs(2)).await;

    println!("==> main: adding worker-2");
    mgr.add(Worker::new(2));

    // ------------------------------------------------------------------
    // 3. Graceful shutdown after a short delay
    // ------------------------------------------------------------------
    println!("==> main: running 5 s before global shutdown");
    sleep(Duration::from_secs(5)).await;

    println!("==> main: initiating graceful shutdown");
    handle.shutdown().await;

    // Allow children to print their exit messages
    sleep(Duration::from_secs(1)).await;
}