pipx 0.1.0

Built around the idea that every pipeline should remain explicit, predictable, and easy to reason about.

pipx is a small, framework-agnostic pipeline crate for building clear, composable, and predictable data flows in Rust.

The crate is built around a continuation-based pipeline model that gives each pipe full control over execution. A pipe may continue the chain, wrap the downstream result, short-circuit execution, recover from failures, or return an error.

pipx is fully generic over both the passable value and error type, making it suitable for application pipelines, middleware systems, validation flows, request processing, background jobs, and other data transformation workflows.

Individual pipes may use their own error types as long as they can be converted into PipelineError, allowing applications to maintain domain-specific errors internally while exposing a consistent public execution API.

Features

Type-safe pipelines with Pipe, Next, and Pipeline
Optional asynchronous pipeline support through the async feature
Conditional pipeline composition with when and unless
Error recovery with rescue
Finalization hooks with finally
Helper macros with steps! and async_steps!
Optional procedural macros through the macros feature
Fully generic passable values and error types
Centralized error handling through PipelineError
Support for synchronous and asynchronous execution models
Suitable for middleware, validation, processing, and workflow pipelines

Installation

The pipx crate can be installed with the default feature set or with optional features depending on your use case.

With default

Use the default installation if you only need synchronous pipelines.

Cargo

cargo add pipx

Manual

[dependencies]

pipx = "0.1.0"

Async Feature

Enable the async feature if you want to use asynchronous pipelines.

Cargo

cargo add pipx --features async

Manual

[dependencies]

pipx = { version = "0.1.0", features = ["async"] }

Macros Feature

Enable the macros feature if you want to use procedural macros for pipe implementations.

Cargo

cargo add pipx --features macros

Manual

[dependencies]

pipx = { version = "0.1.0", features = ["macros"] }

Full feature

Enable the full feature if you want to use all available pipx features.

Cargo

cargo add pipx --features full

Manual

[dependencies]

pipx = { version = "0.1.0", features = ["full"] }

Quick Start

Each example demonstrates one behavior at a time so you can quickly understand how pipelines are created, how steps are attached, how execution is finalized, and how synchronous and asynchronous pipelines differ.

Pipeline

Use pipeline when you want to create a synchronous pipeline with an initial value.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Trim;

impl Pipe<String> for Trim {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        // Modify the incoming value before passing it to the next step.
        let passable = passable.trim().to_string();

        // Continue the pipeline with the updated value.
        next.handle(passable)
    }
}

fn main() -> pipx::PipelineResult<()> {
    // Create a pipeline with an initial value.
    let output = pipeline("  hello  ".to_string())
        // Append a single pipeline step.
        .pipe(Trim)
        // Execute the pipeline and return the final value.
        .then_return()?;

    assert_eq!(output, "hello");

    Ok(())
}

Pipeline with `then_return`

Use then_return when you want the pipeline to return the final processed value without applying an additional transformation.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Uppercase;

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        // Convert the value before continuing the pipeline.
        next.handle(passable.to_uppercase())
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("hello".to_string())
        .pipe(Uppercase)
        .then_return()?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Pipeline with `then`

Use then when you want to execute the pipeline and transform the final value into another type.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Uppercase;

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase())
    }
}

fn main() -> pipx::PipelineResult<()> {
    let length = pipeline("hello".to_string())
        .pipe(Uppercase)
        // Execute the pipeline, then map the final String into its length.
        .then(|value| value.len())?;

    assert_eq!(length, 5);

    Ok(())
}

Pipeline with `through`

Use through when you want to set the full list of pipeline steps at once.

use pipx::{pipeline, steps, Next, Pipe, PipelineResult};

struct Trim;
struct Uppercase;

impl Pipe<String> for Trim {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.trim().to_string())
    }
}

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase())
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("  hello  ".to_string())
        // Replace the current pipeline steps with the provided step list.
        .through(steps![Trim, Uppercase])
        .then_return()?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Pipeline with `when`

Use when when you want to append a step only if a condition is true.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Uppercase;

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase())
    }
}

