Class PipelineBuilderExtensions

Namespace

DotnetKit.DataflowBuilder

Assembly

DotnetKit.DataflowBuilder.dll

Provides fluent extension methods for configuring stages, transformations, batching, grouping, and targets on pipeline builders.

public static class PipelineBuilderExtensions

Inheritance

object

PipelineBuilderExtensions

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.MemberwiseClone()

object.ReferenceEquals(object, object)

object.ToString()

Methods

Batch<TSource, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, int)

Appends a batching block (BatchBlock<T>) to the pipeline, grouping incoming elements into arrays of fixed size.

public static IFluentBuilder<ISourceBlock<TInput[]>, IPipelineBuilder<TSource>> Batch<TSource, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, int batchSize = 100)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

batchSize int

The maximum number of items per batch. Defaults to 100.

Returns

IFluentBuilder<ISourceBlock<TInput[]>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the batch block yielding arrays of TInput.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type to be batched.

Examples

builder.Batch(batchSize: 50);

Batch<TSource, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, int, TimeSpan)

Appends a temporal batching block (TimeBatchBlock<T>) to the pipeline, grouping elements into batches of up to batchSize items, or flushing sooner if timeout expires since the first buffered element.

public static IFluentBuilder<ISourceBlock<TInput[]>, IPipelineBuilder<TSource>> Batch<TSource, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, int batchSize, TimeSpan timeout)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

batchSize int

The maximum number of items per batch.

timeout TimeSpan

The maximum time to wait before emitting a partial batch.

Returns

IFluentBuilder<ISourceBlock<TInput[]>, IPipelineBuilder<TSource>>

A fluent builder yielding arrays of TInput.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type to be batched.

Build<TSource, TInput>(IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>>)

Builds and finalizes a single-source dataflow pipeline from the configured builder stages.

public static IPipeline<TSource> Build<TSource, TInput>(this IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder)

Parameters

pipelineBuilder IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>>

The fluent builder terminating at a target block.

Returns

IPipeline<TSource>

An executable IPipeline<TSource> instance.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The element type consumed at the final target stage.

Build<TSource1, TSource2, TInput>(IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<Tuple<TSource1, TSource2>>>)

Builds and finalizes a multi-source dataflow pipeline from the configured builder stages.

public static IPipeline<TSource1, TSource2> Build<TSource1, TSource2, TInput>(this IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<Tuple<TSource1, TSource2>>> pipelineBuilder)

Parameters

pipelineBuilder IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<Tuple<TSource1, TSource2>>>

The fluent builder terminating at a target block.

Returns

IPipeline<TSource1, TSource2>

An executable IPipeline<TSource1, TSource2> instance.

Type Parameters

TSource1

The first entry element type of the pipeline.

TSource2

The second entry element type of the pipeline.

TInput

The element type consumed at the final target stage.

Exceptions

InvalidOperationException

Thrown if the underlying builder is not a multi-source builder.

FilterAsync<TSource, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, Task<bool>>, int)

Filters pipeline items asynchronously, allowing only items that satisfy predicate to pass downstream.

public static IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> FilterAsync<TSource, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, Task<bool>> predicate, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

predicate Func<TInput, Task<bool>>

An asynchronous function testing each item for a condition.

maxDegreeOfParallelism int

The maximum degree of parallel processing. Defaults to 1.

Returns

IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the filter stage.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type to filter.

Filter<TSource, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, bool>, int)

Filters pipeline items synchronously, allowing only items that satisfy predicate to pass downstream.

public static IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> Filter<TSource, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, bool> predicate, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

predicate Func<TInput, bool>

A function testing each item for a condition.

maxDegreeOfParallelism int

The maximum degree of parallel processing. Defaults to 1.

Returns

IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the filter stage.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type to filter.

Group<TSource, TGroupKey, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, TGroupKey>, int, int)

Appends a grouping block (GroupBlock<TGroupKey, TInput>) to the pipeline, grouping elements by a key selector and emitting batches when each group's capacity is reached or upon pipeline completion.

