NPipeline

Why NPipeline?

Traditional data processing in .NET often devolves into unmanageable chains of nested loops or heavy dependencies on external ETL tools. NPipeline bridges the gap, offering a code-first approach that prioritizes developer experience and execution efficiency.

Graph-Based Architecture

Visualize your data flow clearly. Pipelines are built as interconnected nodes (Sources, Transforms, Sinks), ensuring visibility into the data's journey.

Compile-Time Safety

Connect nodes using a fluent, type-safe API. The compiler ensures that the output type of an upstream node matches the input of the downstream node, eliminating runtime type errors.

Async-First

Built for modern .NET, leveraging asynchronous patterns and IAsyncEnumerable<T> for non-blocking I/O and efficient resource usage.

Use Cases

NPipeline is engineered to handle structured, high-throughput scenarios where reliability is paramount.

ETL Workflows

Build reliable ingestion pipelines that extract data, apply complex transformations, and load into storage.

Real-time Stream Processing

Handle low-latency data streams from message queues or IoT devices.

Data Validation & Cleansing

Implement rigorous quality checks and validation rules as discrete, testable steps.

Microservice Integration

Facilitate structured data exchange and transformation between decoupled services.

Engineered for Performance

Performance is not an afterthought; it is the cornerstone of NPipeline's design. The library is architected to minimize garbage collection overhead and maximize throughput.

Plan-Based Execution

NPipeline separates the "planning" phase from the "execution" phase. The pipeline structure is compiled once, meaning per-item processing avoids expensive reflection or routing logic during the steady state.

Zero-Allocation Fast Paths

In high-throughput systems, every allocation counts. NPipeline utilizes ValueTask<T> to implement a "two-path" pattern.

Fast Path (Synchronous)

If a result is available immediately (e.g., cache hits, simple math), it allocates on the stack with zero GC pressure.

Slow Path (Asynchronous)

Seamlessly transitions to true async only when I/O or heavy computation is required.

Impact

In high-cache-hit scenarios, this can eliminate thousands of allocations per second, drastically reducing garbage collection pauses.

Resilience & Reliability

Production pipelines fail—it's inevitable. NPipeline provides a comprehensive resilience framework to detect, handle, and recover from failures without crashing your application.

Granular Error Handling

Node-Level

Handle specific item failures (e.g., malformed JSON) by retrying, skipping, or routing to a Dead Letter Queue.

Pipeline-Level

Manage stream-wide failures (e.g., database outages) with Circuit Breakers and automatic Node Restarts.

Build-Time Analyzers

Don't wait for a 3 AM page to find a configuration error. NPipeline includes Roslyn analyzers that enforce best practices at compile time.

Prevents Silent Failures

Detects missing prerequisites for node restarts (e.g., missing materialization buffers).

Enforces Streaming Patterns

Flags blocking operations or non-streaming patterns that could lead to memory exhaustion.

Fluent API & Type Safety

Constructing a pipeline is intuitive and readable. The builder pattern allows you to define sources, transforms, and sinks, and connect them logically.

public void Define(PipelineBuilder builder, PipelineContext context) 
{
    // 1. Add Nodes
    var source = builder.AddSource<OrderSource, Order>();
    var filter = builder.AddTransform<FraudCheckTransform, Order, Order>();
    var sink = builder.AddSink<DatabaseSink, Order>();

    // 2. Connect the Graph
    // The compiler ensures type compatibility between nodes
    builder.Connect(source, filter);
    builder.Connect(filter, sink);
    
    // 3. Configure Resilience
    builder.WithRetryOptions(new PipelineRetryOptions(
        MaxItemRetries: 3, 
        MaxNodeRestartAttempts: 2
    ));
}

Connectors & Extensions

NPipeline is modular by design. Keep the core lightweight and add capabilities as you need them.

Dependency Injection

Seamlessly integrate with Microsoft.Extensions.DependencyInjection for robust service management.

Parallelism

Utilize the ParallelExecutionStrategy to process nodes concurrently, increasing throughput for CPU-bound tasks.

Connectors

Use pre-built Source and Sink connectors for common targets like CSV files, leveraging a unified storage abstraction.

Testing

Specialized libraries including integration with FluentAssertions and AwesomeAssertions to unit test your pipelines in memory.

Get Started

Ready to build better pipelines?

Install via NuGet:

dotnet add package NPipeline