Lambda Nodes

Prerequisites: Defining Pipelines

For simple transformations, creating a full class for every node is overkill. Lambda nodes let you define sources, transforms, and sinks inline as functions.

Lambda Sources

Provide a factory that returns the data to emit:

// From a synchronous collection
var source = builder.AddSource(() => new[] { 1, 2, 3, 4, 5 }, "numbers");

// From an async stream (preferred for large or I/O-bound data)
var source = builder.AddSource(
    (CancellationToken ct) => ReadLinesAsync("data.txt", ct),
    "file-lines");

The synchronous overload accepts Func<IEnumerable<T>>. The async overload accepts Func<CancellationToken, IAsyncEnumerable<T>> and streams items lazily.

Lambda Transforms

Map each input item to an output:

// Synchronous - no allocations via ValueTask fast path
var doubled = builder.AddTransform((int x) => x * 2, "double");

// Asynchronous - for I/O-bound work
var enriched = builder.AddTransform(
    async (Order order, CancellationToken ct) =>
    {
        var details = await _api.GetDetailsAsync(order.Id, ct);
        return new EnrichedOrder(order, details);
    },
    "enrich");

The synchronous overload (Func<TIn, TOut>) uses an optimized ValueTask path internally, avoiding per-item Task allocations.

Filtering

AddFilter passes through the items that satisfy a predicate and drops the rest:

// Synchronous
var active = builder.AddFilter((Order o) => o.Status == "Active", "active-only");

// Asynchronous - for a decision that needs I/O
var allowed = builder.AddFilter(
    async (Order o, CancellationToken ct) => await _rules.IsAllowedAsync(o, ct),
    "allowed");

A transform cannot drop an item — TransformAsync returns exactly one output per input — so a filter is a stream transform under the hood. Items are dropped as they are read; nothing is buffered.

One-to-Many

AddSelectMany expands each input item into zero or more output items — the other shape a transform cannot express:

// Synchronous
var lines = builder.AddSelectMany((Order o) => o.Lines, "order-lines");

// Asynchronous
var pages = builder.AddSelectMany(
    (Query q, CancellationToken ct) => _api.StreamPagesAsync(q, ct),
    "pages");

Each item's results are yielded as they are produced, so a selector returning a lazy sequence stays lazy.

Note: With item-level lineage enabled, both nodes declare their cardinality (OneToZeroOrOne and OneToMany), so the lineage mismatch detector stays quiet. Declaring a non-1:1 cardinality does mean lineage falls back to a materializing mapping strategy for that node — set LineageOptions.MaterializationCap if that matters for your stream sizes.

Lambda Sinks

Consume each item as it arrives:

// Synchronous
var logger = builder.AddSink((string line) => Console.WriteLine(line), "console");

// Asynchronous
var dbSink = builder.AddSink(
    async (Order order, CancellationToken ct) =>
    {
        await _db.InsertAsync(order, ct);
    },
    "database");

Complete Example

A full pipeline using only lambda nodes:

public class PricingPipeline : IPipelineDefinition
{
    public void Define(PipelineBuilder builder, PipelineContext context)
    {
        var source = builder.AddSource(
            () => new[] { 100m, 250m, 50m, 75m }, "prices");

        var discount = builder.AddTransform(
            (decimal price) => price * 0.9m, "apply-discount");

        var format = builder.AddTransform(
            (decimal price) => $"${price:F2}", "format");

        var print = builder.AddSink(
            (string price) => Console.WriteLine($"Final: {price}"), "display");

        builder.Connect(source, discount);
        builder.Connect(discount, format);
        builder.Connect(format, print);
    }
}

Mixing Lambdas with Class-Based Nodes

Lambda nodes and class-based nodes are interchangeable in the same pipeline. Use lambdas for simple operations and classes for complex logic:

public void Define(PipelineBuilder builder, PipelineContext context)
{
    var source = builder.AddSource(() => LoadOrders(), "orders");
    var validate = builder.AddTransform<OrderValidator, Order, ValidatedOrder>("validate");
    var log = builder.AddSink(
        (ValidatedOrder o) => Console.WriteLine($"Processed: {o.Id}"), "log");

    builder.Connect(source, validate);
    builder.Connect(validate, log);
}

When to Use Lambda Nodes

Use Lambdas When	Use Classes When
Logic fits in 1–3 lines	Transform has complex branching or state
No constructor dependencies needed	Node needs injected services
Prototyping or scripting	Node is reused across pipelines
The transform is purely functional	Node holds resources requiring disposal

💡 Tip: You can also pass a pre-configured node instance directly: builder.AddSource(myNodeInstance, "name") or builder.AddSink(mySinkInstance, "name").

Next Steps

• Custom Nodes - write full node classes for complex logic
• Defining Pipelines - the builder API in detail
• Branching and Merging - fan-out patterns