fn main() -> pipx::PipelineResult<()> {
    let should_uppercase = true;

    let output = pipeline("hello".to_string())
        // Append Uppercase only when the condition is true.
        .when(should_uppercase, Uppercase)
        .then_return()?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Pipeline with `unless`

Use unless when you want to append a step only if a condition is false.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Uppercase;

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase())
    }
}

fn main() -> pipx::PipelineResult<()> {
    let skip_uppercase = false;

    let output = pipeline("hello".to_string())
        // Append Uppercase only when the condition is false.
        .unless(skip_uppercase, Uppercase)
        .then_return()?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Pipeline with `finally`

Use finally when you want to run a callback after pipeline execution completes.

The callback runs whether the pipeline succeeds or fails.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Uppercase;

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase())
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("hello".to_string())
        .pipe(Uppercase)
        // Run after the pipeline finishes.
        .finally(|result| {
            println!("Pipeline finished: {result:?}");
        })
        .then_return()?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Pipeline with `rescue`

Use rescue when you want to recover from pipeline execution errors with a fallback value.

use pipx::{pipeline, Next, Pipe, PipelineError, PipelineResult};

struct Fail;

impl Pipe<String> for Fail {
    fn handle(
        &self,
        _passable: String,
        _next: Next<'_, String>,
    ) -> PipelineResult<String> {
        // Stop the pipeline by returning an error.
        Err(PipelineError::Message("pipeline failed".to_string()))
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("hello".to_string())
        .pipe(Fail)
        // Convert the error into a fallback value.
        .rescue(|_| "fallback".to_string())?;

    assert_eq!(output, "fallback");

    Ok(())
}

Pipeline Short-Circuiting

A pipe can stop the chain by not calling next.handle.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Stop;

impl Pipe<String> for Stop {
    fn handle(
        &self,
        passable: String,
        _next: Next<'_, String>,
    ) -> PipelineResult<String> {
        // Return early without continuing to the remaining steps.
        Ok(format!("{passable}:stopped"))
    }
}

struct NeverRuns;

impl Pipe<String> for NeverRuns {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(format!("never:{passable}"))
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("hello".to_string())
        .pipe(Stop)
        .pipe(NeverRuns)
        .then_return()?;

    assert_eq!(output, "hello:stopped");

    Ok(())
}

Pipeline Wrapping

A pipe can call next.handle and then wrap or modify the downstream result.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Wrap;

impl Pipe<String> for Wrap {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        // Continue the pipeline first.
        let result = next.handle(passable)?;

        // Modify the value after downstream steps have completed.
        Ok(format!("[{result}]"))
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("hello".to_string())
        .pipe(Wrap)
        .then_return()?;

    assert_eq!(output, "[hello]");

    Ok(())
}

Async Pipeline

Use async_pipeline when you want to create an asynchronous pipeline.

This requires the async feature.

use async_trait::async_trait;
use pipx::{async_pipeline, AsyncNext, AsyncPipe, PipelineResult};

struct AsyncUppercase;

#[async_trait]
impl AsyncPipe<String> for AsyncUppercase {
    async fn handle(
        &self,
        passable: String,
        next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        // Modify the value asynchronously before continuing.
        next.handle(passable.to_uppercase()).await
    }
}

#[tokio::main]
async fn main() -> pipx::PipelineResult<()> {
    let output = async_pipeline("hello".to_string())
        .pipe(AsyncUppercase)
        .then_return()
        .await?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Async Pipeline with `then_return`

Use then_return when you want the asynchronous pipeline to return the final processed value.

use async_trait::async_trait;
use pipx::{async_pipeline, AsyncNext, AsyncPipe, PipelineResult};

struct AsyncTrim;

#[async_trait]
impl AsyncPipe<String> for AsyncTrim {
    async fn handle(
        &self,
        passable: String,
        next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.trim().to_string()).await
    }
}

#[tokio::main]
async fn main() -> pipx::PipelineResult<()> {
    let output = async_pipeline("  hello  ".to_string())
        .pipe(AsyncTrim)
        .then_return()
        .await?;

    assert_eq!(output, "hello");

    Ok(())
}

Async Pipeline with `then`

Use then when you want to execute an asynchronous pipeline and transform the final value.

use async_trait::async_trait;
use pipx::{async_pipeline, AsyncNext, AsyncPipe, PipelineResult};

struct AsyncTrim;

#[async_trait]
impl AsyncPipe<String> for AsyncTrim {
    async fn handle(
        &self,
        passable: String,
        next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.trim().to_string()).await
    }
}

