capp_queue/
task.rs

1use serde::{Deserialize, Serialize};
2use std::{str::FromStr, time::SystemTime};
3use uuid::Uuid;
4
5#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
6pub enum TaskStatus {
7    Queued,
8    InProgress,
9    Completed,
10    Failed,
11    DeadLetter,
12}
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
15pub struct TaskId(Uuid);
16
17/// A `Task` struct represents a single unit of work that will be processed
18/// by a worker. It contains payload of type `D`, which is used by the worker
19/// during processing. The `Task` struct also includes fields for managing
20/// the task's lifecycle, including the task's UUID, the start and
21/// finish times, the number of retries, and any error messages.
22#[derive(Clone, Debug, Serialize, Deserialize)]
23pub struct Task<D: Clone> {
24    pub task_id: TaskId,
25    pub payload: D,
26    pub status: TaskStatus,
27    pub queued_at: SystemTime,
28    pub started_at: Option<SystemTime>,
29    pub finished_at: Option<SystemTime>,
30    pub retries: u32,
31    pub error_msg: Option<String>,
32}
33
34impl<D: Clone> Task<D> {
35    pub fn new(payload: D) -> Self {
36        Task {
37            task_id: TaskId::new(),
38            payload,
39            status: TaskStatus::Queued,
40            queued_at: SystemTime::now(),
41            started_at: None,
42            finished_at: None,
43            retries: 0,
44            error_msg: None,
45        }
46    }
47
48    pub fn set_in_progress(&mut self) {
49        self.status = TaskStatus::InProgress;
50        self.started_at = Some(SystemTime::now());
51    }
52
53    pub fn set_succeed(&mut self) {
54        self.status = TaskStatus::Completed;
55        self.finished_at = Some(SystemTime::now());
56    }
57
58    pub fn set_retry(&mut self, err_msg: &str) {
59        self.status = TaskStatus::Failed;
60        self.finished_at = Some(SystemTime::now());
61        self.retries += 1;
62        self.error_msg = Some(err_msg.to_string());
63    }
64
65    pub fn set_dlq(&mut self, err_msg: &str) {
66        self.status = TaskStatus::DeadLetter;
67        self.finished_at = Some(SystemTime::now());
68        self.error_msg = Some(err_msg.to_string());
69    }
70
71    pub fn set_status(&mut self, new_status: TaskStatus) {
72        self.status = new_status;
73    }
74
75    pub fn get_payload(&self) -> &D {
76        &self.payload
77    }
78}
79
80impl TaskId {
81    pub fn new() -> Self {
82        Self(Uuid::now_v7())
83    }
84
85    pub fn get(&self) -> Uuid {
86        self.0
87    }
88
89    pub fn as_bytes(&self) -> &[u8; 16] {
90        self.0.as_bytes()
91    }
92
93    pub fn from_uuid(uuid: Uuid) -> Self {
94        Self(uuid)
95    }
96
97    pub fn from_string(uuid_str: &str) -> Result<Self, uuid::Error> {
98        let uuid = Uuid::from_str(uuid_str)?;
99        Ok(Self(uuid))
100    }
101}
102
103impl FromStr for TaskId {
104    type Err = uuid::Error;
105    fn from_str(s: &str) -> Result<Self, Self::Err> {
106        Self::from_string(s)
107    }
108}
109
110impl Default for TaskId {
111    fn default() -> Self {
112        Self::new()
113    }
114}
115
116// Custom serialization for TaskId.
117impl serde::Serialize for TaskId {
118    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
119    where
120        S: serde::Serializer,
121    {
122        // Directly serialize the inner Uuid.
123        self.0.serialize(serializer)
124    }
125}
126
127// Custom deserialization for TaskId.
128impl<'de> serde::Deserialize<'de> for TaskId {
129    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
130    where
131        D: serde::Deserializer<'de>,
132    {
133        // Deserialize a Uuid and then wrap it in a TaskId.
134        let uuid = Uuid::deserialize(deserializer)?;
135        Ok(TaskId(uuid))
136    }
137}
138
139impl std::fmt::Display for TaskId {
140    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
141        write!(f, "TaskId({})", self.0)
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    use core::panic;
148
149    use super::*;
150    use serde::{Deserialize, Serialize};
151
152    #[derive(Clone, Serialize, Deserialize, Default)]
153    struct TaskData {
154        value: u32,
155    }
156
157    #[test]
158    fn task_id_from_string_roundtrip() {
159        let original = TaskId::new();
160        let parsed = TaskId::from_string(&original.get().to_string()).unwrap();
161        assert_eq!(original, parsed);
162        let parsed_via_fromstr: TaskId =
163            original.get().to_string().parse().unwrap();
164        assert_eq!(original, parsed_via_fromstr);
165    }
166
167    #[test]
168    fn task_id_from_string_rejects_invalid() {
169        assert!(TaskId::from_string("not-a-uuid").is_err());
170        assert!("not-a-uuid".parse::<TaskId>().is_err());
171    }
172
173    #[test]
174    fn task_uuid_conversion() {
175        let task = Task::new(TaskData { value: 42 });
176        let uuid_string = task.task_id.get().to_string();
177        assert!(!uuid_string.contains("TaskId"));
178    }
179
180    #[test]
181    fn task_id_serde() {
182        let task = Task::new(TaskData { value: 1 });
183        let task_id = task.task_id;
184        let serialized_task_value = serde_json::to_value(task).unwrap();
185        let serialized_task_json = serialized_task_value.to_string();
186        let desrialized_task: Task<TaskData> =
187            serde_json::from_str(&serialized_task_json).unwrap();
188        assert_eq!(task_id, desrialized_task.task_id);
189    }
190
191    #[test]
192    fn test_task_creation() {
193        let task = Task::new(TaskData::default());
194        assert!(task.started_at.is_none());
195        assert!(task.finished_at.is_none());
196        assert_eq!(task.retries, 0);
197        assert_eq!(task.payload.value, 0);
198        match task.status {
199            TaskStatus::Queued => {}
200            _ => panic!("Wrong status (task.status)"),
201        };
202    }
203
204    #[test]
205    fn test_in_progress() {
206        let mut task = Task::new(TaskData::default());
207
208        task.set_in_progress();
209        match task.status {
210            TaskStatus::InProgress => {}
211            _ => panic!("Wrong status (task.status)"),
212        };
213        assert!(task.started_at.is_some());
214        assert!(task.finished_at.is_none());
215    }
216
217    #[test]
218    fn test_succeed() {
219        let mut task = Task::new(TaskData::default());
220
221        task.set_succeed();
222        match task.status {
223            TaskStatus::Completed => {}
224            _ => panic!("Wrong status (task.status)"),
225        };
226        assert!(task.finished_at.is_some());
227        assert!(task.started_at.is_none());
228    }
229
230    #[test]
231    fn test_set_retry() {
232        let mut task = Task::new(TaskData::default());
233
234        task.set_retry("Wrong task value");
235        match task.status {
236            TaskStatus::Failed => {}
237            _ => panic!("Wrong status (task.status)"),
238        };
239        assert!(task.finished_at.is_some());
240        assert_eq!(task.retries, 1);
241        assert!(task.error_msg.is_some());
242        assert!(task.started_at.is_none());
243    }
244
245    #[test]
246    fn test_set_dlq() {
247        let mut task = Task::new(TaskData::default());
248
249        task.set_dlq("Wrong task value");
250        match task.status {
251            TaskStatus::DeadLetter => {}
252            _ => panic!("Wrong status (task.status)"),
253        };
254        assert!(task.finished_at.is_some());
255        assert!(task.started_at.is_none());
256        assert!(task.error_msg.is_some());
257    }
258
259    #[test]
260    fn task_flow_succeed() {
261        let mut task = Task::new(TaskData::default());
262
263        task.set_in_progress();
264        task.payload.value += 1;
265
266        std::thread::sleep(std::time::Duration::from_millis(5));
267
268        task.set_retry("Wrong task value");
269        task.payload.value += 1;
270
271        task.set_in_progress();
272        std::thread::sleep(std::time::Duration::from_millis(5));
273
274        task.set_succeed();
275
276        match task.status {
277            TaskStatus::Completed => {}
278            _ => panic!("Wrong status (task.status)"),
279        };
280        assert_eq!(task.retries, 1);
281        assert_eq!(task.get_payload().value, 2);
282        assert!(task.started_at.is_some());
283        assert!(task.finished_at.is_some());
284
285        // finished_at - started_at
286        assert!(
287            task.finished_at
288                .unwrap()
289                .duration_since(task.started_at.unwrap())
290                .unwrap()
291                < std::time::Duration::from_millis(50)
292        );
293        // finished_at - queue_at
294        assert!(
295            task.finished_at
296                .unwrap()
297                .duration_since(task.queued_at)
298                .unwrap()
299                >= std::time::Duration::from_millis(10)
300        );
301    }
302
303    #[test]
304    fn test_flow() {
305        let mut task = Task::new(TaskData::default());
306
307        task.set_in_progress();
308        task.payload.value += 1;
309
310        std::thread::sleep(std::time::Duration::from_millis(5));
311
312        task.set_retry("Wrong task value");
313        task.payload.value += 1;
314
315        task.set_in_progress();
316        std::thread::sleep(std::time::Duration::from_millis(5));
317
318        task.set_dlq("Failed to complete task");
319
320        match task.status {
321            TaskStatus::DeadLetter => {}
322            _ => panic!("Wrong status (task.status)"),
323        };
324        assert_eq!(task.retries, 1);
325        assert_eq!(task.get_payload().value, 2);
326        assert!(task.started_at.is_some());
327        assert!(task.finished_at.is_some());
328
329        // finished_at - started_at
330        assert!(
331            task.finished_at
332                .unwrap()
333                .duration_since(task.started_at.unwrap())
334                .unwrap()
335                < std::time::Duration::from_millis(50)
336        );
337        // finished_at - queue_at
338        assert!(
339            task.finished_at
340                .unwrap()
341                .duration_since(task.queued_at)
342                .unwrap()
343                >= std::time::Duration::from_millis(10)
344        );
345    }
346}
capp_queue/task.rs

capp_queue/
task.rs
