Orka is an asynchronous, pluggable and type-safe workflow engine for Rust, designed to orchestrate complex multi-step business processes with robust context management and conditional logic. It simplifies the development of intricate, stateful workflows by providing a clear structure for defining steps, managing shared data, handling errors consistently, and enabling dynamic execution paths, thereby improving code organization and maintainability for complex operations.
Orka 0.2.0 is vastly simpler than 0.1.0, no more turbofishes, no more littering box pin all over, it got unbelievably better. The example look vastly better than 0.1.0.
use orka::prelude::*;
#[derive(Clone, Debug, Default)]
struct OrderContext {
order_id: String,
total: u64,
paid: bool,
}
#[derive(Debug, thiserror::Error)]
enum AppError {
#[error(transparent)]
Orka(#[from] OrkaError),
#[error("payment declined: {0}")]
Declined(String),
}
#[tokio::main]
async fn main() -> Result<(), AppError> {
let mut pipeline = Pipeline::<OrderContext, AppError>::new(["price", "charge", "notify"]);
pipeline
.optional("notify")
.skip_if("charge", |ctx| ctx.read().total == 0)
.on_root("price", |ctx| async move {
ctx.write().total = 4_200;
Ok(PipelineControl::Continue)
})
.on_root("charge", |ctx| async move {
let total = ctx.read().total;
if total > 10_000 {
return Err(AppError::Declined("over limit".into()));
}
ctx.write().paid = true;
Ok(PipelineControl::Continue)
})
.on_root("notify", |ctx| async move {
println!("order {} paid", ctx.read().order_id);
Ok(PipelineControl::Continue)
});
let orka = Orka::<AppError>::new();
orka.register_pipeline(pipeline)?;
let ctx = ContextData::new(OrderContext::default());
match orka.run(ctx.clone()).await? {
PipelineResult::Completed => println!("done, paid = {}", ctx.read().paid),
PipelineResult::Stopped => println!("stopped early"),
}
Ok(())
}
Compare this to 0.1.0
use orka::{Pipeline, ContextData, PipelineControl, OrkaError, OrkaResult, Orka}; use std::sync::Arc; use tracing::info; // For logging // Define Context Data #[derive(Clone, Debug, Default)] struct MySimpleContext { counter: i32, message: String, } // Define Application Error (must be From<OrkaError>) #[derive(Debug, thiserror::Error)] enum MyAppError { #[error("Orka framework error: {0}")] Orka(#[from] OrkaError), #[error("Task failed: {0}")] Task(String), } async fn run_simple_pipeline_example() -> Result<(), MyAppError> { // Initialize tracing (optional, for example output) // tracing_subscriber::fmt::init(); // 1. Define the pipeline let mut pipeline = Pipeline::<MySimpleContext, MyAppError>::new(&[ ("initialize", false, None), ("increment", false, None), ("finalize", false, None), ]); // 2. Register handlers pipeline.on_root("initialize", |ctx_data: ContextData<MySimpleContext>| { Box::pin(async move { let mut guard = ctx_data.write(); guard.counter = 0; guard.message = "Initialized".to_string(); info!("Step Initialize: Counter set to {}, Message: '{}'", guard.counter, guard.message); Ok(PipelineControl::Continue) }) }); pipeline.on_root("increment", |ctx_data: ContextData<MySimpleContext>| { Box::pin(async move { let mut guard = ctx_data.write(); guard.counter += 1; // Example of a handler-specific error if guard.counter > 5 { // Arbitrary condition for failure return Err(MyAppError::Task("Counter exceeded limit in increment step".to_string())); } info!("Step Increment: Counter incremented to {}", guard.counter); Ok(PipelineControl::Continue) }) }); pipeline.on_root("finalize", |ctx_data: ContextData<MySimpleContext>| { Box::pin(async move { let guard = ctx_data.read(); guard.message.push_str(" and Finalized"); info!("Step Finalize: Message: '{}'", guard.message); Ok(PipelineControl::Continue) }) }); // 3. Create initial context let initial_context = ContextData::new(MySimpleContext::default()); // 4. Run the pipeline (can be run directly or via Orka registry) info!("Running pipeline directly..."); let result = pipeline.run(initial_context.clone()).await?; info!("Pipeline direct run result: {:?}", result); let final_guard = initial_context.read(); assert_eq!(final_guard.counter, 1); assert_eq!(final_guard.message, "Initialized and Finalized"); info!("Final Context: Counter = {}, Message = '{}'", final_guard.counter, final_guard.message); Ok(()) } // To run this (e.g., in a test or main): // #[tokio::main] // async fn main() { run_simple_pipeline_example().await.unwrap(); }
0.2.0 pushes the bar on simplicity for workflow engines.
Orka is used in all Excerion Sun LLC applications, it's used primarily in the application bootstrap process, shockingly. I never imagined it would be used there, but there it is. It was realistically the best way to clearly define the bootstrap process and allow the flexibility you want without recreating a bunch of simple error prone code that Orka already does for free and tested. Application bootstrap requires a set of defined steps where some configuration options may or may not disable the step or even add or extend steps.
Originally Orka was imagined to be a pipeline, a way to make SaaS applications easier to create since a lot of it is just the same processes over and over. Payments use the same payment flow typically, so with Orka you could create a step that could hook into the engine and export it via a library to make it reusable. Some steps just need some pieces of data from context data to execute what they need to execute, so there are mappers from context data to the domain data that would be used by the externally provided step. Enough of these kinds of steps around and you can easily mix and match steps to accomplish what you need.
Fortunately, that is still the vision: sharable pipeline steps that you can hook into your SaaS applications to create workflows using Orka.