boson_runtime/worker/
manual.rs

1//! Manual single-step worker for tests (no background task).
2
3use std::sync::Arc;
4
5use boson_core::ExecutionContextFactory;
6use boson_core::QueueBackend;
7use tokio::sync::Mutex;
8
9use super::claim::claim_next_job;
10use super::config::WorkerSettings;
11use super::loop_::WorkerEngine;
12use crate::registry::TaskRegistry;
13
14/// Manual single-step worker for tests (no background task).
15///
16/// Use [`BosonBuilder::build_manual`](crate::BosonBuilder::build_manual) to obtain one alongside
17/// [`Boson`](crate::Boson). Call [`try_run_next`](Self::try_run_next) to claim and execute at most
18/// one queued job — useful in unit tests and the [`task_macro` example](https://github.com/unified-field-dev/boson/blob/main/boson/examples/task_macro.rs).
19///
20/// # Example
21///
22/// ```rust,no_run
23/// use std::sync::Arc;
24///
25/// use boson_backend_mem::MemQueueBackend;
26/// use boson_core::{ExecutionContext, JsonExecutionContextFactory};
27/// use boson_macros::task;
28/// use boson_runtime::{configure, Boson, ManualWorker};
29///
30/// #[task(name = "ping")]
31/// async fn ping(_ctx: Box<dyn ExecutionContext>) -> boson_core::Result<()> {
32///     Ok(())
33/// }
34///
35/// # async fn run() -> boson_core::Result<()> {
36/// let (boson, manual) = Boson::builder()
37///     .queue_backend(Arc::new(MemQueueBackend::new()))
38///     .execution_context_factory(JsonExecutionContextFactory)
39///     .auto_registry()
40///     .build_manual()?;
41/// configure(boson);
42///
43/// Ping::send_with(serde_json::json!({"System": {}}), PingParams {}).await?;
44/// assert!(manual.try_run_next().await); // runs the handler once
45/// # Ok(())
46/// # }
47/// ```
48pub struct ManualWorker {
49    inner: Arc<WorkerEngine>,
50    lock: Mutex<()>,
51}
52
53impl ManualWorker {
54    /// Create a worker that can be driven step-by-step in tests.
55    pub fn new(
56        backend: Arc<dyn QueueBackend>,
57        registry: Arc<TaskRegistry>,
58        identity: Arc<dyn ExecutionContextFactory>,
59        worker: WorkerSettings,
60    ) -> Self {
61        Self {
62            inner: Arc::new(WorkerEngine {
63                backend,
64                registry,
65                identity,
66                worker,
67            }),
68            lock: Mutex::new(()),
69        }
70    }
71
72    /// Process at most one job across all pools.
73    pub async fn try_run_next(&self) -> bool {
74        let _guard = self.lock.lock().await;
75        let discovered = self
76            .inner
77            .backend
78            .distinct_pools_queued()
79            .await
80            .unwrap_or_default();
81        let pools = self.inner.worker.pools_to_poll(discovered);
82        for pool in pools {
83            if let Ok(Some((job, lease_id))) = claim_next_job(
84                &self.inner.backend,
85                &pool,
86                &self.inner.worker.worker_id,
87                self.inner.worker.lease_ttl_secs,
88            )
89            .await
90            {
91                self.inner.drive_run(job, lease_id).await;
92                return true;
93            }
94        }
95        false
96    }
97}
boson_runtime/worker/manual.rs

boson_runtime/worker/
manual.rs
