Mab.PipelineFramework.Core 1.0.2

Pipeline Framework

Event-based pipeline framework with pluggable modules and middleware support for .NET. Perfect for multi-tenant SaaS applications with subscription-based feature access.

Features

• 🚀 Source Generator - 27.4x Faster! - Optional compile-time code generation (verified: 1.542ns vs 42.194ns)
• Event-Based Architecture - Define fixed event flows with pluggable module implementations
• Middleware Pipeline - ASP.NET Core-style middleware for cross-cutting concerns
• Production-Ready Middleware Templates - Logging, Retry, Timeout, Validation, Exception Handling, Circuit Breaker
• Testing Framework - Fluent API and event spy for comprehensive pipeline testing
• Mixed Sync/Async - Support both synchronous and asynchronous modules in same pipeline
• Dependency Injection - First-class support for constructor dependency injection
• Multi-Tenant Ready - Built-in patterns for subscription-based feature access
• Type-Safe - Generic base classes for compile-time type checking
• External DLL Support - Load modules from external assemblies

Installation

dotnet add package PipelineFramework.Core
dotnet add package PipelineFramework.Testing  # For testing utilities

Or via Package Manager:

Install-Package PipelineFramework.Core
Install-Package PipelineFramework.Testing  # For testing utilities

Quick Start

1. Define Your Events

public class OrderPipelineEvents
{
    public Func<OrderContext, Task>? ValidateOrder { get; set; }
    public Func<OrderContext, Task>? ProcessPayment { get; set; }
    public Func<OrderContext, Task>? ShipOrder { get; set; }
}

2. Create Modules

public class ValidationModule : PipelineModule<OrderPipelineEvents>
{
    public override void Initialize(OrderPipelineEvents events,
        IReadOnlyDictionary<string, string>? parameters)
    {
        events.ValidateOrder += async (context) =>
        {
            if (context.Order.Total <= 0)
                throw new InvalidOperationException("Invalid order total");

            context.IsValid = true;
            await Task.CompletedTask;
        };
    }
}

3. Register and Execute

// Register modules
services.AddPipelineModule<OrderPipelineEvents, ValidationModule>()
        .AddPipelineModule<OrderPipelineEvents, PaymentModule>()
        .AddPipelineModule<OrderPipelineEvents, ShippingModule>();

// Initialize pipeline
var events = Backbone<OrderPipelineEvents>
    .InitializePipeline("OrderProcessing", serviceProvider);

// Execute
var context = new OrderContext { Order = order, Customer = customer };
await events.ValidateOrder?.Invoke(context)!;
await events.ProcessPayment?.Invoke(context)!;
await events.ShipOrder?.Invoke(context)!;

Multi-Tenant SaaS Example

Perfect for subscription-based applications with different feature tiers:

Free Tier

services.AddPipelineModule<InvoiceEvents, BasicValidationModule>()
        .AddPipelineModule<InvoiceEvents, StandardTaxModule>()
        .AddPipelineModule<InvoiceEvents, BasicPDFModule>();

Premium Tier

services.AddPipelineModule<InvoiceEvents, AdvancedValidationModule>()
        .AddPipelineModule<InvoiceEvents, MultiCurrencyTaxModule>()
        .AddPipelineModule<InvoiceEvents, CustomBrandedPDFModule>()
        .AddPipelineModule<InvoiceEvents, AutomaticRemindersModule>();

Enterprise Tier

services.AddPipelineModule<InvoiceEvents, EnterpriseValidationModule>()
        .AddPipelineModule<InvoiceEvents, RegionalTaxModule>() // EU VAT, US Sales Tax
        .AddPipelineModule<InvoiceEvents, WorkflowApprovalsModule>()
        .AddPipelineModule<InvoiceEvents, QuickBooksIntegrationModule>()
        .AddPipelineModule<InvoiceEvents, WebhookNotificationsModule>();

Middleware Support

Add cross-cutting concerns with middleware:

