rust-tokio-supervisor 0.1.4

A Rust tokio supervisor with declarative task supervision, restart policy, shutdown coordination, and observability.
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
135
136

//! Supervisor role adapter.
//!
//! The adapter owns the bridge from [`SupervisorRole`] lifecycle methods to the
//! existing [`TaskFactory`](crate::task::factory::TaskFactory) runtime contract.

use crate::role::context::supervisor::SupervisorContext;
use crate::role::lifecycle::RoleLifecyclePhase;
use crate::role::result::supervisor::SupervisorRoleError;
use crate::role::traits::supervisor::SupervisorRole;
use crate::runtime::supervisor::Supervisor;
use crate::task::context::TaskContext;
use crate::task::factory::{BoxTaskFuture, Service, TaskResult};
use std::sync::Arc;
use tokio::sync::Mutex;

/// Adapter that turns a supervisor role into a task factory service.
pub struct SupervisorRoleAdapter<T> {
    /// Role instance protected across runtime task starts.
    role: Arc<Mutex<T>>,
}

impl<T> SupervisorRoleAdapter<T>
where
    T: SupervisorRole,
{
    /// Creates a supervisor role adapter.
    ///
    /// # Arguments
    ///
    /// - `role`: Supervisor role instance that owns user lifecycle state.
    ///
    /// # Returns
    ///
    /// Returns an adapter that implements the task factory service contract.
    ///
    /// # Examples
    ///
    /// ```
    /// use rust_supervisor::role::adapter::supervisor::SupervisorRoleAdapter;
    /// use rust_supervisor::role::context::supervisor::SupervisorContext;
    /// use rust_supervisor::role::result::supervisor::SupervisorResult;
    /// use rust_supervisor::role::traits::supervisor::SupervisorRole;
    /// use rust_supervisor::spec::supervisor::SupervisorSpec;
    ///
    /// struct ExampleSupervisor;
    ///
    /// impl SupervisorRole for ExampleSupervisor {
    ///     async fn build_tree(
    ///         &mut self,
    ///         ctx: &SupervisorContext,
    ///     ) -> SupervisorResult<SupervisorSpec> {
    ///         let _ = ctx;
    ///         Ok(SupervisorSpec::root(Vec::new()))
    ///     }
    /// }
    ///
    /// let _adapter = SupervisorRoleAdapter::new(ExampleSupervisor);
    /// ```
    pub fn new(role: T) -> Self {
        Self {
            role: Arc::new(Mutex::new(role)),
        }
    }
}

impl<T> Service for SupervisorRoleAdapter<T>
where
    T: SupervisorRole,
{
    /// Builds one task future for the adapted supervisor role.
    fn call(&self, ctx: TaskContext) -> BoxTaskFuture {
        let role = self.role.clone();
        let supervisor_context = SupervisorContext::new(ctx);
        Box::pin(async move {
            let mut role = role.lock().await;
            run_supervisor_lifecycle(&mut *role, &supervisor_context).await
        })
    }
}

/// Runs the supervisor lifecycle and maps role errors into task results.
///
/// # Arguments
///
/// - `role`: Supervisor role implementation.
/// - `ctx`: Supervisor-specific context passed to lifecycle methods.
///
/// # Returns
///
/// Returns a [`TaskResult`] consumed by the runtime.
async fn run_supervisor_lifecycle<T>(role: &mut T, ctx: &SupervisorContext) -> TaskResult
where
    T: SupervisorRole,
{
    let spec = match role.build_tree(ctx).await {
        Ok(spec) => spec,
        Err(error) => return supervisor_error_to_task_result(error),
    };
    let handle = match Supervisor::start(spec).await {
        Ok(handle) => handle,
        Err(error) => {
            return supervisor_error_to_task_result(SupervisorRoleError::from_supervisor_error(
                ctx.child_id().clone(),
                RoleLifecyclePhase::BuildTree,
                error,
            ));
        }
    };

    let run_result = role.run(ctx, &handle).await;
    let shutdown_result = role.shutdown(ctx, &handle).await;
    if let Err(error) = run_result {
        return supervisor_error_to_task_result(error);
    }
    if let Err(error) = shutdown_result {
        return supervisor_error_to_task_result(error);
    }
    if ctx.is_shutdown_requested() {
        TaskResult::Cancelled
    } else {
        TaskResult::Succeeded
    }
}

/// Converts a supervisor role error into a failed task result.
///
/// # Arguments
///
/// - `error`: Supervisor role error returned by a lifecycle method.
///
/// # Returns
///
/// Returns [`TaskResult::Failed`] with a typed task failure payload.
fn supervisor_error_to_task_result(error: SupervisorRoleError) -> TaskResult {
    TaskResult::Failed(error.into_task_failure())
}