Nexus.Pipeline.Sample.MvcApp 1.0.3

Nexus.Pipeline

Typed steps. Structured results. Full execution history.

Nexus.Pipeline is a typed, in-process execution pipeline library for .NET. You register an ordered sequence of steps on a fluent builder, attach cross-cutting behaviors (logging, retry, timeout, transactions, OpenTelemetry metrics), and execute against a shared IExecutionContext that carries the request, the evolving response, and a per-step execution history. Every execution returns a PipelineResult<TResponse> — structured errors, no exception-based control flow. Supports inline lambda steps and DI-resolved class-based steps interchangeably. Targets .NET 4.6.2 through .NET 10, with AOT and trim compatibility on .NET 8+.

graph TB
    subgraph APP["Application Layer"]
        API["ASP.NET Core / Worker"]
        LEGACY["net462 Console / WCF / MVC 5"]
    end

    subgraph CORE["Core Engine"]
        BUILDER["PipelineBuilder / PipelineDefinitionBuilder"]
        EXEC["PipelineExecutor"]
        CTX["ExecutionContext + ContextScope"]
        COMP["CompensationEngine"]
    end

    subgraph BEH["Behaviors (middleware chain)"]
        LOG["LoggingBehavior (−1000)"]
        RETRY["RetryBehavior (−800)"]
        TIMEOUT["TimeoutBehavior (−600)"]
        EXC["ExceptionHandlingBehavior (MaxValue)"]
    end

    subgraph DI["DI / Hosting"]
        SVC["AddNexusPipeline()"]
        RES["IStepResolver (DI-backed)"]
    end

    APP --> DI
    DI --> CORE
    CORE --> BEH
    CORE --> COMP

Feature Highlights

Packages

Package	NuGet	TFMs	Description
Nexus.Pipeline		net462 · ns2.0 · net6–10	Execution engine, behaviors, DI integration, and visualization

Diagram export (Mermaid, PlantUML, JSON) targets netstandard2.0 and .NET 6+ (not net462).

Installation

dotnet add package Nexus.Pipeline

# Package Manager Console
Install-Package Nexus.Pipeline

Quick Start

Pattern 1 — Pipeline.For<>(): Build-and-Execute

The fastest way to run a pipeline. Each call to .ExecuteAsync() is a fresh execution.

using Nexus.Pipeline;
using Nexus.Pipeline.Behaviors;
using Nexus.Pipeline.DependencyInjection.Extensions;

// DI wiring (once at startup)
services.AddNexusPipeline();
services.AddPipelineStep<ValidatePaymentStep>();
services.AddPipelineStep<CheckBalanceStep>();
services.AddPipelineStep<CreateTransactionStep>();
services.AddTransient(typeof(LoggingBehavior<,>));
services.AddTransient(typeof(ExceptionHandlingBehavior<,>));

// Execution (per-request)
var result = await Pipeline
    .For<PaymentRequest, PaymentResponse>(resolver)
    .WithName("PaymentPipeline")
    .AddBehavior<LoggingBehavior<PaymentRequest, PaymentResponse>>()
    .AddBehavior<ExceptionHandlingBehavior<PaymentRequest, PaymentResponse>>()
    .Validation<ValidatePaymentStep>()
    .Step<CheckBalanceStep>()
    .Step<CreateTransactionStep>()
    .ExecuteAsync(request, cancellationToken);

// Result
result.IsSuccess          // bool
result.Response           // PaymentResponse?
result.Errors             // IReadOnlyList<ErrorInfo>
result.ExecutionHistory   // per-step records
result.Metadata.Duration  // TimeSpan?

Pattern 2 — Pipeline.Define<>(): Reusable Definition

Build once at startup; call ExecuteAsync many times. Each call gets its own CorrelationId and ExecutionContext — fully thread-safe.

// Built ONCE (e.g. in a singleton or static field)
IPipelineDefinition<PaymentRequest, PaymentResponse> _pipeline =
    Pipeline.Define<PaymentRequest, PaymentResponse>(resolver)
        .WithName("PaymentPipeline")
        .WithVersion("2.0")
        .WithLoggerFactory(loggerFactory)
        .WithCorrelationIdFactory(() => "PAY-" + Guid.NewGuid().ToString("N")[..8])
        .AddBehavior<LoggingBehavior<PaymentRequest, PaymentResponse>>()
        .WithPolicy(p => p.Retry(3).Timeout(TimeSpan.FromSeconds(30)))
        .Validation<ValidatePaymentStep>()
        .Step<CheckBalanceStep>()
        .Step<CreateTransactionStep>()
        .Build();

