apalis_workflow/graph/
mod.rs

1use std::{
2    collections::{HashMap, VecDeque},
3    fmt::{self, Debug},
4    marker::PhantomData,
5    sync::Mutex,
6};
7
8use apalis_core::{
9    backend::{Backend, BackendConfig, WireFormatBackend, codec::Codec},
10    error::BoxDynError,
11    task::Task,
12    task::task_fn::{TaskFn, task_fn},
13};
14use petgraph::{
15    Direction,
16    algo::toposort,
17    dot::Config,
18    graph::{DiGraph, EdgeIndex, NodeIndex},
19};
20/// Graph service implementations
21pub mod service;
22
23/// Graph error definitions
24pub mod error;
25/// Graph node implementations
26pub mod node;
27
28/// Graph context implementations
29pub mod context;
30
31/// Graph response implementations
32pub mod response;
33
34/// Graph Decoding and encoding utilities
35pub mod decode;
36
37use serde::{Serialize, de::DeserializeOwned};
38use serde_json::Value;
39use tower::{Service, util::BoxCloneSyncService};
40
41use crate::{
42    NodeService,
43    graph::{decode::GraphCodec, error::GraphFlowError, node::GraphNodeService},
44};
45
46pub use context::GraphFlowContext;
47pub use service::RootGraphService;
48
49/// Directed Acyclic Graph (Graph) workflow builder
50#[derive(Debug)]
51pub struct GraphFlow<B>
52where
53    B: WireFormatBackend,
54{
55    pub(crate) name: String,
56    graph: Mutex<DiGraph<NodeService<B::Compact>, ()>>,
57    node_mapping: Mutex<HashMap<String, NodeIndex>>,
58}
59
60impl<B> fmt::Display for GraphFlow<B>
61where
62    B: Backend + WireFormatBackend,
63{
64    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
65        writeln!(f, "Graph name: {}", self.name)?;
66        writeln!(f, "Dot format:")?;
67        f.write_str(&self.to_dot())
68    }
69}
70
71impl<B> GraphFlow<B>
72where
73    B: Backend + WireFormatBackend,
74{
75    /// Create a new Graph workflow builder
76    #[must_use]
77    pub fn new(name: &str) -> Self {
78        Self {
79            name: name.to_owned(),
80            graph: Mutex::new(DiGraph::new()),
81            node_mapping: Mutex::new(HashMap::new()),
82        }
83    }
84    /// Create a new Graph workflow builder with typed inputs and output
85    #[must_use]
86    pub fn new_typed(name: &str) -> Self {
87        Self {
88            name: name.to_owned(),
89            graph: Mutex::new(DiGraph::new()),
90            node_mapping: Mutex::new(HashMap::new()),
91        }
92    }
93
94    /// Export the Graph to DOT format
95    pub fn to_dot(&self) -> String {
96        let names = self
97            .node_mapping
98            .lock()
99            .expect("could not lock nodes")
100            .iter()
101            .map(|(name, &idx)| (idx, name.clone()))
102            .collect::<HashMap<_, _>>();
103        let get_node_attributes = |_, (index, _)| {
104            format!(
105                "label=\"{}\"",
106                names.get(&index).cloned().unwrap_or_default()
107            )
108        };
109        let graph = self.graph.lock().expect("could not lock graph");
110        let dot = petgraph::dot::Dot::with_attr_getters(
111            &*graph,
112            &[Config::NodeNoLabel, Config::EdgeNoLabel],
113            &|_, _| String::new(),
114            &get_node_attributes,
115        );
116        format!("{dot:?}")
117    }
118}
119
120impl<B> GraphFlow<B>
121where
122    B: Backend + WireFormatBackend,
123{
124    /// Add a named node to the Graph
125    #[must_use]
126    pub fn add_node<S, Input, CodecError>(
127        &self,
128        name: &str,
129        service: S,
130    ) -> NodeBuilder<'_, Input, S::Response, B>
131    where
132        S: Service<Task<Input>> + Send + 'static + Sync + Clone,
133        S::Future: Send + 'static,
134        B::Codec: Codec<Input, Compact = B::Compact, Error = CodecError>
135            + Codec<S::Response, Compact = B::Compact, Error = CodecError>
136            + Send
137            + Sync
138            + Clone
139            + 'static,
140        CodecError: Into<BoxDynError> + Send + 'static,
141        S::Error: Into<BoxDynError>,
142        B: BackendConfig + Send + Sync + 'static,
143        Input: GraphCodec<B, Error = CodecError> + Send + Sync + 'static + DeserializeOwned,
144        S::Response: Serialize,
145    {
146        let svc: GraphNodeService<S, B, Input> = GraphNodeService::new(service);
147        let node = self
148            .graph
149            .lock()
150            .expect("Failed to lock graph mutex")
151            .add_node(BoxCloneSyncService::new(svc));
152        self.node_mapping
153            .lock()
154            .expect("Failed to lock node_mapping mutex")
155            .insert(name.to_owned(), node);
156        NodeBuilder {
157            id: node,
158            graph: self,
159            _phantom: PhantomData,
160        }
161    }
162
163    /// Add a task function node to the Graph
164    pub fn add_task<F, Input, O, FnArgs, Err, CodecError>(
165        &self,
166        task: F,
167    ) -> NodeBuilder<'_, Input, O, B>
168    where
169        TaskFn<F, Input, FnArgs>: Service<Task<Input>, Response = O, Error = Err> + Clone,
170        F: Send + 'static + Sync,
171        Input: Send + 'static + Sync,
172        FnArgs: Send + 'static + Sync,
173        <TaskFn<F, Input, FnArgs> as Service<Task<Input>>>::Future: Send + 'static,
174        B::Codec: Codec<Input, Compact = B::Compact, Error = CodecError> + 'static,
175        B::Codec:
176            Codec<O, Compact = B::Compact, Error = CodecError> + Send + Sync + Clone + 'static,
177        CodecError: Into<BoxDynError> + Send + 'static,
178        Err: Into<BoxDynError>,
179        B: BackendConfig + Send + Sync + 'static,
180        Input: GraphCodec<B, Error = CodecError> + Send + Sync + DeserializeOwned + 'static,
181        O: Serialize,
182    {
183        self.add_node(std::any::type_name::<F>(), task_fn(task))
184    }
185
186    /// Add a routing node to the Graph
187    pub fn route<F, Input, O, FnArgs, Err, CodecError>(
188        &self,
189        router: F,
190    ) -> NodeBuilder<'_, Input, O, B>
191    where
192        TaskFn<F, Input, FnArgs>: Service<Task<Input>, Response = O, Error = Err> + Clone,
193        F: Send + 'static + Sync,
194        Input: Send + 'static + Sync,
195        FnArgs: Send + 'static + Sync,
196        <TaskFn<F, Input, FnArgs> as Service<Task<Input>>>::Future: Send + 'static,
197        O: Into<NodeIndex> + Serialize,
198        B::Codec: Codec<Input, Compact = B::Compact, Error = CodecError> + 'static,
199        B::Codec:
200            Codec<O, Compact = B::Compact, Error = CodecError> + Send + Sync + Clone + 'static,
201        CodecError: Into<BoxDynError> + Send + 'static,
202        Err: Into<BoxDynError>,
203        B: BackendConfig + Send + Sync + 'static,
204        Input: GraphCodec<B, Error = CodecError> + Send + Sync + 'static + DeserializeOwned,
205    {
206        self.add_node::<TaskFn<F, Input, FnArgs>, Input, CodecError>(
207            std::any::type_name::<F>(),
208            task_fn(router),
209        )
210    }
211
212    /// Validate the Graph for cycles
213    pub fn validate(&self) -> Result<(), GraphFlowError> {
214        // Validate Graph (check for cycles)
215        toposort(
216            &*self.graph.lock().expect("Failed to lock graph mutex"),
217            None,
218        )
219        .map_err(GraphFlowError::CyclicDAG)?;
220        Ok(())
221    }
222
223    /// Build the Graph executor
224    pub(crate) fn build(self, backend: B) -> Result<RootGraphService<B>, GraphFlowError> {
225        // Validate Graph (check for cycles)
226        let sorted = toposort(
227            &*self.graph.lock().expect("Failed to lock graph mutex"),
228            None,
229        )
230        .map_err(GraphFlowError::CyclicDAG)?;
231
232        fn find_edge_nodes<N, E>(graph: &DiGraph<N, E>, direction: Direction) -> Vec<NodeIndex> {
233            graph
234                .node_indices()
235                .filter(|&n| graph.neighbors_directed(n, direction).count() == 0)
236                .collect()
237        }
238
239        let graph = self
240            .graph
241            .into_inner()
242            .expect("Failed to unlock graph mutex");
243
244        Ok(RootGraphService {
245            start_nodes: find_edge_nodes(&graph, Direction::Incoming),
246            end_nodes: find_edge_nodes(&graph, Direction::Outgoing),
247            graph,
248            node_mapping: self
249                .node_mapping
250                .into_inner()
251                .expect("Failed to unlock node_mapping mutex"),
252            topological_order: sorted,
253            not_ready: VecDeque::new(),
254            backend,
255        })
256    }
257}
258
259/// Builder for a node in the Graph
260pub struct NodeBuilder<'a, Input, Output, B>
261where
262    B: Backend + WireFormatBackend,
263{
264    pub(crate) id: NodeIndex,
265    pub(crate) graph: &'a GraphFlow<B>,
266    _phantom: PhantomData<(Input, Output)>,
267}
268
269impl<'a, Input, Output, B> std::fmt::Debug for NodeBuilder<'a, Input, Output, B>
270where
271    B: Backend + WireFormatBackend,
272{
273    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
274        f.debug_struct("NodeBuilder")
275            .field("id", &self.id)
276            .finish_non_exhaustive()
277    }
278}
279
280impl<'a, Input, Output, B> Clone for NodeBuilder<'a, Input, Output, B>
281where
282    B: Backend + WireFormatBackend,
283{
284    fn clone(&self) -> Self {
285        Self {
286            id: self.id,
287            graph: self.graph,
288            _phantom: PhantomData,
289        }
290    }
291}
292
293impl<Input, Output, B> NodeBuilder<'_, Input, Output, B>
294where
295    B: Backend + WireFormatBackend,
296{
297    /// Specify dependencies for this node
298    #[allow(clippy::needless_pass_by_value)]
299    pub fn depends_on<D>(self, deps: D) -> NodeHandle<Input, Output>
300    where
301        D: DepsCheck<Input>,
302    {
303        let mut edges = Vec::new();
304        for dep in deps.to_node_indices() {
305            edges.push(self.graph.graph.lock().unwrap().add_edge(dep, self.id, ()));
306        }
307        NodeHandle {
308            id: self.id,
309            edges,
310            _phantom: PhantomData,
311        }
312    }
313}
314
315/// Handle for a node in the Graph
316#[derive(Clone, Debug)]
317pub struct NodeHandle<Input, Output> {
318    pub(crate) id: NodeIndex,
319    pub(crate) edges: Vec<EdgeIndex>,
320    pub(crate) _phantom: PhantomData<(Input, Output)>,
321}
322
323impl<Input, Output> NodeHandle<Input, Output> {
324    /// Get the node ID
325    #[must_use]
326    pub fn id(&self) -> NodeIndex {
327        self.id
328    }
329
330    /// Get the edge IDs
331    #[must_use]
332    pub fn edges(&self) -> &[EdgeIndex] {
333        &self.edges
334    }
335}
336
337/// Trait for converting dependencies into node IDs
338pub trait DepsCheck<Input> {
339    /// Convert dependencies to node indices
340    fn to_node_indices(&self) -> Vec<NodeIndex>;
341}
342
343impl<'a, Input, Output, B> DepsCheck<Output> for &NodeBuilder<'a, Input, Output, B>
344where
345    B: Backend + WireFormatBackend,
346{
347    fn to_node_indices(&self) -> Vec<NodeIndex> {
348        vec![self.id]
349    }
350}
351
352impl<'a, Input, Output, B> DepsCheck<Output> for NodeBuilder<'a, Input, Output, B>
353where
354    B: Backend + WireFormatBackend,
355{
356    fn to_node_indices(&self) -> Vec<NodeIndex> {
357        vec![self.id]
358    }
359}
360
361impl<Input, Output> DepsCheck<Output> for &NodeHandle<Input, Output> {
362    fn to_node_indices(&self) -> Vec<NodeIndex> {
363        vec![self.id]
364    }
365}
366
367impl<Input, Output> DepsCheck<Output> for NodeHandle<Input, Output> {
368    fn to_node_indices(&self) -> Vec<NodeIndex> {
369        vec![self.id]
370    }
371}
372
373impl<Input, Output> DepsCheck<Output> for (&NodeHandle<Input, Output>,) {
374    fn to_node_indices(&self) -> Vec<NodeIndex> {
375        vec![self.0.id]
376    }
377}
378
379impl<Input, Output> DepsCheck<Output> for (NodeHandle<Input, Output>,) {
380    fn to_node_indices(&self) -> Vec<NodeIndex> {
381        vec![self.0.id]
382    }
383}
384
385impl<'a, Input, Output, B> DepsCheck<Output> for (NodeBuilder<'a, Input, Output, B>,)
386where
387    B: Backend + WireFormatBackend,
388{
389    fn to_node_indices(&self) -> Vec<NodeIndex> {
390        vec![self.0.id]
391    }
392}
393
394impl<'a, Input, Output, B> DepsCheck<Output> for (&NodeBuilder<'a, Input, Output, B>,)
395where
396    B: Backend + WireFormatBackend,
397{
398    fn to_node_indices(&self) -> Vec<NodeIndex> {
399        vec![self.0.id]
400    }
401}
402
403impl<Output, T: DepsCheck<Output>> DepsCheck<Vec<Output>> for Vec<T> {
404    fn to_node_indices(&self) -> Vec<NodeIndex> {
405        assert!(self.len() > 1, "dependencies should be more than one");
406        self.iter()
407            .flat_map(|item| item.to_node_indices())
408            .collect()
409    }
410}
411
412macro_rules! impl_deps_check {
413    ($( $len:literal => ( $( $in:ident $out:ident $idx:tt ),+ ) ),+ $(,)?) => {
414        $(
415            impl<'a, $( $in, )+ $( $out, )+ B> DepsCheck<( $( $out, )+ )>
416                for ( $( &NodeBuilder<'a, $in, $out, B>, )+ ) where B: Backend + WireFormatBackend
417            {
418                fn to_node_indices(&self) -> Vec<NodeIndex> {
419                    vec![ $( self.$idx.id ),+ ]
420                }
421            }
422
423            impl<$( $in, )+ $( $out, )+> DepsCheck<( $( $out, )+ )>
424                for ( $( &NodeHandle<$in, $out>, )+ )
425            {
426                fn to_node_indices(&self) -> Vec<NodeIndex> {
427                    vec![ $( self.$idx.id ),+ ]
428                }
429            }
430        )+
431    };
432}
433
434impl_deps_check! {
435    1 => (Input1 Output1 0),
436    2 => (Input1 Output1 0, Input2 Output2 1),
437    3 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2),
438    4 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3),
439    5 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3, Input5 Output5 4),
440    6 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3, Input5 Output5 4, Input6 Output6 5),
441    7 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3, Input5 Output5 4, Input6 Output6 5, Input7 Output7 6),
442    8 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3, Input5 Output5 4, Input6 Output6 5, Input7 Output7 6, Input8 Output8 7),
443}
444
445/// Input for graph nodes
446#[derive(Debug, Clone)]
447#[non_exhaustive]
448pub enum NodeInput<Compact> {
449    /// A single node with compact data
450    Single(Compact),
451    /// Fan in input from previous nodes
452    FanIn(Vec<Value>),
453}
454
455#[cfg(test)]
456mod tests {
457    use std::{collections::BTreeMap, num::ParseIntError};
458
459    use apalis_core::{
460        error::BoxDynError,
461        task::task_fn::task_fn,
462        worker::{
463            builder::WorkerBuilder, context::WorkerContext, event::Event,
464            ext::event_listener::EventListenerExt,
465        },
466    };
467    use apalis_file_storage::JsonStorage;
468    use futures_util::StreamExt;
469    use petgraph::graph::NodeIndex;
470    use serde::{Deserialize, Serialize};
471    use serde_json::Value;
472
473    use crate::{WorkflowSink, graph::response::GraphNodeResponse, in_memory::InMemoryWorkflow};
474
475    use super::*;
476
477    #[tokio::test]
478    async fn test_basic_workflow() {
479        let graph = GraphFlow::new("sequential-workflow");
480        let start = graph.add_node("start", task_fn(|task: u32| async move { task as usize }));
481        let middle = graph
482            .add_node(
483                "middle",
484                task_fn(|task: usize| async move { task.to_string() }),
485            )
486            .depends_on(&start);
487
488        let _end = graph
489            .add_node(
490                "end",
491                task_fn(|task: String, worker: WorkerContext| async move {
492                    worker.stop().unwrap();
493                    task.parse::<usize>()
494                }),
495            )
496            .depends_on(&middle);
497
498        println!("Graph in DOT format:\n{}", graph.to_dot());
499
500        let mut backend = JsonStorage::new_temp().unwrap();
501
502        backend.push_start(42).await.unwrap();
503
504        let worker = WorkerBuilder::new("rango-tango")
505            .backend(backend)
506            .on_event(|worker, ev| {
507                println!("On Event = {ev:?}");
508                if matches!(ev, Event::Error(_)) {
509                    worker.stop().unwrap();
510                }
511            })
512            .build(graph);
513        worker.run().await.unwrap();
514    }
515
516    #[tokio::test]
517    async fn test_fan_out_workflow() {
518        let graph = GraphFlow::new("fan-out-workflow");
519        let source = graph.add_node("source", task_fn(|task: u32| async move { task as usize }));
520        let plus_one = graph
521            .add_node("plus_one", task_fn(|task: usize| async move { task + 1 }))
522            .depends_on(&source);
523
524        let multiply = graph
525            .add_node("multiply", task_fn(|task: usize| async move { task * 2 }))
526            .depends_on(&source);
527        let squared = graph
528            .add_node("squared", task_fn(|task: usize| async move { task * task }))
529            .depends_on(&source);
530
531        let _collector = graph
532            .add_node(
533                "collector",
534                task_fn(|task: (usize, usize, usize), w: WorkerContext| async move {
535                    w.stop().unwrap();
536                    task.0 + task.1 + task.2
537                }),
538            )
539            .depends_on((&plus_one, &multiply, &squared));
540
541        println!("Graph in DOT format:\n{}", graph.to_dot());
542
543        let mut backend = InMemoryWorkflow::create();
544
545        backend.push_start(42).await.unwrap();
546
547        let worker = WorkerBuilder::new("rango-tango")
548            .backend(backend)
549            .on_event(|_w, ev| {
550                println!("On Event = {ev:?}");
551                if matches!(ev, Event::Error(_)) {
552                    panic!("{}", ev);
553                }
554            })
555            .build(graph);
556        worker.run().await.unwrap();
557    }
558
559    #[tokio::test]
560    async fn test_fan_in_workflow() {
561        let graph = GraphFlow::new("fan-in-workflow");
562        let get_name = graph.add_node(
563            "get_name",
564            task_fn(|task: usize| async move { task as usize }),
565        );
566        let get_age = graph.add_node(
567            "get_age",
568            task_fn(|task: u32| async move { task.to_string() }),
569        );
570        let get_address = graph.add_node(
571            "get_address",
572            task_fn(|task: i32| async move { task as usize }),
573        );
574        let main_collector = graph
575            .add_node(
576                "main_collector",
577                task_fn(|task: (String, usize, usize)| async move {
578                    task.2 + task.1 + task.0.parse::<usize>().unwrap()
579                }),
580            )
581            .depends_on((&get_age, &get_name, &get_address));
582
583        let side_collector = graph
584            .add_node(
585                "side_collector",
586                task_fn(
587                    |task: (usize, usize)| async move { [task.0, task.1].iter().sum::<usize>() },
588                ),
589            )
590            .depends_on((&get_name, &get_address));
591
592        let _final_node = graph
593            .add_node(
594                "final_node",
595                task_fn(|task: (usize, usize), w: WorkerContext| async move {
596                    w.stop().unwrap();
597                    task.0 + task.1
598                }),
599            )
600            .depends_on((&main_collector, &side_collector));
601
602        println!("Graph in DOT format:\n{}", graph.to_dot());
603
604        let mut backend: JsonStorage<Value> = JsonStorage::new_temp().unwrap();
605
606        backend
607            .start_fan_out((42usize, 43u32, 44i32))
608            .await
609            .unwrap();
610
611        let worker = WorkerBuilder::new("rango-tango")
612            .backend(backend)
613            .on_event(|_w, ev| {
614                println!("On Event = {ev:?}");
615                if matches!(ev, Event::Error(_)) {
616                    panic!("{}", ev);
617                }
618            })
619            .build(graph);
620        worker.run().await.unwrap();
621    }
622
623    #[tokio::test]
624    async fn fan_in_completes_once_with_testworker() {
625        use apalis_core::task::task_fn::task_fn;
626        use apalis_file_storage::JsonStorage;
627        use serde_json::Value;
628
629        let graph = GraphFlow::new("fan-in-testworker");
630
631        let a = graph.add_node("a", task_fn(|t: u32| async move { t }));
632        let b = graph.add_node("b", task_fn(|t: u32| async move { t }));
633
634        let _fan_in = graph
635            .add_node("fan_in", task_fn(|t: (u32, u32)| async move { t.0 + t.1 }))
636            .depends_on((&a, &b));
637
638        let mut backend: JsonStorage<Value> = JsonStorage::new_temp().unwrap();
639        backend.start_fan_out((1u32, 2u32)).await.unwrap();
640
641        let worker = WorkerBuilder::new("test-worker")
642            .backend(backend)
643            .build(graph)
644            .stream();
645
646        let (res_map, final_res) = worker.take(6).collect::<Vec<_>>().await.into_iter().fold(
647            (BTreeMap::new(), None),
648            |(mut res_map, final_res), item| {
649                let resp = match item {
650                    Ok(Event::Success(res)) => res.downcast::<GraphNodeResponse>().unwrap(),
651                    Ok(Event::Error(e)) => panic!("execution error {e}"),
652                    Ok(_) => return (res_map, final_res),
653                    Err(e) => panic!("worker error: {e:?}"),
654                };
655
656                let kind = match *resp {
657                    GraphNodeResponse::EntryFanOut { .. } => "EntryFanOut",
658                    GraphNodeResponse::FanOut { .. } => "FanOut",
659                    GraphNodeResponse::EnqueuedNext { .. } => "EnqueuedNext",
660                    GraphNodeResponse::WaitingForDependencies { .. } => "WaitingForDependencies",
661                    GraphNodeResponse::Complete { .. } => "Complete",
662                };
663                *res_map.entry(kind).or_insert(0) += 1;
664
665                if let GraphNodeResponse::Complete { result } = *resp {
666                    return (res_map, Some(result));
667                }
668                (res_map, final_res)
669            },
670        );
671
672        assert_eq!(final_res, Some(Value::from(3)));
673        assert_eq!(res_map["EntryFanOut"], 1); // Entry
674        assert_eq!(res_map["EnqueuedNext"], 2); // Node Results
675        assert_eq!(res_map["WaitingForDependencies"], 1); // Non Handler node
676        assert_eq!(res_map["Complete"], 1); // Handler Node
677    }
678
679    #[tokio::test]
680    async fn test_routed_workflow() {
681        let graph = GraphFlow::new("routed-workflow");
682
683        let entry1 = graph.add_node("entry1", task_fn(|task: u32| async move { task as usize }));
684        let entry2 = graph.add_node("entry2", task_fn(|task: u32| async move { task as usize }));
685        let entry3 = graph.add_node("entry3", task_fn(|task: u32| async move { task as usize }));
686
687        #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
688        enum EntryRoute {
689            Entry1(NodeIndex),
690            Entry2(NodeIndex),
691            Entry3(NodeIndex),
692        }
693
694        impl Into<NodeIndex> for EntryRoute {
695            fn into(self) -> NodeIndex {
696                match self {
697                    EntryRoute::Entry1(idx) => idx,
698                    EntryRoute::Entry2(idx) => idx,
699                    EntryRoute::Entry3(idx) => idx,
700                }
701            }
702        }
703
704        impl DepsCheck<usize> for EntryRoute {
705            fn to_node_indices(&self) -> Vec<NodeIndex> {
706                vec![(*self).into()]
707            }
708        }
709
710        async fn collect(task: (usize, usize, usize)) -> usize {
711            task.0 + task.1 + task.2
712        }
713        let collector = graph
714            .add_task(collect)
715            .depends_on((&entry1, &entry2, &entry3));
716
717        async fn vec_collect(task: Vec<usize>, _wrk: WorkerContext) -> usize {
718            task.iter().sum::<usize>()
719        }
720
721        let vec_collector = graph
722            .add_task(vec_collect)
723            .depends_on(vec![&entry1, &entry2, &entry3]);
724
725        async fn exit(task: (usize, usize)) -> Result<u32, ParseIntError> {
726            (task.0.to_string() + &task.1.to_string()).parse()
727        }
728
729        let on_collect = graph
730            .add_task(exit)
731            .depends_on((&collector, &vec_collector));
732
733        async fn check_approval(
734            task: u32,
735            worker: WorkerContext,
736        ) -> Result<EntryRoute, BoxDynError> {
737            println!("Approval check for task: {}", task);
738            worker.stop().unwrap();
739            match task % 3 {
740                0 => Ok(EntryRoute::Entry1(NodeIndex::new(0))),
741                1 => Ok(EntryRoute::Entry2(NodeIndex::new(1))),
742                2 => Ok(EntryRoute::Entry3(NodeIndex::new(2))),
743                _ => Err(BoxDynError::from("Invalid task")),
744            }
745        }
746
747        let _ = graph.route(check_approval).depends_on(&on_collect);
748
749        println!("Graph in DOT format:\n{}", graph.to_dot());
750
751        let mut backend = JsonStorage::new_temp().unwrap();
752
753        backend.start_fan_out(vec![17, 18, 19]).await.unwrap();
754
755        let worker = WorkerBuilder::new("rango-tango")
756            .backend(backend)
757            .on_event(|_, ev| {
758                println!("On Event = {ev:?}");
759                if matches!(ev, Event::Error(_)) {
760                    panic!("{}", ev);
761                }
762            })
763            .build(graph);
764        worker.run().await.unwrap();
765    }
766}
apalis_workflow/graph/mod.rs

apalis_workflow/graph/
mod.rs
