Class DataFlowPipelineBuilder<TSource>

Namespace

DotnetKit.DataflowBuilder

Assembly

DotnetKit.DataflowBuilder.dll

Implements a fluent pipeline builder managing a sequence of linked IDataflowBlock instances.

public class DataFlowPipelineBuilder<TSource> : IPipelineBuilder<TSource>, IPipelineBuilder

Type Parameters

TSource

The type of elements accepted by the entry block of the pipeline.

Inheritance

object

DataFlowPipelineBuilder<TSource>

Implements

IPipelineBuilder<TSource>

IPipelineBuilder

Derived

DataFlowPipelineBuilder<TSource1, TSource2>

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.MemberwiseClone()

object.ReferenceEquals(object, object)

object.ToString()

Constructors

DataFlowPipelineBuilder(IDataflowBlock, CancellationToken)

Initializes a new instance of the DataFlowPipelineBuilder<TSource> class.

public DataFlowPipelineBuilder(IDataflowBlock initialBlock, CancellationToken cancellationToken)

Parameters

initialBlock IDataflowBlock

The entry block of the pipeline.

cancellationToken CancellationToken

The cancellation token governing pipeline lifecycle.

Exceptions

InvalidCastException

Thrown if initialBlock cannot be cast to ISourceBlock<TOutput>.

Fields

Blocks

The collection of blocks participating in the pipeline.

protected readonly List<IDataflowBlock> Blocks

Field Value

List<IDataflowBlock>

Properties

CancellationToken

Gets the cancellation token associated with the pipeline.

public CancellationToken CancellationToken { get; }

Property Value

CancellationToken

SourceBlock

Gets the entry source block of the pipeline.

public ISourceBlock<TSource> SourceBlock { get; }

Property Value

ISourceBlock<TSource>

Methods

AddPropagatorBlock<TInput, TOutput>(IPropagatorBlock<TInput, TOutput>)

Links the current head block to the specified propagator block with completion propagation and appends it to the pipeline.

public void AddPropagatorBlock<TInput, TOutput>(IPropagatorBlock<TInput, TOutput> propagatorBlock)

Parameters

propagatorBlock IPropagatorBlock<TInput, TOutput>

The propagator block to link and append.

Type Parameters

TInput

The input type expected by the propagator block.

TOutput

The output type produced by the propagator block.

Exceptions

InvalidOperationException

Thrown when the previous block is not a source block.

AddTargetBlock<TInput>(ITargetBlock<TInput>)

Links the current head block to the specified target block with completion propagation and appends it to the pipeline.

public void AddTargetBlock<TInput>(ITargetBlock<TInput> targetBlock)

Parameters

targetBlock ITargetBlock<TInput>

The target block to link and append.

Type Parameters

TInput

The input type accepted by the target block.

Exceptions

InvalidOperationException

Thrown when the previous block is not a source block.

Build()

Builds and encapsulates the configured dataflow blocks into an executable IPipeline<TSource>.

public virtual IPipeline<TSource> Build()

Returns

IPipeline<TSource>

An executable dataflow pipeline.

CreateActionBlock<TInput>(Action<TInput>, int)

Creates a synchronous action target block with error trapping that faults the pipeline entry upon uncaught exceptions.

public ITargetBlock<TInput> CreateActionBlock<TInput>(Action<TInput> targetAction, int maxDegreeOfParallelism)

Parameters

targetAction Action<TInput>

The action executed for each item.

maxDegreeOfParallelism int

The maximum degree of parallel execution.

Returns

ITargetBlock<TInput>

A configured action block.

Type Parameters

TInput

The input item type.

CreateAscynActionBlock<TInput>(Func<TInput, Task>, int)

Creates an asynchronous terminal action block (ActionBlock<TInput>) with error handling.

[Obsolete("Use CreateAsyncActionBlock instead. This method has a typo and will be removed in a future release.")]
public ITargetBlock<TInput> CreateAscynActionBlock<TInput>(Func<TInput, Task> targetAction, int maxDegreeOfParallelism)

Parameters

targetAction Func<TInput, Task>

The asynchronous consumer action invoked for each incoming item.

maxDegreeOfParallelism int

The maximum number of concurrent tasks allowed.

Returns

ITargetBlock<TInput>

A configured target block.

Type Parameters

TInput

The input item type.

CreateAsyncActionBlock<TInput>(Func<TInput, Task>, int)

Creates an asynchronous action target block with error trapping that faults the pipeline entry upon uncaught exceptions.

public ITargetBlock<TInput> CreateAsyncActionBlock<TInput>(Func<TInput, Task> targetAction, int maxDegreeOfParallelism)