// Executed MANY times (concurrent-safe)
var result = await _pipeline.ExecuteAsync(request, cancellationToken);

Pattern 3 — Implementing a Step

public sealed class CheckBalanceStep : IPipelineStep<PaymentRequest, PaymentResponse>
{
    private readonly IAccountRepository _accounts;

    public CheckBalanceStep(IAccountRepository accounts) => _accounts = accounts;

    public string StepName => "CheckBalance";
    public int Order => 2;

    public async Task<StepOutcome> ExecuteAsync(
        IExecutionContext<PaymentRequest, PaymentResponse> context,
        CancellationToken cancellationToken = default)
    {
        var balance = await _accounts.GetBalanceAsync(context.Request.CustomerId, cancellationToken);

        if (context.Request.Amount > balance)
            return StepOutcome.Fail;

        context.Items.Set("available_balance", balance);
        return StepOutcome.Continue;
    }
}

Advanced Patterns

Saga Compensation (Rollback)

Compensation steps are registered per-step and run in reverse order on pipeline failure.

await Pipeline
    .For<TransferRequest, TransferResponse>(resolver)
    .Validation<ValidateTransferStep>()
    .Step<ReserveAmountStep>()
    .WithCompensation<ReleaseReservationStep>()   // runs if pipeline fails AFTER this step
    .Step<ExecuteTransferStep>()
    .WithCompensation<ReverseTransferStep>()       // runs first (reverse order)
    .Step<AuditTransferStep>()
    .ExecuteAsync(request, ct);

If AuditTransferStep fails → ReverseTransferStep runs, then ReleaseReservationStep.

Conditional Routing

Pipeline.Define<ApprovalRequest, ApprovalResponse>(resolver)
    .Step<CheckAutoApprovalStep>()   // sets ctx.Scope["auto_approved"] = true/false
    .If(ctx => ctx.Items.TryGet<bool>("auto_approved", out var v) && v,
        b => b.Step<AutoApproveStep>())
    .If(ctx => ctx.Items.TryGet<bool>("manager_required", out var v) && v,
        b => b.Step<RequestManagerApprovalStep>())
    .Build();

Parallel Branches

pipeline
    .Step<ProcessOrderStep>()
    .Parallel(
        b => b.Step<NotifyApplicantStep>(),     // concurrent
        b => b.Step<NotifySlackChannelStep>())   // concurrent
    .Step<FinalizeStep>()

DAG (Dependency Graph)

pipeline
    .Step<StepA>()
    .Step<StepB>().DependsOn<StepA>()    // B runs after A
    .Step<StepC>().DependsOn<StepA>()    // C runs after A (parallel with B)
    .Step<StepD>().DependsOn<StepB>().DependsOn<StepC>()  // D waits for B and C

Validation Step

public sealed class ValidatePaymentStep : IValidationStep<PaymentRequest, PaymentResponse>
{
    public string StepName => "ValidatePayment";
    public int Order => 0;

    public Task<ValidationResult> ValidateAsync(
        IExecutionContext<PaymentRequest, PaymentResponse> context,
        CancellationToken cancellationToken = default)
    {
        var errors = new List<ErrorInfo>();

        if (context.Request.Amount <= 0)
            errors.Add(ErrorInfo.Validation("PAY-VAL-001", "Amount must be positive", nameof(PaymentRequest.Amount)));
        if (string.IsNullOrEmpty(context.Request.CustomerId))
            errors.Add(ErrorInfo.Validation("PAY-VAL-002", "CustomerId is required", nameof(PaymentRequest.CustomerId)));

        return Task.FromResult(
            errors.Count == 0 ? ValidationResult.Valid() : ValidationResult.Invalid(errors.ToArray()));
    }

    public Task<StepOutcome> ExecuteAsync(
        IExecutionContext<PaymentRequest, PaymentResponse> context,
        CancellationToken cancellationToken = default)
        => Task.FromResult(StepOutcome.Continue);
}

