Pipeline

Processes data through an ordered sequence of steps (stages) that transform its input into output.

Context

A request, message, or data must go through several transformations: parse → validate → enrich → save → notify. You want each step to be independent and reorderable.

Problem

A monolithic function with 8 responsibilities is difficult to test.
Reordering steps implies rewriting the flow.

Solution

Define a common interface for each stage and compose the pipeline by connecting them. Each stage does one thing and passes the result to the next one.

#Structure

IPipelineStep<TIn, TOut> — generic contract.
Pipeline<T> — chains compatible steps.
Each step is testable in isolation.

#C# Example — generic pipeline

public interface IPipelineStep<in TIn, out TOut>
{
    TOut Process(TIn input);
}

public class Pipeline<TIn, TOut>
{
    private readonly Func<TIn, TOut> _run;
    private Pipeline(Func<TIn, TOut> run) => _run = run;

    public static Pipeline<T, T> Start<T>() => new(x => x);

    public Pipeline<TIn, TNext> Then<TNext>(IPipelineStep<TOut, TNext> next) =>
        new(x => next.Process(_run(x)));

    public TOut Run(TIn input) => _run(input);
}

// Stages
public class Trim     : IPipelineStep<string, string>      { public string Process(string s)   => s.Trim(); }
public class ToUpper  : IPipelineStep<string, string>      { public string Process(string s)   => s.ToUpperInvariant(); }
public class Hash     : IPipelineStep<string, byte[]>      { public byte[] Process(string s)   => System.Security.Cryptography.SHA256.HashData(System.Text.Encoding.UTF8.GetBytes(s)); }

// Composition
var pipe = Pipeline<string, string>.Start<string>()
    .Then(new Trim())
    .Then(new ToUpper())
    .Then(new Hash());

byte[] digest = pipe.Run("  hola  ");

#C# Example — asynchronous pipeline (middleware style)

public delegate Task PipelineDelegate<TContext>(TContext ctx, Func<Task> next);

public class AsyncPipeline<TContext>
{
    private readonly List<PipelineDelegate<TContext>> _steps = new();
    public AsyncPipeline<TContext> Use(PipelineDelegate<TContext> step) { _steps.Add(step); return this; }

    public Task RunAsync(TContext ctx)
    {
        Func<Task> next = () => Task.CompletedTask;
        for (int i = _steps.Count - 1; i >= 0; i--)
        {
            var step = _steps[i];
            var current = next;
            next = () => step(ctx, current);
        }
        return next();
    }
}

This is exactly the idea of ASP.NET Core middleware and MediatR pipeline behaviors.

When NOT to use it

When the flow has a lot of branching: Pipeline shines when it is linear.
When there is high coupling between steps (they need to share internal state).

Tradeoffs

Pro	Con
Each step is SRP and testable	Jumps in the code follow the flow
Reorderable and reusable	Generic types can become complex
Fits with parallelism / streaming	Errors must be propagated explicitly

#Variants

Classic Pipes & Filters (stream-oriented).
Middleware pipeline (with next() — ASP.NET style).
Channel-based pipeline (with System.Threading.Channels).

#extra #pipeline #data-flow