Parameters

targetAction Func<TInput, Task>

The asynchronous action executed for each item.

maxDegreeOfParallelism int

The maximum degree of parallel execution.

Returns

ITargetBlock<TInput>

A configured action block.

Type Parameters

TInput

The input item type.

CreateAsyncTransformBlock<TInput, TOutput>(Func<TInput, Task<TOutput>>, int)

Creates an asynchronous transform block mapping an input item to a task yielding an output item.

public IPropagatorBlock<TInput, TOutput> CreateAsyncTransformBlock<TInput, TOutput>(Func<TInput, Task<TOutput>> propagatorFunc, int maxDegreeOfParallelism)

Parameters

propagatorFunc Func<TInput, Task<TOutput>>

The asynchronous transform function.

maxDegreeOfParallelism int

The maximum degree of parallel execution.

Returns

IPropagatorBlock<TInput, TOutput>

A configured transform block.

Type Parameters

TInput

The input item type.

TOutput

The transformed item type.

CreateAsyncTransformManyBlock<TInput, TOutput>(Func<TInput, Task<IEnumerable<TOutput>>>, int)

Creates an asynchronous transform-many block expanding each input item into an asynchronous sequence of output items.

public IPropagatorBlock<TInput, TOutput> CreateAsyncTransformManyBlock<TInput, TOutput>(Func<TInput, Task<IEnumerable<TOutput>>> propagatorFunc, int maxDegreeOfParallelism)

Parameters

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

The asynchronous transform function returning an enumerable of outputs.

maxDegreeOfParallelism int

The maximum degree of parallel execution.

Returns

IPropagatorBlock<TInput, TOutput>

A configured transform-many block.

Type Parameters

TInput

The input item type.

TOutput

The output sequence element type.

CreateBatchBlock<TInput>(int)

Creates a batching block that groups consecutive input items into fixed-size arrays.

public IPropagatorBlock<TInput, TInput[]> CreateBatchBlock<TInput>(int batchSize)

Parameters

batchSize int

The maximum number of items in each batch.

Returns

IPropagatorBlock<TInput, TInput[]>

A batching propagator block.

Type Parameters

TInput

The type of input items.

CreateTransformBlock<TInput, TOutput>(Func<TInput, TOutput>, int)

Creates a synchronous transform block mapping an input item to an output item.

public IPropagatorBlock<TInput, TOutput> CreateTransformBlock<TInput, TOutput>(Func<TInput, TOutput> propagatorFunc, int maxDegreeOfParallelism)

Parameters

propagatorFunc Func<TInput, TOutput>

The transform function.

maxDegreeOfParallelism int

The maximum degree of parallel execution.

Returns

IPropagatorBlock<TInput, TOutput>

A configured transform block.

Type Parameters

TInput

The input item type.

TOutput

The transformed item type.

CreateTransformManyBlock<TInput, TOutput>(Func<TInput, IEnumerable<TOutput>>, int)

Creates a synchronous transform-many block expanding each input item into a sequence of output items.

public IPropagatorBlock<TInput, TOutput> CreateTransformManyBlock<TInput, TOutput>(Func<TInput, IEnumerable<TOutput>> propagatorFunc, int maxDegreeOfParallelism = 1)

Parameters

propagatorFunc Func<TInput, IEnumerable<TOutput>>

The transform function returning an enumerable of outputs.

maxDegreeOfParallelism int

The maximum degree of parallel execution.

Returns

IPropagatorBlock<TInput, TOutput>

A configured transform-many block.

Type Parameters

TInput

The input item type.

TOutput

The output sequence element type.

HandleAsyncAction<TInput>(Func<TInput, Task>)

Wraps an asynchronous action with exception handling that faults the pipeline entry if an unhandled error occurs.

protected Func<TInput, Task> HandleAsyncAction<TInput>(Func<TInput, Task> asyncAction)

Parameters

asyncAction Func<TInput, Task>

The delegate to wrap.

Returns

Func<TInput, Task>

The wrapped delegate.

Type Parameters

TInput

The input item type.

HandleAsyncFunction<TInput, TOutput>(Func<TInput, Task<TOutput>>)

Wraps an asynchronous function with exception handling that faults the pipeline entry if an unhandled error occurs.

protected Func<TInput, Task<TOutput>> HandleAsyncFunction<TInput, TOutput>(Func<TInput, Task<TOutput>> asyncFunction)

Parameters

asyncFunction Func<TInput, Task<TOutput>>

The delegate to wrap.

Returns

Func<TInput, Task<TOutput>>

The wrapped delegate.

Type Parameters

TInput

The input item type.

TOutput

The output item type.