harmony/harmony_composer/src/instrumentation.rs

use log::debug;
use once_cell::sync::Lazy;
use tokio::sync::broadcast;

use crate::{HarmonyProfile, HarmonyTarget};

#[derive(Debug, Clone)]
pub enum HarmonyComposerEvent {
    HarmonyComposerStarted,
    ProjectInitializationStarted,
    ProjectInitialized,
    ProjectCompilationStarted {
        details: String,
    },
    ProjectCompiled,
    ProjectCompilationFailed {
        details: String,
    },
    DeploymentStarted {
        target: HarmonyTarget,
        profile: HarmonyProfile,
    },
    DeploymentCompleted,
    DeploymentFailed {
        details: String,
    },
    Shutdown,
}

static HARMONY_COMPOSER_EVENT_BUS: Lazy<broadcast::Sender<HarmonyComposerEvent>> =
    Lazy::new(|| {
        // TODO: Adjust channel capacity
        let (tx, _rx) = broadcast::channel(16);
        tx
    });

pub fn instrument(event: HarmonyComposerEvent) -> Result<(), &'static str> {
    #[cfg(not(test))]
    {
        match HARMONY_COMPOSER_EVENT_BUS.send(event) {
            Ok(_) => Ok(()),
            Err(_) => Err("send error: no subscribers"),
        }
    }

    #[cfg(test)]
    {
        let _ = event; // Suppress the "unused variable" warning for `event`
        Ok(())
    }
}

pub async fn subscribe<F, Fut>(name: &str, mut handler: F)
where
    F: FnMut(HarmonyComposerEvent) -> Fut + Send + 'static,
    Fut: Future<Output = bool> + Send,
{
    let mut rx = HARMONY_COMPOSER_EVENT_BUS.subscribe();
    debug!("[{name}] Service started. Listening for events...");
    loop {
        match rx.recv().await {
            Ok(event) => {
                if !handler(event).await {
                    debug!("[{name}] Handler requested exit.");
                    break;
                }
            }
            Err(broadcast::error::RecvError::Lagged(n)) => {
                debug!("[{name}] Lagged behind by {n} messages.");
            }
            Err(_) => break,
        }
    }
}