boson_runtime/worker/
manual.rs

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

boson_runtime/worker/
manual.rs
