apalis_core/monitor/
context.rs

1//! Monitor context and worker control.
2//!
3//! This module provides [`MonitorContext`], a shared handle for inspecting
4//! and controlling the workers supervised by a monitor.
5//!
6//! A monitor context provides access to the workers managed by the monitor and
7//! exposes operations for controlling their execution, including pausing,
8//! resuming, stopping, waking, and cancelling tasks.
9//!
10//! It also provides a non-blocking [`MonitorContext::shutdown`] operation for
11//! initiating shutdown of the monitor.
12//!
13//! # Worker control
14//!
15//! Workers are identified by name and can be controlled through the following
16//! methods:
17//!
18//! * [`MonitorContext::pause_worker`] pauses a worker.
19//! * [`MonitorContext::resume_worker`] resumes a paused worker.
20//! * [`MonitorContext::stop_worker`] stops a worker.
21//! * [`MonitorContext::wake_worker`] wakes a worker so it can continue polling.
22//!
23//! # Task cancellation
24//!
25//! [`MonitorContext::cancel_task`] cancels a task currently registered with a
26//! worker. The task is identified by both its worker name and task ID.
27//!
28//! # Shutdown
29//!
30//! [`MonitorContext::shutdown`] initiates monitor shutdown by signalling the
31//! monitor's shutdown handle. It does not wait for the supervised workers to
32//! finish stopping.
33//!
34//! # Errors
35//!
36//! Worker operations return [`WorkerError`] when the requested worker cannot
37//! be found or when the requested operation fails. Task cancellation may
38//! return a boxed error containing the underlying worker or task error.
39use crate::{
40    error::{WorkerError, WorkerStateError},
41    monitor::{MonitorError, shutdown::Shutdown},
42    worker::context::WorkerContext,
43};
44
45/// A handle to the monitor which allows probing
46/// Shared context for the monitor, providing access to the shutdown
47/// handle and the set of workers it supervises.
48#[derive(Debug, Clone)]
49pub struct MonitorContext {
50    /// Handle used to signal and coordinate shutdown of the monitor.
51    pub(super) shutdown: Shutdown,
52    /// The workers currently managed by this monitor.
53    pub(super) workers: Vec<WorkerContext>,
54}
55
56impl MonitorContext {
57    /// Returns a reference to all workers managed by this monitor.
58    #[must_use]
59    pub fn workers(&self) -> &Vec<WorkerContext> {
60        &self.workers
61    }
62
63    /// Initiates a shutdown of the monitor.
64    ///
65    /// This signals the shutdown handle but does not block waiting
66    /// for workers to finish.
67    pub fn shutdown(&self) -> Result<(), MonitorError> {
68        self.shutdown.start_shutdown();
69        // Setting the flag does not re-poll a parked worker; wake each one so it observes it.
70        for worker in &self.workers {
71            worker.wake();
72        }
73        Ok(())
74    }
75
76    /// Resumes the worker with the given `name`.
77    ///
78    /// # Errors
79    /// Returns [`WorkerError`] if no worker with `name` exists, or if
80    /// the resume itself fails.
81    pub fn resume_worker(&self, name: &str) -> Result<(), WorkerError> {
82        self.get_worker(name)?.resume()
83    }
84
85    /// Pauses the worker with the given `name`.
86    ///
87    /// # Errors
88    /// Returns [`WorkerError`] if no worker with `name` exists, or if
89    /// pausing fails.
90    pub fn pause_worker(&self, name: &str) -> Result<(), WorkerError> {
91        self.get_worker(name)?.pause()
92    }
93
94    /// Stops the worker with the given `name`.
95    ///
96    /// # Errors
97    /// Returns [`WorkerError`] if no worker with `name` exists, or if
98    /// stopping fails.
99    pub fn stop_worker(&self, name: &str) -> Result<(), WorkerError> {
100        self.get_worker(name)?.stop()
101    }
102
103    /// Wakes the worker with the given `name`.
104    ///
105    /// # Errors
106    /// Returns [`WorkerError`] if no worker with `name` exists.
107    pub fn wake_worker(&self, name: &str) -> Result<(), WorkerError> {
108        self.get_worker(name)?.wake();
109        Ok(())
110    }
111
112    /// Looks up a worker by name.
113    ///
114    /// # Errors
115    /// Returns [`WorkerError::StateError`] if no worker with `name`
116    /// exists among the managed workers.
117    pub fn get_worker(&self, name: &str) -> Result<&WorkerContext, WorkerError> {
118        self.workers
119            .iter()
120            .find(|w| w.name() == name)
121            .ok_or_else(|| {
122                WorkerError::StateError(WorkerStateError::InvalidState("NOT_FOUND".to_owned()))
123            })
124    }
125
126    /// Attempt to cancel a task
127    #[cfg(feature = "task-context")]
128    pub fn cancel_task(
129        &self,
130        worker: &str,
131        task_id: &str,
132    ) -> Result<(), crate::error::BoxDynError> {
133        let worker = self.get_worker(worker)?;
134        let context = worker.get_task(task_id)?;
135        worker.cancel_task(&context)?;
136        Ok(())
137    }
138}
apalis_core/monitor/context.rs

apalis_core/monitor/
context.rs