public static IFluentBuilder<ISourceBlock<TInput[]>, IPipelineBuilder<TSource>> Group<TSource, TGroupKey, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, TGroupKey> groupingSelector, int boundedCapacity = 100, int maxDegreeOfParallelism = 1) where TGroupKey : notnull

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

groupingSelector Func<TInput, TGroupKey>

A function extracting the grouping key from each element.

boundedCapacity int

The maximum number of items per group before a batch is emitted downstream. Defaults to 100.

maxDegreeOfParallelism int

The maximum degree of parallelism. Defaults to 1.

Returns

IFluentBuilder<ISourceBlock<TInput[]>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the grouping block yielding arrays of TInput.

Type Parameters

TSource

The entry element type of the pipeline.

TGroupKey

The non-nullable key type used for grouping.

TInput

The incoming element type to be grouped.

ProcessAsync<TSource, TInput, TOutput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, Task<TOutput>>, int)

Appends an asynchronous transformation block (TransformBlock<TInput, TOutput>) to the pipeline.

public static IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>> ProcessAsync<TSource, TInput, TOutput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, Task<TOutput>> propagatorFunc, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

propagatorFunc Func<TInput, Task<TOutput>>

The asynchronous transformation function mapping each input to a task producing an output.

maxDegreeOfParallelism int

The maximum degree of parallel processing. Defaults to 1.

Returns

IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the new transformation block.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type to be transformed.

TOutput

The transformed element type emitted downstream.

Examples

builder.ProcessAsync(async id => await repository.GetByIdAsync(id), maxDegreeOfParallelism: 4);

ProcessAsync<TSource, TInput, TOutput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, Task<TOutput>>, int, int, TimeSpan?, Action<TInput, Exception>?)

Appends an asynchronous transformation block with configurable retry resilience.

public static IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>> ProcessAsync<TSource, TInput, TOutput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, Task<TOutput>> propagatorFunc, int maxDegreeOfParallelism, int retryCount, TimeSpan? retryDelay = null, Action<TInput, Exception>? onFailed = null)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

propagatorFunc Func<TInput, Task<TOutput>>

The asynchronous transformation function.

maxDegreeOfParallelism int

The maximum degree of parallel processing.

retryCount int

Number of retry attempts upon failure.

retryDelay TimeSpan?

Optional delay between retry attempts.

onFailed Action<TInput, Exception>

Optional callback invoked with the item and final exception if all retries are exhausted.

Returns

IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the transformation block.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type to be transformed.

TOutput

The transformed element type emitted downstream.

ProcessManyAsync<TSource, TInput, TOutput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, Task<IEnumerable<TOutput>>>, int)

Appends an asynchronous one-to-many transformation block (TransformManyBlock<TInput, TOutput>) to the pipeline, expanding each input item asynchronously into a sequence of output items.

public static IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>> ProcessManyAsync<TSource, TInput, TOutput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, Task<IEnumerable<TOutput>>> propagatorFunc, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

propagatorFunc Func<TInput, Task<IEnumerable<TOutput>>>

The asynchronous function mapping each input item to a task returning an enumerable collection of outputs.

maxDegreeOfParallelism int

The maximum degree of parallel processing. Defaults to 1.

Returns

IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the new transform-many block.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type.

TOutput

The element type of the expanded sequence.

ProcessMany<TSource, TInput, TOutput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, IEnumerable<TOutput>>, int)

Appends a synchronous one-to-many transformation block (TransformManyBlock<TInput, TOutput>) to the pipeline, expanding each input item into a sequence of output items.

public static IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>> ProcessMany<TSource, TInput, TOutput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, IEnumerable<TOutput>> propagatorFunc, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

propagatorFunc Func<TInput, IEnumerable<TOutput>>

The function mapping each input item to an enumerable collection of output items.

maxDegreeOfParallelism int

The maximum degree of parallel processing. Defaults to 1.

Returns

IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the new transform-many block.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type.

TOutput

The element type of the expanded sequence.

Examples

builder.ProcessMany(line => line.Split(','));

Process<TSource, TInput, TOutput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, TOutput>, int)

Appends a synchronous transformation block (TransformBlock<TInput, TOutput>) to the pipeline.

public static IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>> Process<TSource, TInput, TOutput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, TOutput> propagatorFunc, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

propagatorFunc Func<TInput, TOutput>

The synchronous transformation function mapping each input to an output.

maxDegreeOfParallelism int

The maximum degree of parallel processing. Defaults to 1.

Returns

IFluentBuilder<ISourceBlock<TOutput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the output of the new transformation block.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The incoming element type to be transformed.

TOutput

The transformed element type emitted downstream.

Examples

builder.Process(x => x.Trim());

ToChannel<TSource, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, ChannelWriter<TInput>, bool, int)

Connects the output of the pipeline to a ChannelWriter<T>, forwarding all emitted items into the channel.

public static IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>> ToChannel<TSource, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, ChannelWriter<TInput> channelWriter, bool completeChannelOnCompletion = true, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

channelWriter ChannelWriter<TInput>

The channel writer receiving items.

completeChannelOnCompletion bool

Whether to complete the channel when the pipeline terminates.

maxDegreeOfParallelism int

The maximum degree of parallel channel writes.

Returns

IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the terminal block, ready to call Build().

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The element type written to the channel.

ToTargetAsync<TSource, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, Task>, int)

Appends an asynchronous terminal action block (ActionBlock<TInput>) to consume pipeline items.

public static IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>> ToTargetAsync<TSource, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, Task> targetAction, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

targetAction Func<TInput, Task>

The asynchronous action executed for each incoming item.

maxDegreeOfParallelism int

The maximum degree of parallel executions. Defaults to 1.

Returns

IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>>

A fluent builder configured at the target termination step, ready to call Build().

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The element type consumed by the target block.

Examples

builder.ToTargetAsync(async item => await database.SaveAsync(item), maxDegreeOfParallelism: 4);

ToTargetAsync<TSource, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Func<TInput, Task>, int, int, TimeSpan?, Action<TInput, Exception>?)

Appends an asynchronous terminal action block with configurable retry resilience.

public static IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>> ToTargetAsync<TSource, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Func<TInput, Task> targetAction, int maxDegreeOfParallelism, int retryCount, TimeSpan? retryDelay = null, Action<TInput, Exception>? onFailed = null)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

targetAction Func<TInput, Task>

The asynchronous action executed for each incoming item.

maxDegreeOfParallelism int

The maximum degree of parallel executions.

retryCount int

Number of retry attempts upon failure.

retryDelay TimeSpan?

Optional delay between retry attempts.

onFailed Action<TInput, Exception>

Optional callback invoked with the item and final exception if all retries are exhausted.

Returns

IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>>

A fluent builder positioned at the target termination step.

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The element type consumed by the target block.

ToTarget<TSource, TInput>(IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>, Action<TInput>, int)

Appends a synchronous terminal action block (ActionBlock<TInput>) to consume pipeline items.

public static IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>> ToTarget<TSource, TInput>(this IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>> pipelineBuilder, Action<TInput> targetAction, int maxDegreeOfParallelism = 1)

Parameters

pipelineBuilder IFluentBuilder<ISourceBlock<TInput>, IPipelineBuilder<TSource>>

The fluent pipeline builder instance.

targetAction Action<TInput>

The synchronous action executed for each incoming item.

maxDegreeOfParallelism int

The maximum degree of parallel executions. Defaults to 1.

Returns

IFluentBuilder<ITargetBlock<TInput>, IPipelineBuilder<TSource>>

A fluent builder configured at the target termination step, ready to call Build().

Type Parameters

TSource

The entry element type of the pipeline.

TInput

The element type consumed by the target block.

Examples

builder.ToTarget(item => Console.WriteLine(item));