Class BufferedTransformer<T>

Namespace

Wolfgang.Etl.Transformers

Assembly

Wolfgang.Etl.Transformers.dll

A transformer that decouples its upstream and downstream stages using a bounded Channel<T>, enabling pipeline parallelism: the upstream stage can run ahead while the downstream stage consumes, instead of being lock-stepped by IAsyncEnumerable<T>'s pull model.

public sealed class BufferedTransformer<T> : ITransformAsync<T, T> where T : notnull

Type Parameters

T

The type of items flowing through the transformer. Must be non-null.

Inheritance

object

BufferedTransformer<T>

Implements

ITransformAsync<T, T>

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

Extension Methods

TransformerExtensions.Then<TSource, TIntermediate, TDestination>(ITransformAsync<TSource, TIntermediate>, ITransformAsync<TIntermediate, TDestination>)

Examples

extractor
    .Pipe(new WhereTransformer<Row>(r => r.IsValid))
    .Pipe(new BufferedTransformer<Row>(capacity: 500))   // parallelism boundary
    .Pipe(new SelectTransformer<Row, Record>(Parse))
    .Pipe(loader);

Remarks

With plain IAsyncEnumerable<T> chaining, every stage runs on the same logical call path - throughput is bounded by the slowest stage and fast stages idle while waiting. Inserting a BufferedTransformer<T> between two stages introduces a producer task that drains the upstream into a bounded buffer; the downstream stage then reads from the buffer. The two sides run concurrently, so total throughput approaches max(stage speeds) rather than min(stage speeds).

The buffer is bounded - once full, the producer task awaits free space, providing backpressure on the source.

Cancellation: consumers do not need a transformer-level token. External cancellation supplied via .WithCancellation(token) on the returned sequence is propagated to the internal producer task via a linked CancellationTokenSource, so the source enumerator and producer are cleaned up promptly.

Error propagation: exceptions thrown by the source enumerator or by writes into the buffer are surfaced to the consumer through the channel - the consumer's await foreach throws the original exception (unwrapped) once any already-buffered items have drained.

Implements only Wolfgang.Etl.Abstractions.ITransformAsync<TSource, TDestination> - matching the lightweight pattern of the rest of the library. The producer task's lifecycle is managed internally via the iterator's disposal contract.

Constructors

BufferedTransformer(int)

Initializes a new instance with the given buffer capacity.

public BufferedTransformer(int capacity)

Parameters

capacity int

The maximum number of items that may be buffered between the upstream and downstream stages. Must be at least 1.

Exceptions

ArgumentOutOfRangeException

capacity is less than 1.

Properties

Capacity

The maximum number of items the internal buffer holds.

public int Capacity { get; }

Property Value

int

Methods

TransformAsync(IAsyncEnumerable<T>)

Asynchronously yields each item from items, with a bounded buffer in between that allows the upstream and downstream stages to run concurrently.

public IAsyncEnumerable<T> TransformAsync(IAsyncEnumerable<T> items)

Parameters

items IAsyncEnumerable<T>

The asynchronous source sequence.

Returns

IAsyncEnumerable<T>

An asynchronous sequence containing the same items as items, in the same order.

Exceptions

ArgumentNullException

items is null.