apalis_workflow/sequential/
workflow.rs

1use std::marker::PhantomData;
2
3use apalis_core::{
4    backend::{Backend, BackendConfig, WireFormatBackend},
5    error::BoxDynError,
6    task::{Task, task_id::GenerateId},
7    worker::service::{IntoWorkerService, WorkerService},
8};
9use futures_sink::Sink;
10
11use crate::{
12    sequential::backend::WorkflowBackend,
13    sequential::{
14        router::WorkflowRouter,
15        service::WorkflowService,
16        step::{Identity, Layer, Stack, Step},
17    },
18};
19
20/// A workflow represents a sequence of steps to be executed in order.
21#[derive(Debug)]
22pub struct SteppedFlow<Start, Current, Backend, T = Identity> {
23    pub(crate) inner: T,
24    pub(crate) name: String,
25    _marker: PhantomData<(Start, Current, Backend)>,
26}
27
28impl<Start, Backend> SteppedFlow<Start, Start, Backend> {
29    #[allow(missing_docs)]
30    #[must_use]
31    pub fn new(name: &str) -> Self {
32        Self {
33            inner: Identity,
34            name: name.to_owned(),
35            _marker: PhantomData,
36        }
37    }
38}
39
40impl<Start, Cur, B, L> SteppedFlow<Start, Cur, B, L> {
41    /// Adds a new step to the workflow pipeline.
42    ///
43    /// This method should be used with caution, as it allows adding arbitrary steps
44    /// and manipulating types. It is recommended to use higher-level abstractions for
45    /// common workflow patterns.
46    #[must_use]
47    pub fn add_step<S, Output>(self, step: S) -> SteppedFlow<Start, Output, B, Stack<S, L>> {
48        SteppedFlow {
49            inner: Stack::new(step, self.inner),
50            name: self.name,
51            _marker: PhantomData,
52        }
53    }
54
55    /// Finalizes the workflow by attaching a root step.
56    pub fn finalize<S>(self, root: S) -> SteppedFlow<Start, Cur, B, L::Step>
57    where
58        S: Step<Cur, B>,
59        L: Layer<S>,
60        B: Backend + WireFormatBackend,
61    {
62        SteppedFlow {
63            inner: self.inner.layer(root),
64            name: self.name,
65            _marker: PhantomData,
66        }
67    }
68}
69
70impl<Start, Cur, B, L> SteppedFlow<Start, Cur, B, L>
71where
72    B: Backend,
73{
74    /// Builds the workflow by layering the root step.
75    pub fn build<N>(self) -> L::Step
76    where
77        L: Layer<RootStep<N>>,
78    {
79        let root = RootStep(std::marker::PhantomData);
80        self.inner.layer(root)
81    }
82}
83
84/// The root step of a workflow.
85#[derive(Clone, Debug)]
86pub struct RootStep<Res>(std::marker::PhantomData<Res>);
87
88impl<Res> Default for RootStep<Res> {
89    fn default() -> Self {
90        Self(std::marker::PhantomData)
91    }
92}
93
94impl<Input, Current, B: Backend + WireFormatBackend> Step<Input, B> for RootStep<Current> {
95    type Response = Current;
96    type Error = BoxDynError;
97    fn register(&mut self, _ctx: &mut WorkflowRouter<B>) -> Result<(), BoxDynError> {
98        Ok(())
99    }
100}
101
102impl<Input, Output, Current, B, Compact, L, Err>
103    IntoWorkerService<B, WorkflowService<B, Input, Output>> for SteppedFlow<Input, Current, B, L>
104where
105    B: Backend<Task = Task<Compact>, Error = Err>
106        + WireFormatBackend<Compact = Compact>
107        + BackendConfig<Args = Input>
108        + Send
109        + Sync
110        + 'static
111        + Sink<Task<Compact>, Error = Err>
112        + Clone
113        + Unpin,
114    B::Id: Send + 'static + Default + GenerateId,
115    L: Layer<RootStep<Current>>,
116    L::Step: Step<Output, B>,
117    B::Codec: Clone,
118    Err: std::error::Error + Send + Sync + 'static,
119    Compact: Send + 'static,
120{
121    type Task = Task<Compact>;
122    type Backend = WorkflowBackend<B>;
123    fn into_service(
124        self,
125        backend: B,
126    ) -> WorkerService<Self::Backend, WorkflowService<B, Input, Output>> {
127        let mut ctx = WorkflowRouter::<B>::new();
128
129        let mut root = self.finalize(RootStep(std::marker::PhantomData));
130
131        root.inner
132            .register(&mut ctx)
133            .expect("Failed to register workflow steps");
134
135        WorkerService {
136            service: WorkflowService::new(ctx.steps, backend.clone()),
137            backend: WorkflowBackend::new(backend),
138        }
139    }
140}
apalis_workflow/sequential/workflow.rs

apalis_workflow/sequential/
workflow.rs