#[tokio::main]
async fn main() -> pipx::PipelineResult<()> {
    let length = async_pipeline("  hello  ".to_string())
        .pipe(AsyncTrim)
        // Execute the async pipeline, then map the final value into another type.
        .then(|value| value.len())
        .await?;

    assert_eq!(length, 5);

    Ok(())
}

Async Pipeline with `through`

Use through with async_steps! when you want to set multiple asynchronous steps at once.

use async_trait::async_trait;
use pipx::{async_pipeline, async_steps, AsyncNext, AsyncPipe, PipelineResult};

struct AsyncTrim;
struct AsyncUppercase;

#[async_trait]
impl AsyncPipe<String> for AsyncTrim {
    async fn handle(
        &self,
        passable: String,
        next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.trim().to_string()).await
    }
}

#[async_trait]
impl AsyncPipe<String> for AsyncUppercase {
    async fn handle(
        &self,
        passable: String,
        next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase()).await
    }
}

#[tokio::main]
async fn main() -> pipx::PipelineResult<()> {
    let output = async_pipeline("  hello  ".to_string())
        // Replace the current async pipeline steps with the provided step list.
        .through(async_steps![AsyncTrim, AsyncUppercase])
        .then_return()
        .await?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Async Pipeline with `when`

Use when when you want to append an asynchronous step only if a condition is true.

use async_trait::async_trait;
use pipx::{async_pipeline, AsyncNext, AsyncPipe, PipelineResult};

struct AsyncUppercase;

#[async_trait]
impl AsyncPipe<String> for AsyncUppercase {
    async fn handle(
        &self,
        passable: String,
        next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase()).await
    }
}

#[tokio::main]
async fn main() -> pipx::PipelineResult<()> {
    let should_uppercase = true;

    let output = async_pipeline("hello".to_string())
        .when(should_uppercase, AsyncUppercase)
        .then_return()
        .await?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Async Pipeline with `unless`

Use unless when you want to append an asynchronous step only if a condition is false.

use async_trait::async_trait;
use pipx::{async_pipeline, AsyncNext, AsyncPipe, PipelineResult};

struct AsyncUppercase;

#[async_trait]
impl AsyncPipe<String> for AsyncUppercase {
    async fn handle(
        &self,
        passable: String,
        next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase()).await
    }
}

#[tokio::main]
async fn main() -> pipx::PipelineResult<()> {
    let skip_uppercase = false;

    let output = async_pipeline("hello".to_string())
        .unless(skip_uppercase, AsyncUppercase)
        .then_return()
        .await?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Async Pipeline with `finally`

Use finally when you want to run a callback after asynchronous pipeline execution completes.

use async_trait::async_trait;
use pipx::{async_pipeline, AsyncNext, AsyncPipe, PipelineResult};

struct AsyncUppercase;

#[async_trait]
impl AsyncPipe<String> for AsyncUppercase {
    async fn handle(
        &self,
        passable: String,
        next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase()).await
    }
}

