Pipeline

Procesa datos a través de una secuencia ordenada de pasos (stages) que transforman su entrada en salida.

Contexto

Una request, un mensaje o un dato debe atravesar varias transformaciones: parsear → validar → enriquecer → guardar → notificar. Quieres que cada paso sea independiente y reordenable.

Problema

Una función monolítica con 8 responsabilidades es difícil de testear.
Reordenar pasos implica reescribir el flujo.

Solución

Define una interfaz común para cada stage y compón el pipeline conectándolos. Cada stage hace una sola cosa y pasa el resultado al siguiente.

#Estructura

IPipelineStep<TIn, TOut> — contrato genérico.
Pipeline<T> — encadena pasos compatibles.
Cada paso es testeable en aislamiento.

#Ejemplo en C# — pipeline genérico

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)); }

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

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

#Ejemplo en C# — pipeline asíncrono (estilo middleware)

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();
    }
}

Es exactamente la idea del middleware de ASP.NET Core y de MediatR pipeline behaviors.

Cuándo NO aplicarlo

Cuando el flujo tiene mucho branching: Pipeline brilla cuando es lineal.
Cuando hay alto acoplamiento entre pasos (necesitan compartir estado interno).

Tradeoffs

Pro	Contra
Cada paso es SRP y testeable	Saltos en el código siguen el flujo
Reordenable y reutilizable	Tipos genéricos pueden volverse complejos
Encaja con paralelismo / streaming	Errores deben propagarse explícitamente

#Variantes

Pipes & Filters clásico (orientado a streams).
Middleware pipeline (con next() — estilo ASP.NET).
Channel-based pipeline (con System.Threading.Channels).

#extra #pipeline #data-flow