Implement ComponentManager for Hierarchical Component Organization

[thep2p]

Overview

The Go implementation has a ComponentManager that can manage multiple components and itself acts as a component. This enables building hierarchical, tree-like structures where managers can contain other managers, providing powerful composition capabilities.

Background

Reference implementation: skipgraph-go/modules/component/manager.go

ComponentManager provides:

• Hierarchical component organization
• Coordinated startup/shutdown
• Aggregate ready/done signaling
• Recursive management structures

Requirements

1. Define ComponentManager Trait

use async_trait::async_trait;
use std::sync::Arc;

/// A component that can manage other components
#[async_trait]
pub trait ComponentManager: Component {
    /// Add a component to be managed
    /// 
    /// # Panics
    /// - If called after the manager has been started
    /// - If the same component is added multiple times
    fn add(&mut self, component: Arc<dyn Component>);
}

2. Implement Manager Struct

use std::sync::{Arc, Mutex};
use tokio::sync::watch;

pub struct Manager {
    components: Arc<Mutex<Vec<Arc<dyn Component>>>>,
    started: Arc<Mutex<bool>>,
    lifecycle: LifecycleState,
    start_once: StartOnce,
}

impl Manager {
    pub fn new() -> Self {
        Self {
            components: Arc::new(Mutex::new(Vec::new())),
            started: Arc::new(Mutex::new(false)),
            lifecycle: LifecycleState::new(),
            start_once: StartOnce::new(),
        }
    }
    
    /// Add a component to this manager
    pub fn add(&mut self, component: Arc<dyn Component>) {
        let mut started = self.started.lock().unwrap();
        if *started {
            panic!("Cannot add component after manager has started");
        }
        
        let mut components = self.components.lock().unwrap();
        
        // Check for duplicates
        for existing in components.iter() {
            if Arc::ptr_eq(existing, &component) {
                panic!("Cannot add the same component multiple times");
            }
        }
        
        components.push(component);
    }
}

#[async_trait]
impl Startable for Manager {
    async fn start(&self, ctx: Arc<dyn ThrowableContext>) {
        if let Err(e) = self.start_once.ensure_once() {
            panic!("{}", e);
        }
        
        {
            let mut started = self.started.lock().unwrap();
            *started = true;
        }
        
        let components = self.components.clone();
        let lifecycle = self.lifecycle.clone();
        
        // Start all components
        let components_list = components.lock().unwrap().clone();
        for component in &components_list {
            component.start(ctx.clone()).await;
        }
        
        // Spawn ready waiter
        let components_clone = components_list.clone();
        let lifecycle_clone = lifecycle.clone();
        tokio::spawn(async move {
            wait_all_ready(components_clone).await;
            lifecycle_clone.signal_ready();
        });
        
        // Spawn done waiter
        let components_clone = components_list.clone();
        let lifecycle_clone = lifecycle.clone();
        tokio::spawn(async move {
            wait_all_done(components_clone).await;
            lifecycle_clone.signal_done();
        });
    }
}

impl ReadyDoneAware for Manager {
    fn ready(&self) -> watch::Receiver<bool> {
        self.lifecycle.ready()
    }
    
    fn done(&self) -> watch::Receiver<bool> {
        self.lifecycle.done()
    }
}

// Helper functions
async fn wait_all_ready(components: Vec<Arc<dyn Component>>) {
    for component in components {
        let mut ready = component.ready();
        while !*ready.borrow() {
            ready.changed().await.ok();
        }
    }
}

async fn wait_all_done(components: Vec<Arc<dyn Component>>) {
    for component in components {
        let mut done = component.done();
        while !*done.borrow() {
            done.changed().await.ok();
        }
    }
}

3. Hierarchical Usage Example

async fn build_application() -> Arc<Manager> {
    let mut root = Manager::new();
    
    // Create sub-managers
    let mut network_manager = Manager::new();
    let mut storage_manager = Manager::new();
    
    // Add components to sub-managers
    network_manager.add(Arc::new(TcpServer::new()));
    network_manager.add(Arc::new(GrpcServer::new()));
    
    storage_manager.add(Arc::new(Database::new()));
    storage_manager.add(Arc::new(Cache::new()));
    
    // Create recursive structure
    let mut monitoring_manager = Manager::new();
    monitoring_manager.add(Arc::new(MetricsCollector::new()));
    monitoring_manager.add(Arc::new(HealthChecker::new()));
    
    // Add sub-managers to root
    root.add(Arc::new(network_manager));
    root.add(Arc::new(storage_manager));
    root.add(Arc::new(monitoring_manager));
    
    Arc::new(root)
}

async fn start_application() {
    let ctx = Arc::new(Context::new());
    let app = build_application().await;
    
    // Start entire tree
    app.start(ctx.clone()).await;
    
    // Wait for all components to be ready
    let mut ready = app.ready();
    while !*ready.borrow() {
        ready.changed().await.ok();
    }
    
    println!("Application fully initialized");
    
    // On shutdown...
    ctx.cancel();
    
    let mut done = app.done();
    while !*done.borrow() {
        done.changed().await.ok();
    }
    
    println!("Application cleanly shut down");
}

4. Tree Structure Visualization

        Root Manager
             │
      ┌──────┼──────┐
      │      │      │
  Network Storage Monitor
  Manager Manager Manager
     │       │       │
  ┌──┴──┐ ┌──┴──┐ ┌──┴──┐
 TCP gRPC DB Cache Metrics Health
 
Ready Flow: Leaves → Sub-managers → Root
Done Flow: Leaves → Sub-managers → Root

Key Features

• Recursive Structure: Managers can contain other managers
• Lifecycle Coordination: Parent waits for all children
• Thread Safety: Safe concurrent access
• Empty Manager Support: Managers with no components signal ready/done immediately

Testing Requirements

• Test basic manager operations
• Test hierarchical structures (3+ levels deep)
• Test empty managers
• Test concurrent component additions (should panic)
• Test ready/done propagation
• Test with blocking components
• Test error scenarios

Dependencies

• async-trait
• tokio
• Depends on: Implement ThrowableContext for Error Propagation #52, Implement ReadyDoneAware Trait for Lifecycle State Management #53, Implement Startable Trait for Component Initialization #54, Implement Component Trait Combining Startable and ReadyDoneAware #55

Priority

High - Essential for building complex applications

Related Issues

• Depends on: Component trait (Implement Component Trait Combining Startable and ReadyDoneAware #55)
• Enables: Engine abstraction