Dependency Injection

Prerequisites: Defining Pipelines

The NPipeline.Extensions.DependencyInjection package integrates NPipeline with Microsoft.Extensions.DependencyInjection. It handles node resolution, service lifetimes, and pipeline execution through the DI container.

Installation

dotnet add package NPipeline.Extensions.DependencyInjection

Quick Setup

The simplest approach scans assemblies for all NPipeline types:

var host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.AddNPipeline(Assembly.GetExecutingAssembly());
    })
    .Build();

await host.Services.RunPipelineAsync<OrderPipeline>();

AddNPipeline(Assembly[]) auto-discovers and registers:

• All INode implementations
• All IPipelineDefinition implementations
• All IResiliencePolicy implementations
• All IDeadLetterSink implementations
• All lineage sink implementations

Manual Registration

For explicit control, use the builder API:

services.AddNPipeline(builder =>
{
    builder.AddNode<OrderSource>();
    builder.AddNode<ValidateOrder>();
    builder.AddNode<OrderSink>();
    builder.AddPipeline<OrderPipeline>();
    builder.AddResiliencePolicy<RetryOnTransientErrors>();
    builder.AddDeadLetterSink<FileDeadLetterSink>();
});

Mix both approaches - scan assemblies, then add individual registrations:

services.AddNPipeline(builder =>
{
    builder.ScanAssemblies(Assembly.GetExecutingAssembly());
    builder.AddNode<ExternalNode>(ServiceLifetime.Singleton);
});

Service Lifetimes

All types default to Transient registration. Override the lifetime per type:

builder.AddNode<ExpensiveNode>(ServiceLifetime.Singleton);
builder.AddNode<CachedLookup>(ServiceLifetime.Scoped);

Lifetime	When to Use
Transient (default)	Stateless nodes, new instance per use
Scoped	Nodes that share state within a pipeline run
Singleton	Thread-safe nodes with expensive initialization

⚠️ Warning: Singleton nodes must be thread-safe. If your node holds mutable state, use Transient or Scoped.

Running Pipelines

From IServiceProvider

await host.Services.RunPipelineAsync<MyPipeline>();

// With parameters
await host.Services.RunPipelineAsync<MyPipeline>(
    new Dictionary<string, object> { ["date"] = DateTime.Today });

RunPipelineAsync creates a DI scope, resolves the runner and all dependencies, and executes the pipeline. The pipeline determines ownership per instance: the container manages nodes it resolves, while an instance the run creates or the builder hands over (for example through AddTap, AddBatcher or AddSink(instance)) is disposed at the end of the run that owns it.

From an Injected Runner

Inject IPipelineRunner into your own services, and create the context with CreatePipelineContext:

public class OrderService(IPipelineRunner runner, IServiceProvider services)
{
    public async Task ProcessOrdersAsync(CancellationToken ct)
    {
        await using var context = services.CreatePipelineContext(PipelineContextConfiguration.WithCancellation(ct));
        await runner.RunAsync<OrderPipeline>(context);
    }
}

CreatePipelineContext fills whatever the configuration leaves unset from the container: the error handler, lineage and observability factories, the ILoggerFactory and the IPipelineTracer. It also attaches the registered execution observer, which collects metrics when you use AddNPipelineObservability. A context created with new PipelineContext() or PipelineContext.CreateDefault() gets none of these, so lineage reports, metrics and NPipeline's own logging are silently lost. Pass the provider of the scope the run belongs to, the same one the runner comes from, and dispose the context when the run ends.

Constructor Injection in Nodes

When nodes are resolved through DI, they can take constructor dependencies:

public class EnrichOrder : TransformNode<Order, EnrichedOrder>
{
    private readonly IHttpClientFactory _httpFactory;
    private readonly ILogger<EnrichOrder> _logger;

    public EnrichOrder(IHttpClientFactory httpFactory, ILogger<EnrichOrder> logger)
    {
        _httpFactory = httpFactory;
        _logger = logger;
    }

    public override async ValueTask<EnrichedOrder> TransformAsync(
        Order item, PipelineContext context, CancellationToken ct)
    {
        var client = _httpFactory.CreateClient();
        _logger.LogInformation("Enriching order {OrderId}", item.Id);
        var details = await client.GetFromJsonAsync<Details>($"/orders/{item.Id}", ct);
        return new EnrichedOrder(item, details!);
    }
}

📝 Note: Without DI, nodes must have a parameterless constructor. The NodeParameterlessConstructorAnalyzer (NP9403) warns about this at build time.

Custom Factories and Ownership

The pipeline disposes what it creates, and leaves alone what a container owns. A run resolves nodes, dead-letter sinks, and lineage sinks through INodeFactory, IErrorHandlerFactory, and ILineageFactory, then asks each factory whether the instance it handed over belongs to the run:

public bool IsOwnedByRun(NodeDefinition nodeDefinition, INode instance) => true;
public bool CallerOwnsCreatedInstance(object instance) => true;

The default answer is true, which suits a factory that always builds fresh instances. A factory backed by a container overrides the method and reports false for the instances it resolved:

public sealed class MyNodeFactory(IServiceProvider serviceProvider) : INodeFactory
{
    private readonly HashSet<INode> _containerOwned = new(ReferenceEqualityComparer.Instance);

    public INode Create(NodeDefinition nodeDefinition, PipelineGraph graph) { /* ... */ }

    public bool IsOwnedByRun(NodeDefinition nodeDefinition, INode instance) => !_containerOwned.Contains(instance);
}

WARNING

Returning true for a container-owned instance disposes it twice: once by the run, once when the scope ends. Concurrent runs can share one factory (for example, a scoped factory used by several runs at once), so back the collection with a thread-safe type or guard it with a lock.

A single factory can mix both. Track only the instances the container resolves and let everything the factory constructs itself be disposed by the run.

What Gets Registered

AddNPipeline registers these core services automatically:

Service	Lifetime
IPipelineRunner	Scoped
PipelineBuilder	Transient
IPipelineFactory	Singleton
INodeFactory	Scoped
INodeExecutor	Scoped
ITopologyService	Scoped
IErrorHandlingService	Transient

Next Steps

• DI Extension Reference - registered services, node factory, and registry details
• Pipeline Context - pass runtime parameters and state
• Defining Pipelines - the builder API
• Parallel Execution - thread safety with DI-resolved nodes