public class TenantIsolationMiddleware : IPipelineMiddleware<InvoiceContext>
{
    public async Task InvokeAsync(InvoiceContext context, Func<Task> next)
    {
        // Before pipeline
        if (!context.Properties.ContainsKey("TenantId"))
            throw new InvalidOperationException("Tenant context required");

        context.Properties["TenantVerified"] = "true";

        // Execute pipeline
        await next();

        // After pipeline
        // Cleanup, logging, etc.
    }
}

// Use middleware
var pipeline = serviceProvider.CreateMiddlewarePipeline<InvoiceContext>()
    .Use<TenantIsolationMiddleware>()
    .Use<AuditMiddleware>()
    .Use<LoggingMiddleware>();

await pipeline.ExecuteAsync(context);

Production-Ready Middleware Templates

The framework includes production-ready middleware templates for common cross-cutting concerns:

Logging Middleware

using Pipeline.Middleware.Templates;

var loggingOptions = new LoggingMiddlewareOptions
{
    LogBeforeExecution = true,
    LogAfterExecution = true,
    LogProperties = true, // Log context properties
    BeforeLogLevel = LogLevel.Information
};

pipeline.Use<LoggingMiddleware<InvoiceContext>>(sp =>
    new LoggingMiddleware<InvoiceContext>(sp.GetRequiredService<ILogger<LoggingMiddleware<InvoiceContext>>>(), loggingOptions));

Retry Middleware with Exponential Backoff

var retryOptions = new RetryMiddlewareOptions
{
    MaxRetries = 3,
    InitialDelayMs = 100,
    BackoffStrategy = RetryBackoffStrategy.Exponential, // Constant, Linear, or Exponential
    UseJitter = true, // Prevent thundering herd
    JitterFactor = 0.3 // 30% jitter
};

pipeline.Use<RetryMiddleware<InvoiceContext>>(sp =>
    new RetryMiddleware<InvoiceContext>(sp.GetRequiredService<ILogger<RetryMiddleware<InvoiceContext>>>(), retryOptions));

Timeout Middleware

var timeoutOptions = new TimeoutMiddlewareOptions
{
    Timeout = TimeSpan.FromSeconds(30),
    ThrowOnTimeout = true,
    LogTimeout = true
};

pipeline.Use<TimeoutMiddleware<InvoiceContext>>(sp =>
    new TimeoutMiddleware<InvoiceContext>(sp.GetRequiredService<ILogger<TimeoutMiddleware<InvoiceContext>>>(), timeoutOptions));

Validation Middleware

var validationOptions = new ValidationMiddlewareOptions
{
    UseDataAnnotations = true,
    RequiredProperties = new List<string> { "TenantId", "UserId" },
    CustomValidators = new List<Func<PipelineContext, PipelineValidationResult>>
    {
        ctx => ctx.Properties.ContainsKey("Amount") && (decimal)ctx.Properties["Amount"] > 0
            ? PipelineValidationResult.Success()
            : PipelineValidationResult.Failure("Amount must be positive")
    },
    ThrowOnValidationFailure = false,
    ContinueOnValidationFailure = false // Stop pipeline if validation fails
};

pipeline.Use<ValidationMiddleware<InvoiceContext>>(sp =>
    new ValidationMiddleware<InvoiceContext>(sp.GetRequiredService<ILogger<ValidationMiddleware<InvoiceContext>>>(), validationOptions));

Exception Handling Middleware

var exceptionOptions = new ExceptionHandlingMiddlewareOptions
{
    SuppressAllExceptions = true,
    SuppressCancellations = true,
    SuppressTimeouts = false,
    OnException = (context, ex) =>
    {
        // Custom error handling
        context.Properties["ErrorHandled"] = true;
        // Log to external service, send alert, etc.
    }
};

pipeline.Use<ExceptionHandlingMiddleware<InvoiceContext>>(sp =>
    new ExceptionHandlingMiddleware<InvoiceContext>(sp.GetRequiredService<ILogger<ExceptionHandlingMiddleware<InvoiceContext>>>(), exceptionOptions));

Circuit Breaker Middleware

var circuitBreakerOptions = new CircuitBreakerMiddlewareOptions
{
    FailureThreshold = 5, // Open circuit after 5 failures
    OpenDuration = TimeSpan.FromSeconds(60), // Keep open for 60s
    ThrowWhenOpen = true,
    LogStateTransitions = true
};

