apalis_core/task/
status.rs

1//! The status of a task
2//!
3//! ## Overview
4//!
5//! The `Status` enum defines the various states
6//! a task can be in, such as `Pending`, `Running`, `Done`, `Failed`, etc.
7//!
8//! - It includes functionality for parsing a `Status` from a string and formatting it for display.
9//! - This is useful for tracking the lifecycle of tasks.
10use core::fmt;
11use std::{
12    str::FromStr,
13    sync::{
14        Arc,
15        atomic::{AtomicU8, Ordering},
16    },
17};
18
19/// Represents the state of a task
20#[repr(u8)]
21#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
22#[non_exhaustive]
23#[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
24pub enum Status {
25    /// Task is pending
26    Pending,
27    /// Task is queued for execution, but no worker has picked it up
28    Queued,
29    /// Task is running
30    Running,
31    /// Task was done successfully
32    Done,
33    /// Task has failed.
34    Failed,
35    /// Task has been killed
36    Killed,
37}
38
39impl Default for Status {
40    fn default() -> Self {
41        Self::Pending
42    }
43}
44
45/// Errors that can occur when parsing a `Status` from a string
46#[derive(Debug, thiserror::Error)]
47pub enum StatusError {
48    #[error("Unknown state: {0}")]
49    /// Unknown state error
50    UnknownState(String),
51}
52
53impl FromStr for Status {
54    type Err = StatusError;
55
56    fn from_str(s: &str) -> Result<Self, Self::Err> {
57        match s {
58            "Pending" => Ok(Self::Pending),
59            "Queued" => Ok(Self::Queued),
60            "Running" => Ok(Self::Running),
61            "Done" => Ok(Self::Done),
62            "Failed" => Ok(Self::Failed),
63            "Killed" => Ok(Self::Killed),
64            _ => Err(StatusError::UnknownState(s.to_owned())),
65        }
66    }
67}
68
69impl fmt::Display for Status {
70    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
71        match &self {
72            Self::Pending => write!(f, "Pending"),
73            Self::Queued => write!(f, "Queued"),
74            Self::Running => write!(f, "Running"),
75            Self::Done => write!(f, "Done"),
76            Self::Failed => write!(f, "Failed"),
77            Self::Killed => write!(f, "Killed"),
78        }
79    }
80}
81
82impl Status {
83    fn from_u8(val: u8) -> Option<Self> {
84        match val {
85            0 => Some(Self::Pending),
86            1 => Some(Self::Queued),
87            2 => Some(Self::Running),
88            3 => Some(Self::Done),
89            4 => Some(Self::Failed),
90            5 => Some(Self::Killed),
91            _ => None,
92        }
93    }
94}
95
96/// Atomic version of `Status` for concurrent scenarios
97#[repr(transparent)]
98#[derive(Debug, Clone, Default)]
99pub struct AtomicStatus(Arc<AtomicU8>);
100
101impl AtomicStatus {
102    /// Create a new `AtomicStatus` with the given initial status
103    #[must_use]
104    pub fn new(status: Status) -> Self {
105        Self(Arc::new(AtomicU8::new(status as u8)))
106    }
107
108    /// Load the current status
109    #[must_use]
110    pub fn load(&self) -> Status {
111        Status::from_u8(self.0.load(Ordering::Acquire)).unwrap()
112    }
113
114    /// Store a new status
115    pub fn store(&self, status: Status) {
116        self.0.store(status as u8, Ordering::Release);
117    }
118    /// Swap the current status with a new one, returning the old status
119    #[must_use]
120    pub fn swap(&self, status: Status) -> Status {
121        Status::from_u8(self.0.swap(status as u8, Ordering::AcqRel)).unwrap()
122    }
123}
124
125impl From<AtomicStatus> for Status {
126    fn from(val: AtomicStatus) -> Self {
127        val.load()
128    }
129}
130
131impl From<Status> for AtomicStatus {
132    fn from(val: Status) -> Self {
133        Self::new(val)
134    }
135}
136
137#[cfg(feature = "serde")]
138mod serde_impl {
139    use serde::{Deserialize, Deserializer, Serialize, Serializer};
140
141    use super::*;
142
143    // Custom serialization function
144    fn serialize<S>(status: &AtomicStatus, serializer: S) -> Result<S::Ok, S::Error>
145    where
146        S: Serializer,
147    {
148        let value = status.load();
149        serializer.serialize_str(&value.to_string())
150    }
151
152    // Custom deserialization function
153    fn deserialize<'de, D>(deserializer: D) -> Result<AtomicStatus, D::Error>
154    where
155        D: Deserializer<'de>,
156    {
157        let value = String::deserialize(deserializer)?;
158        let status = Status::from_str(&value).map_err(serde::de::Error::custom)?;
159        Ok(AtomicStatus::new(status))
160    }
161
162    impl Serialize for AtomicStatus {
163        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
164        where
165            S: Serializer,
166        {
167            serialize(self, serializer)
168        }
169    }
170
171    impl<'de> Deserialize<'de> for AtomicStatus {
172        fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
173        where
174            D: Deserializer<'de>,
175        {
176            deserialize(deserializer)
177        }
178    }
179}
apalis_core/task/status.rs

apalis_core/task/
status.rs
