noema-actix-webapi 0.1.0

use std::sync::Arc;

/// Process-lifetime loop (consumers, pollers). Not an event [`noema::events::BackgroundSpawner`].
///
/// Pass instances to [`spawn_background_tasks`] after [`crate::start`].
#[async_trait::async_trait]
pub trait BackgroundTask: Send + Sync {
    fn name(&self) -> &'static str;

    async fn init(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        Ok(())
    }

    async fn run(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
}

pub type DynBackgroundTask = dyn BackgroundTask + Send + Sync;

/// Box a concrete task for [`spawn_background_tasks`].
pub fn background_task(task: impl BackgroundTask + 'static) -> Arc<DynBackgroundTask> {
    Arc::new(task)
}

/// Spawn each task on the Actix/tokio runtime. Does not steal `HttpServer`.
pub fn spawn_background_tasks(tasks: impl IntoIterator<Item = Arc<DynBackgroundTask>>) {
    for task in tasks {
        let _ = actix_web::rt::spawn(async move {
            if let Err(err) = task.init().await {
                tracing::error!(name = task.name(), error = %err, "background task init failed");
                return;
            }
            if let Err(err) = task.run().await {
                tracing::error!(name = task.name(), error = %err, "background task failed");
            }
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;

    struct FlagTask {
        flag: Arc<AtomicBool>,
    }

    #[async_trait::async_trait]
    impl BackgroundTask for FlagTask {
        fn name(&self) -> &'static str {
            "flag"
        }

        async fn run(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
            self.flag.store(true, Ordering::SeqCst);
            Ok(())
        }
    }

    #[actix_web::test]
    async fn run_is_callable() {
        let task = FlagTask {
            flag: Arc::new(AtomicBool::new(false)),
        };
        task.run().await.unwrap();
        assert!(task.flag.load(Ordering::SeqCst));
    }

    #[actix_web::test]
    async fn spawn_skips_run_when_init_fails() {
        struct Boom;
        #[async_trait::async_trait]
        impl BackgroundTask for Boom {
            fn name(&self) -> &'static str {
                "boom"
            }
            async fn init(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                Err("nope".into())
            }
            async fn run(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                panic!("run must not run");
            }
        }
        spawn_background_tasks([background_task(Boom)]);
        actix_web::rt::time::sleep(Duration::from_millis(50)).await;
    }

    struct Ordered {
        init: Arc<AtomicBool>,
        run: Arc<AtomicBool>,
    }

    #[async_trait::async_trait]
    impl BackgroundTask for Ordered {
        fn name(&self) -> &'static str {
            "ordered"
        }
        async fn init(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
            self.init.store(true, Ordering::SeqCst);
            Ok(())
        }
        async fn run(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
            assert!(self.init.load(Ordering::SeqCst));
            self.run.store(true, Ordering::SeqCst);
            Ok(())
        }
    }

    #[actix_web::test]
    async fn spawn_inits_then_runs() {
        let init = Arc::new(AtomicBool::new(false));
        let run = Arc::new(AtomicBool::new(false));
        spawn_background_tasks([background_task(Ordered {
            init: Arc::clone(&init),
            run: Arc::clone(&run),
        })]);
        actix_web::rt::time::sleep(Duration::from_millis(50)).await;
        assert!(init.load(Ordering::SeqCst));
        assert!(run.load(Ordering::SeqCst));
    }
}