foxtive_supervisor/
hierarchy.rs1use crate::Supervisor;
7use crate::runtime::TaskRuntime;
8use std::sync::Arc;
9use tokio::sync::Mutex;
10use tracing::info;
11
12pub struct SupervisorHierarchy {
14 root: HierarchyBuilder,
15}
16
17struct HierarchyBuilder {
19 supervisor: Option<Supervisor>,
20 children: Vec<HierarchyBuilder>,
21 id: String,
22}
23
24pub struct HierarchyRuntime {
26 root: Arc<RuntimeNode>,
27}
28
29pub struct RuntimeNode {
31 runtime: Mutex<Option<TaskRuntime>>,
32 children: Vec<Arc<RuntimeNode>>,
33 id: String,
34}
35
36impl HierarchyBuilder {
37 fn new(id: impl Into<String>) -> Self {
38 Self {
39 supervisor: None,
40 children: Vec::new(),
41 id: id.into(),
42 }
43 }
44
45 fn with_supervisor(mut self, supervisor: Supervisor) -> Self {
46 self.supervisor = Some(supervisor);
47 self
48 }
49
50 fn add_child(mut self, child: HierarchyBuilder) -> Self {
51 self.children.push(child);
52 self
53 }
54}
55
56impl SupervisorHierarchy {
57 pub fn new(root_id: impl Into<String>) -> Self {
59 Self {
60 root: HierarchyBuilder::new(root_id),
61 }
62 }
63
64 pub fn add_child(mut self, id: impl Into<String>, supervisor: Supervisor) -> Self {
66 let child = HierarchyBuilder::new(id).with_supervisor(supervisor);
67 self.root = self.root.add_child(child);
68 self
69 }
70
71 pub async fn start_all(self) -> Result<HierarchyRuntime, crate::error::SupervisorError> {
76 info!("Starting hierarchy from root '{}'", self.root.id);
77 let root = Box::pin(Self::start_builder_node(self.root)).await?;
78 Ok(HierarchyRuntime { root })
79 }
80
81 async fn start_builder_node(
83 builder: HierarchyBuilder,
84 ) -> Result<Arc<RuntimeNode>, crate::error::SupervisorError> {
85 let mut children = Vec::new();
87 for child_builder in builder.children {
88 let child_runtime = Box::pin(Self::start_builder_node(child_builder)).await?;
89 children.push(child_runtime);
90 }
91
92 let runtime = if let Some(supervisor) = builder.supervisor {
94 info!(node_id = %builder.id, "Starting supervisor");
95 let rt = supervisor.start().await?;
96 info!(node_id = %builder.id, "Supervisor started with {} tasks", rt.task_count());
97 Some(rt)
98 } else {
99 None
100 };
101
102 Ok(Arc::new(RuntimeNode {
103 runtime: Mutex::new(runtime),
104 children,
105 id: builder.id,
106 }))
107 }
108}
109
110impl RuntimeNode {
111 async fn shutdown_node(self: Arc<Self>) {
113 info!(node_id = %self.id, "Initiating shutdown");
114
115 let shutdown_futures: Vec<_> = self
117 .children
118 .iter()
119 .map(|child| {
120 let child_clone = child.clone();
121 async move {
122 Box::pin(child_clone.shutdown_node()).await;
123 }
124 })
125 .collect();
126
127 futures::future::join_all(shutdown_futures).await;
128
129 let mut guard = self.runtime.lock().await;
131 if let Some(runtime) = guard.take() {
132 info!(node_id = %self.id, "Shutting down supervisor");
133 runtime.shutdown().await;
134 info!(node_id = %self.id, "Supervisor shut down");
135 }
136 }
137
138 fn task_count(&self) -> usize {
140 let local_count = match self.runtime.try_lock() {
141 Ok(guard) => guard.as_ref().map_or(0, |rt| rt.task_count()),
142 Err(_) => 0, };
144
145 let child_count: usize = self.children.iter().map(|child| child.task_count()).sum();
146
147 local_count + child_count
148 }
149
150 #[allow(dead_code)]
152 fn find_node(&self, target_id: &str) -> Option<&Self> {
153 if self.id == target_id {
154 return Some(self);
155 }
156
157 for child in &self.children {
158 if let Some(found) = child.find_node(target_id) {
159 return Some(found);
160 }
161 }
162
163 None
164 }
165}
166
167impl HierarchyRuntime {
168 pub async fn shutdown_all(self) {
170 info!("Initiating hierarchy shutdown from root '{}'", self.root.id);
171 Box::pin(self.root.shutdown_node()).await;
172 info!("Hierarchy shutdown complete");
173 }
174
175 pub fn total_task_count(&self) -> usize {
177 self.root.task_count()
178 }
179}
180
181#[cfg(test)]
182mod tests {
183 use super::*;
184
185 #[tokio::test]
186 async fn test_create_hierarchy() {
187 let hierarchy = SupervisorHierarchy::new("root");
188 let _ = hierarchy;
190 }
191}