apalis_workflow/
lib.rs

1#![cfg_attr(docsrs, feature(doc_cfg))]
2#![doc = include_str!("../README.md")]
3#![warn(
4    missing_debug_implementations,
5    missing_docs,
6    rust_2018_idioms,
7    unreachable_pub
8)]
9
10use apalis_core::{error::BoxDynError, task::Task};
11
12use crate::{
13    graph::NodeInput,
14    sequential::router::{GoTo, StepResponse},
15};
16
17type BoxedService<Input, Output> = tower::util::BoxCloneSyncService<Input, Output, BoxDynError>;
18type SteppedService<Compact> = BoxedService<Task<Compact>, GoTo<StepResponse>>;
19
20type NodeService<Compact> = BoxedService<Task<NodeInput<Compact>>, (Compact, serde_json::Value)>;
21
22/// combinator for chaining multiple workflows.
23pub mod composite;
24/// utilities for directed acyclic graph workflows.
25pub mod graph;
26
27/// utilities for workflow steps.
28pub mod sequential;
29/// utilities for workflow sinks.
30pub mod sink;
31
32/// In memory backend for running workflows without persistence
33pub mod in_memory;
34
35pub use {graph::GraphFlow, sequential::workflow::SteppedFlow, sink::WorkflowSink};
36
37#[cfg(test)]
38mod tests {
39    use std::{collections::HashMap, num::ParseIntError, time::Duration};
40
41    use apalis_core::{
42        backend::TaskSink,
43        task::{metadata::Meta, task_id::TaskId},
44        worker::{
45            builder::WorkerBuilder, context::WorkerContext, event::Event,
46            ext::event_listener::EventListenerExt,
47        },
48    };
49
50    use crate::{
51        in_memory::InMemoryWorkflow,
52        sequential::{repeat_until::RepeaterState, workflow::SteppedFlow},
53    };
54
55    use super::*;
56
57    #[tokio::test]
58    async fn basic_workflow() {
59        type RepeatUntilState = Meta<RepeaterState>;
60        let workflow = SteppedFlow::new("and-then-workflow")
61            .and_then(async |input: i32| (input) as usize)
62            .delay_for(Duration::from_secs(1))
63            .and_then(async |input: usize| (input) as isize)
64            .delay_for(Duration::from_secs(1))
65            .delay_with(|_| Duration::from_secs(1))
66            .repeat_until(|res, state: RepeatUntilState| async move {
67                println!("Iteration {}: got result {}", state.iterations(), res);
68                if state.iterations() < 3 {
69                    Ok::<_, BoxDynError>(None)
70                } else {
71                    Ok(Some(res))
72                }
73            })
74            .and_then(async |input: isize| Ok::<_, BoxDynError>(input.to_string()))
75            .and_then(async |input: String, _task_id: TaskId| input.parse::<usize>())
76            .and_then(async |res: usize| {
77                Ok::<_, BoxDynError>((0..res).enumerate().collect::<HashMap<_, _>>())
78            })
79            .filter_map(async |(index, input): (usize, usize)| {
80                if input % 2 == 0 {
81                    Some(index.to_string())
82                } else {
83                    None
84                }
85            })
86            .fold(
87                async move |(acc, item): (u32, String), _wrk: WorkerContext| {
88                    println!("Folding item {item} with acc {acc}");
89                    let item = item.parse::<u32>()?;
90                    let acc = acc + item;
91                    Ok::<_, ParseIntError>(acc)
92                },
93            )
94            .and_then(async |res: u32, wrk: WorkerContext| {
95                wrk.stop().unwrap();
96                println!("Completed with {res:?}");
97            });
98
99        let mut backend = InMemoryWorkflow::create();
100
101        backend.push(200).await.unwrap();
102
103        let worker = WorkerBuilder::new("rango-tango")
104            .backend(backend)
105            .on_event(|_, ev| {
106                if matches!(ev, Event::Error(_)) {
107                    panic!("{ev}");
108                }
109            })
110            .build(workflow);
111        worker.run().await.unwrap();
112    }
113}
apalis_workflow/lib.rs

apalis_workflow/
lib.rs