Pipeline Visualization

// After execution, export a Mermaid diagram of what ran
var mermaid = result.Metadata.ToMermaid();
Console.WriteLine(mermaid);

// Output:
// graph TD
//   ValidatePayment["ValidatePayment ✓ 2ms"]
//   CheckBalance["CheckBalance ✓ 5ms"]
//   CreateTransaction["CreateTransaction ✓ 12ms"]
//   ValidatePayment --> CheckBalance --> CreateTransaction

Step Outcomes

Built-in Behaviors

Lower Order = outermost (wraps everything inside it). ExceptionHandlingBehavior should always be outermost in practice, so register it last or use int.MaxValue.

WithPolicy() Shortcut

pipeline.WithPolicy(p => p
    .Retry(3, delay: TimeSpan.FromMilliseconds(200), backoffMultiplier: 2.0)
    .Timeout(TimeSpan.FromSeconds(30)));

Execution Context API

Every step receives IExecutionContext<TRequest, TResponse>:

ctx.Request              // TRequest — read-only
ctx.Response             // TResponse? — set by steps
ctx.CorrelationId        // string — unique per pipeline run
ctx.CancellationToken    // propagated cancellation

// Cross-step state bag (typed, dictionary-backed)
ctx.Items.Set("key", value);
ctx.Items.TryGet<T>("key", out var v);
ctx.Items.Remove("key");

// Execution metadata
ctx.ExecutionHistory     // IReadOnlyList<StepExecutionRecord>
ctx.PipelineMetadata     // name, version, tags

Error Model

result.Errors  // IReadOnlyList<ErrorInfo>

// ErrorInfo properties:
error.Code          // "PAY-001"
error.Message       // human-readable
error.PropertyName  // field that caused the error (validation)
error.ErrorType     // Business | Technical | Unknown
error.Category      // Validation | Authorization | NotFound | ...
error.Severity      // Error | Warning | Critical | Info
error.Exception     // Exception? (if from ExceptionHandlingBehavior)

// Pattern matching on result:
var msg = result.Match(
    onSuccess:   r  => $"OK: {r.TransactionId}",
    onFailure:   e  => $"Fail: {e[0].Code}",
    onException: ex => $"Ex: {ex.Message}");

// Fluent callbacks:
result
    .OnSuccess(r => log.LogInformation("Done: {TxnId}", r.TransactionId))
    .OnFailure(errors => log.LogWarning("Failed: {Code}", errors[0].Code))
    .OnException(ex => log.LogError(ex, "Unhandled exception"));

.NET Framework 4.6 Compatibility

All packages ship a net462 target. Add these binding redirects to app.config:

<runtime>
  <assemblyBinding xmlns="urn:schemas-microsoft-com:asm.v1">
    <dependentAssembly>
      <assemblyIdentity name="Microsoft.Extensions.DependencyInjection.Abstractions"
                        publicKeyToken="adb9793829ddae60" culture="neutral" />
      <bindingRedirect oldVersion="0.0.0.0-8.0.0.0" newVersion="8.0.0.0" />
    </dependentAssembly>
    <dependentAssembly>
      <assemblyIdentity name="Microsoft.Extensions.Logging.Abstractions"
                        publicKeyToken="adb9793829ddae60" culture="neutral" />
      <bindingRedirect oldVersion="0.0.0.0-8.0.0.0" newVersion="8.0.0.0" />
    </dependentAssembly>
    
  </assemblyBinding>
</runtime>

Entry point on net462 (no top-level statements):

static void Main(string[] args)
{
    RunAsync().GetAwaiter().GetResult();
}

static async Task RunAsync()
{
    var result = await Pipeline
        .For<OrderRequest, OrderResponse>(resolver)
        .Validation<ValidateOrderStep>()
        .Step<CheckInventoryStep>()
        .Step<ReserveInventoryStep>()
        .WithCompensation<CancelReservationStep>()
        .Step<CreateOrderStep>()
        .ExecuteAsync(request);
}

See docs/samples/net-framework-46.md for ASP.NET MVC 5, WCF, and Autofac integration patterns.

Samples

Documentation

Architecture Diagrams

Mermaid (.mmd) and PlantUML (.puml) diagrams in docs/diagrams/:

License

MIT — see LICENSE for details.