For complex workflows where steps have dependencies, use DagFlow.
- Initialize a new flow with
DagFlow::new("name"). - Define nodes using
.node(function). - Establish dependencies using
.depends_on((&node_a, &node_b, ...)). - Crucial: Call
.validate()? to ensure the graph structure is valid before running. - The collector node (the final step) can receive a tuple of results from its dependencies. The types and order of the tuple must match the dependencies provided in
depends_on.
To stop a workflow manually from within a node, you can use WorkerContext::stop().
use apalis::prelude::*;
use apalis_file_storage::JsonStorage;
use apalis_workflow::{DagFlow, WorkflowSink};
use serde_json::Value;
async fn get_name(user_id: u32) -> Result<String, BoxDynError> {
Ok(user_id.to_string())
}
async fn get_age(user_id: u32) -> Result<usize, BoxDynError> {
Ok(user_id as usize + 20)
}
async fn get_address(user_id: u32) -> Result<usize, BoxDynError> {
Ok(user_id as usize + 100)
}
async fn collector(
(name, age, address): (String, usize, usize),
wrk: WorkerContext,
) -> Result<usize, BoxDynError> {
let result = name.parse::<usize>()? + age + address;
wrk.stop().unwrap();
Ok(result)
}
#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
let mut backend = JsonStorage::new_temp().unwrap();
backend
.push_start(vec![42, 43, 44])
.await
.unwrap();
let dag_flow = DagFlow::new("user-etl-workflow");
let get_name = dag_flow.node(get_name);
let get_age = dag_flow.node(get_age);
let get_address = dag_flow.node(get_address);
dag_flow
.node(collector)
.depends_on((&get_name, &get_age, &get_address)); // Order and types matters here
dag_flow.validate()?; // Ensure DAG is valid
info!("Executing workflow:\n{}", dag_flow); // Print the DAG structure in dot format
WorkerBuilder::new("tasty-banana")
.backend(backend)
.enable_tracing()
.on_event(|_c, e| info!("{e}"))
.build(dag_flow)
.run()
.await?;
Ok(())
}