pipeline.Use<CircuitBreakerMiddleware<InvoiceContext>>(sp =>
    new CircuitBreakerMiddleware<InvoiceContext>(sp.GetRequiredService<ILogger<CircuitBreakerMiddleware<InvoiceContext>>>(), circuitBreakerOptions));

Combining Multiple Middleware

var pipeline = serviceProvider.CreateMiddlewarePipeline<InvoiceContext>()
    .Use<ExceptionHandlingMiddleware<InvoiceContext>>()  // Catch all exceptions
    .Use<LoggingMiddleware<InvoiceContext>>()            // Log execution
    .Use<ValidationMiddleware<InvoiceContext>>()         // Validate context
    .Use<TimeoutMiddleware<InvoiceContext>>()            // Enforce timeout
    .Use<CircuitBreakerMiddleware<InvoiceContext>>()     // Prevent cascading failures
    .Use<RetryMiddleware<InvoiceContext>>();             // Retry on failure

await pipeline.ExecuteAsync(context);

Testing Your Pipelines

The framework includes a comprehensive testing package with fluent API and event spy capabilities:

Quick Testing Example

using PipelineFramework.Testing;

[Fact]
public void OrderPipeline_ValidatesSuccessfully()
{
    // Arrange - Build test pipeline with spy
    var testPipeline = PipelineTestBuilder<OrderPipelineEvents>.For()
        .WithName("TestOrderPipeline")
        .WithModule<ValidationModule>()
        .WithSpy()  // Enable event tracking
        .Build();

    // Act - Execute pipeline
    var context = new OrderContext { Order = validOrder };
    testPipeline.Events.ValidateOrder?.Invoke(context);

    // Assert - Verify event execution
    testPipeline.ShouldHaveExecuted("ValidateOrder");
    context.IsValid.Should().BeTrue();
}

Advanced Testing Features

Event Spy for Tracking Invocations:

var testPipeline = PipelineTestBuilder<InvoiceEvents>.For()
    .WithModule<TaxCalculationModule>()
    .WithModule<PDFGenerationModule>()
    .WithSpy()
    .Build();

await testPipeline.Events.Calculate?.Invoke(context)!;
await testPipeline.Events.GeneratePDF?.Invoke(context)!;

// Verify specific events were called
testPipeline.ShouldHaveExecuted("Calculate");
testPipeline.ShouldHaveExecuted("GeneratePDF", expectedCount: 1);

// Check execution order
var invocations = testPipeline.Spy!.GetAllInvocations();
invocations[0].EventName.Should().Be("Calculate");
invocations[1].EventName.Should().Be("GeneratePDF");

Custom Service Configuration:

var testPipeline = PipelineTestBuilder<OrderPipelineEvents>.For()
    .WithModule<PaymentModule>()
    .ConfigureServices(services =>
    {
        // Add mock dependencies
        services.AddSingleton<IPaymentGateway, MockPaymentGateway>();
        services.AddSingleton<ILogger<PaymentModule>, NullLogger<PaymentModule>>();
    })
    .Build();

Testing Multi-Tenant Scenarios:

// Test Free Tier
var freePipeline = PipelineTestBuilder<InvoiceEvents>.For()
    .WithModule<BasicValidationModule>()
    .WithSpy()
    .Build();

freePipeline.Events.Validate?.Invoke(freeContext);
testPipeline.ShouldHaveExecuted("Validate");
testPipeline.ShouldNotHaveExecuted("ApplyAdvancedRules");

// Test Premium Tier
var premiumPipeline = PipelineTestBuilder<InvoiceEvents>.For()
    .WithModule<BasicValidationModule>()
    .WithModule<AdvancedValidationModule>()
    .WithSpy()
    .Build();

premiumPipeline.Events.Validate?.Invoke(premiumContext);
testPipeline.ShouldHaveExecuted("Validate");
testPipeline.ShouldHaveExecuted("ApplyAdvancedRules");

EventSpy API

The EventSpy tracks all event invocations with timestamps and context:

