apalis_workflow/sequential/
context.rs

1use std::num::ParseIntError;
2
3use apalis_core::{
4    backend::WireFormatBackend,
5    task::metadata::{Metadata, MetadataError, MetadataStore},
6};
7use serde::{Deserialize, Serialize};
8
9/// Context information for the current step in the workflow
10#[derive(Debug)]
11pub struct StepContext<B>
12where
13    B: WireFormatBackend,
14{
15    /// Index of the current step
16    pub current_step: usize,
17    /// Backend associated with the current step
18    pub backend: B,
19    /// Indicates if there is a next step
20    pub has_next: bool,
21}
22
23impl<B> Clone for StepContext<B>
24where
25    B: WireFormatBackend + Clone,
26{
27    fn clone(&self) -> Self {
28        Self {
29            current_step: self.current_step,
30            backend: self.backend.clone(),
31            has_next: self.has_next,
32        }
33    }
34}
35impl<B> StepContext<B>
36where
37    B: WireFormatBackend,
38{
39    /// Creates a new StepContext
40    pub fn new(backend: B, idx: usize, has_next: bool) -> Self {
41        Self {
42            current_step: idx,
43            backend,
44            has_next,
45        }
46    }
47}
48
49/// Metadata stored in each task for workflow processing
50#[derive(Debug, Clone, Deserialize, Serialize, Default)]
51pub struct WorkflowContext {
52    /// Index of the step in the workflow
53    pub step_index: usize,
54}
55
56#[non_exhaustive]
57/// Represents an invalid [`WorkflowContext`] state
58#[derive(Debug, thiserror::Error)]
59pub enum WorkflowContextError {
60    /// An entry for the key is missing
61    #[error("the data for key {WORKFLOW_CONTEXT_KEY} is missing")]
62    MissingKey,
63    /// Could not parse the value provided
64    #[error("Could not parse key {WORKFLOW_CONTEXT_KEY}")]
65    Parse(#[from] ParseIntError),
66
67    /// Duplicate entry
68    #[error("Duplicate entry: {0}")]
69    DuplicateEntry(#[from] MetadataError),
70}
71
72const WORKFLOW_CONTEXT_KEY: &str = "apalis_workflow.context.step_index";
73
74impl Metadata for WorkflowContext {
75    type Error = WorkflowContextError;
76
77    fn extract(map: &MetadataStore) -> Result<Self, Self::Error> {
78        let step_index = map
79            .get(WORKFLOW_CONTEXT_KEY)
80            .ok_or(WorkflowContextError::MissingKey)?
81            .parse::<usize>()
82            .map_err(WorkflowContextError::Parse)?;
83
84        Ok(Self { step_index })
85    }
86
87    fn inject(&self, map: &mut MetadataStore) -> Result<(), WorkflowContextError> {
88        map.insert(WORKFLOW_CONTEXT_KEY, self.step_index.to_string())?;
89        Ok(())
90    }
91}
apalis_workflow/sequential/context.rs

apalis_workflow/sequential/
context.rs