#[tokio::main]
async fn main() -> PipelineResult<()> {
    let output = async_pipeline("hello".to_string())
        .pipe(AsyncUppercase)
        .finally(|result| {
            println!("Async pipeline finished: {result:?}");
        })
        .then_return()
        .await?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Async Pipeline with `rescue`

Use rescue when you want to recover from asynchronous pipeline errors with a fallback value.

use async_trait::async_trait;
use pipx::{async_pipeline, AsyncNext, AsyncPipe, PipelineError, PipelineResult};

struct AsyncFail;

#[async_trait]
impl AsyncPipe<String> for AsyncFail {
    async fn handle(
        &self,
        _passable: String,
        _next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        Err(PipelineError::Message("async pipeline failed".to_string()))
    }
}

#[tokio::main]
async fn main() -> PipelineResult<()> {
    let output = async_pipeline("hello".to_string())
        .pipe(AsyncFail)
        .rescue(|_| "fallback".to_string())
        .await?;

    assert_eq!(output, "fallback");

    Ok(())
}

Advanced

The examples below focus on composition, short-circuiting, wrapping downstream results, custom errors, shared step collections, and asynchronous job-style workflows.

Wrapping Downstream Results

A pipe can call next.handle first and then modify the value returned by the remaining pipeline.

This is useful for response wrapping, instrumentation, logging, tracing, or post-processing.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Wrap;

impl Pipe<String> for Wrap {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        // Continue the rest of the pipeline first.
        let result = next.handle(passable)?;

        // Wrap the downstream result after all following steps have completed.
        Ok(format!("[{result}]"))
    }
}

struct Uppercase;

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase())
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("hello".to_string())
        .pipe(Wrap)
        .pipe(Uppercase)
        .then_return()?;

    assert_eq!(output, "[HELLO]");

    Ok(())
}

Short-Circuiting Execution

A pipe can stop execution by returning a value without calling next.handle.

This is useful for cache hits, authorization failures, validation exits, or fallback behavior.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Stop;

impl Pipe<String> for Stop {
    fn handle(
        &self,
        passable: String,
        _next: Next<'_, String>,
    ) -> PipelineResult<String> {
        // Do not continue to the remaining steps.
        Ok(format!("{passable}:stopped"))
    }
}

struct NeverRuns;

impl Pipe<String> for NeverRuns {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(format!("never:{passable}"))
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("job".to_string())
        .pipe(Stop)
        .pipe(NeverRuns)
        .then_return()?;

    assert_eq!(output, "job:stopped");

    Ok(())
}

Validation Pipeline

Pipelines work well for validation flows because each pipe can either continue with the value or stop with an error.

use pipx::{pipeline, Next, Pipe, PipelineError, PipelineResult};

struct UserInput {
    username: String,
    password: String,
}

struct ValidateUsername;

impl Pipe<UserInput> for ValidateUsername {
    fn handle(
        &self,
        passable: UserInput,
        next: Next<'_, UserInput>,
    ) -> PipelineResult<UserInput> {
        if passable.username.trim().is_empty() {
            return Err(PipelineError::Message("username is required".to_string()));
        }

        next.handle(passable)
    }
}

struct ValidatePassword;

impl Pipe<UserInput> for ValidatePassword {
    fn handle(
        &self,
        passable: UserInput,
        next: Next<'_, UserInput>,
    ) -> PipelineResult<UserInput> {
        if passable.password.len() < 8 {
            return Err(PipelineError::Message("password is too short".to_string()));
        }

        next.handle(passable)
    }
}

fn main() -> pipx::PipelineResult<()> {
    let input = UserInput {
        username: "selcuk".to_string(),
        password: "secret-password".to_string(),
    };

    let validated = pipeline(input)
        .pipe(ValidateUsername)
        .pipe(ValidatePassword)
        .then_return()?;

    assert_eq!(validated.username, "selcuk");

    Ok(())
}

Shared Step Collections

Use steps! when you want to define reusable groups of pipeline steps.

use pipx::{pipeline, steps, Next, Pipe, PipelineResult};

struct Trim;
struct Uppercase;
struct Prefix(&'static str);

impl Pipe<String> for Trim {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.trim().to_string())
    }
}

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase())
    }
}

impl Pipe<String> for Prefix {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(format!("{}{}", self.0, passable))
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("  admin  ".to_string())
        .through(steps![
            Trim,
            Uppercase,
            Prefix("USER:"),
        ])
        .then_return()?;

    assert_eq!(output, "USER:ADMIN");

    Ok(())
}

Conditional Runtime Composition

Use when and unless when the pipeline should be assembled based on runtime state.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Trim;
struct DebugPrefix;