// Check if event was invoked
bool wasInvoked = spy.WasInvoked("EventName");

// Get invocation count
int count = spy.GetInvocationCount("EventName");

// Get specific invocations with context
var invocations = spy.GetInvocations("ValidateOrder");
foreach (var inv in invocations)
{
    Console.WriteLine($"{inv.EventName} at {inv.Timestamp}");
    var ctx = inv.Context as OrderContext;
}

// Get all invoked events
var eventNames = spy.GetInvokedEventNames();

// Clear tracking history
spy.Clear();

Performance Boost - Source Generator 🚀

NEW! Optional source generator for 27.4x faster execution:

// Add [GeneratePipeline] attribute and make class partial
[GeneratePipeline]
public partial class OrderPipelineEvents : PipelineEvents
{
    [PipelineEvent(Order = 1)]
    public PipelineContextAsync<OrderContext>? ValidateOrder { get; set; }

    [PipelineEvent(Order = 2, TimeoutMilliseconds = 5000)]
    public PipelineContextAsync<OrderContext>? ProcessPayment { get; set; }
}

// Generated ExecuteAllAsync method - NO REFLECTION! ⚡
await events.ExecuteAllAsync(context);

Verified Performance Benefits (BenchmarkDotNet on .NET 8.0):

• ⚡ 27.4x faster execution (1.542ns vs 42.194ns)
• 💚 Zero allocations (vs 88 bytes per invocation)
• 📊 More consistent - lower standard deviation (±0.042ns vs ±0.518ns)
• ✅ Compile-time safety - catch errors during build
• 🔍 Better debugging - see and step through generated code
• 🎯 100% optional - works alongside reflection-based code

📖 Complete Source Generator Guide with full benchmark results

Documentation

• Complete Usage Guide - Comprehensive examples with LaFacture SaaS patterns
• Source Generator Guide - 27x performance boost with compile-time code generation
• Session Summary - Development history and architecture decisions

Use Cases

• Multi-Tenant SaaS - Subscription-based feature access (Free/Premium/Enterprise)
• Order Processing - Validation → Tax → Payment → Shipping → Fulfillment
• Invoice Generation - Calculation → Tax → PDF → Email → Accounting
• Data Transformation - Extract → Transform → Validate → Load → Notify
• Workflow Automation - Pluggable business rules and approval workflows

Architecture

The framework uses a hook/plugin architecture:

┌─────────────────────────────────────────────┐
│           Fixed Event Flow                  │
│  (Defined by TEvents interface)             │
├─────────────────────────────────────────────┤
│  ValidateOrder → ProcessPayment → ShipOrder │
│       ↓              ↓               ↓      │
│    [Hook 1]      [Hook 2]        [Hook 3]  │
│       ↓              ↓               ↓      │
│  Module A        Module B         Module C  │
│  Module B        Module C         Module D  │
│  Module C                                   │
└─────────────────────────────────────────────┘

Key Properties:

• Fixed flow ensures predictable execution order
• Multiple modules can hook into same event
• Modules execute in registration order
• Middleware wraps entire execution for cross-cutting concerns

Roadmap

See GitHub Issues for planned features:

• Phase 1: Source Generator (27x performance boost), Pipeline Caching, Async Batching
• Phase 2: Visual Designer, Testing Framework, Hot Reload
• Phase 3: Conditional Loading, Module Dependencies, Circuit Breaker
• Phase 4: OpenTelemetry, Analytics Dashboard, Audit Trail
• Phase 5: Saga Pattern, Event Sourcing, Pipeline Composition
• Phase 6: Feature Flags, Tenant Overrides, Resource Quotas

Requirements

• .NET 8.0 or higher
• Microsoft.Extensions.DependencyInjection 8.0+
• Microsoft.Extensions.Configuration 8.0+

License

MIT License - see LICENSE file for details

Contributing

This is currently a private repository. Contributions will be accepted once the repository is made public.

Author

Mohamed Abou Chalkha

Support

• GitHub Issues: https://github.com/mabouchalkha/pipeline-framework/issues
• Documentation: USAGE_EXAMPLES.md