impl Pipe<String> for Trim {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.trim().to_string())
    }
}

impl Pipe<String> for DebugPrefix {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(format!("debug:{passable}"))
    }
}

fn main() -> pipx::PipelineResult<()> {
    let debug_enabled = true;
    let skip_trim = false;

    let output = pipeline("  request  ".to_string())
        .unless(skip_trim, Trim)
        .when(debug_enabled, DebugPrefix)
        .then_return()?;

    assert_eq!(output, "debug:request");

    Ok(())
}

Error Recovery

Use rescue when pipeline errors should be converted into a fallback value.

InputMissing is not recovered because it represents an invalid pipeline configuration rather than a step execution failure.

use pipx::{pipeline, Next, Pipe, PipelineError, PipelineResult};

struct FailingStep;

impl Pipe<String> for FailingStep {
    fn handle(
        &self,
        _passable: String,
        _next: Next<'_, String>,
    ) -> PipelineResult<String> {
        Err(PipelineError::Message("step failed".to_string()))
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("hello".to_string())
        .pipe(FailingStep)
        .rescue(|error| {
            format!("fallback because: {error}")
        })?;

    assert!(output.starts_with("fallback because:"));

    Ok(())
}

Finalization Hooks

Use finally when you need to observe the final result without changing it.

This is useful for logging, metrics, tracing, cleanup, or debugging.

use pipx::{pipeline, Next, Pipe, PipelineResult};

struct Uppercase;

impl Pipe<String> for Uppercase {
    fn handle(
        &self,
        passable: String,
        next: Next<'_, String>,
    ) -> PipelineResult<String> {
        next.handle(passable.to_uppercase())
    }
}

fn main() -> pipx::PipelineResult<()> {
    let output = pipeline("hello".to_string())
        .pipe(Uppercase)
        .finally(|result| {
            println!("Final pipeline result: {result:?}");
        })
        .then_return()?;

    assert_eq!(output, "HELLO");

    Ok(())
}

Async Background Job Pipeline

Asynchronous pipelines are useful for job processing, queue workflows, I/O-heavy steps, and other async workloads.

use async_trait::async_trait;
use pipx::{async_pipeline, async_steps, AsyncNext, AsyncPipe, PipelineResult};

#[derive(Debug)]
struct Job {
    id: u64,
    attempts: u8,
    events: Vec<String>,
}

struct LoadFromQueue;
struct ExecuteJob;

#[async_trait]
impl AsyncPipe<Job> for LoadFromQueue {
    async fn handle(
        &self,
        mut passable: Job,
        next: AsyncNext<'_, Job>,
    ) -> PipelineResult<Job> {
        passable.events.push("queue:loaded".to_string());

        next.handle(passable).await
    }
}

#[async_trait]
impl AsyncPipe<Job> for ExecuteJob {
    async fn handle(
        &self,
        mut passable: Job,
        next: AsyncNext<'_, Job>,
    ) -> PipelineResult<Job> {
        passable.attempts += 1;
        passable.events.push("job:executed".to_string());

        next.handle(passable).await
    }
}

#[tokio::main]
async fn main() -> pipx::PipelineResult<()> {
    let job = Job {
        id: 100,
        attempts: 0,
        events: Vec::new(),
    };

    let job = async_pipeline(job)
        .through(async_steps![
            LoadFromQueue,
            ExecuteJob,
        ])
        .then_return()
        .await?;

    assert_eq!(job.attempts, 1);
    assert_eq!(job.events, vec!["queue:loaded", "job:executed"]);

    Ok(())
}

Async Error Recovery

Asynchronous pipelines can also recover from execution errors with rescue.

use async_trait::async_trait;
use pipx::{async_pipeline, AsyncNext, AsyncPipe, PipelineError, PipelineResult};

struct AsyncFail;

#[async_trait]
impl AsyncPipe<String> for AsyncFail {
    async fn handle(
        &self,
        _passable: String,
        _next: AsyncNext<'_, String>,
    ) -> PipelineResult<String> {
        Err(PipelineError::Message("async step failed".to_string()))
    }
}

#[tokio::main]
async fn main() -> pipx::PipelineResult<()> {
    let output = async_pipeline("job".to_string())
        .pipe(AsyncFail)
        .rescue(|error| {
            format!("recovered from: {error}")
        })
        .await?;

    assert!(output.starts_with("recovered from:"));

    Ok(())
}

Benchmark

The repository includes Criterion benchmarks for measuring pipeline execution performance under different execution models and workloads.

All Benchmarks

Runs every available benchmark suite.

cargo bench -p pipx

Pipeline Benchmarks

Measures execution throughput, step traversal, and short-circuit behavior for standard pipelines.

cargo bench -p pipx --bench pipeline

Transform Pipeline Benchmarks

Measures throughput for transform pipelines that process values sequentially without pipeline continuations.

cargo bench -p pipx --bench pipeline_transform

Async Pipeline Benchmarks

Measures execution throughput, step traversal, and short-circuit behavior for asynchronous pipelines.

cargo bench -p pipx --features async --bench async_pipeline

Async Transform Pipeline Benchmarks

Measures throughput for asynchronous transform pipelines operating on asynchronous workloads.

cargo bench -p pipx --features async --bench async_pipeline_transform

Examples

Runnable examples live under examples/*.

cargo run -p pipx --example basic_transform

cargo run -p pipx --example middleware_auth

cargo run -p pipx --example axum_adapter

cargo run -p pipx --example actix_web_adapter

cargo run -p pipx --example data_validation

cargo run -p pipx --example gpu_wgpu_pipeline

cargo run -p pipx --features async --example async_jobs

The web and GPU examples use framework-shaped adapter types instead of forcing heavy framework or GPU dependencies into the crate. They show how to place pipx around Axum-like handlers, Actix-like service requests, validation flows, async jobs, and wgpu-style render command pipelines.

Contributing

The pipx project welcomes contributions from the community.

Whether you want to report a bug, suggest a new feature, improve the documentation, or submit code changes, your contributions are greatly appreciated.

You can find detailed information about the contribution process by visiting the link below.

Contributing Guide

Security

The pipx project takes security vulnerabilities seriously.

If you believe you have discovered a security vulnerability, please report it responsibly by contacting Selçuk Çukur at hello@selcukcukur.me.

Please do not disclose security vulnerabilities publicly until they have been reviewed and addressed.

You can find detailed information about the security policy by visiting the link below.

Security Policy

License

The pipx project is published as open source software under the MIT License, which is one of the most widely used open source licenses.

You can find detailed information about the license terms by visiting the link below.

MIT License

Features

Installation

With default

Cargo

Manual

Async Feature

Cargo

Manual

Macros Feature

Cargo

Manual

Full feature

Cargo

Manual

Quick Start

Pipeline

Pipeline with then_return

Pipeline with then

Pipeline with through

Pipeline with when

Pipeline with unless

Pipeline with finally

Pipeline with rescue

Pipeline Short-Circuiting

Pipeline Wrapping

Async Pipeline

Async Pipeline with then_return

Async Pipeline with then

Async Pipeline with through

Async Pipeline with when

Async Pipeline with unless

Async Pipeline with finally

Async Pipeline with rescue

Advanced

Wrapping Downstream Results

Short-Circuiting Execution

Validation Pipeline

Shared Step Collections

Conditional Runtime Composition

Error Recovery

Finalization Hooks

Async Background Job Pipeline

Async Error Recovery

Benchmark

All Benchmarks

Pipeline Benchmarks

Transform Pipeline Benchmarks

Async Pipeline Benchmarks

Async Transform Pipeline Benchmarks

Examples

Contributing

Security

License

Pipeline with `then_return`

Pipeline with `then`

Pipeline with `through`

Pipeline with `when`

Pipeline with `unless`

Pipeline with `finally`

Pipeline with `rescue`

Async Pipeline with `then_return`

Async Pipeline with `then`

Async Pipeline with `through`

Async Pipeline with `when`

Async Pipeline with `unless`

Async Pipeline with `finally`

Async Pipeline with `rescue`