Class ProgressReportingTransformer<T>

Namespace

Wolfgang.Etl.Transformers

Assembly

Wolfgang.Etl.Transformers.dll

A decorator transformer that calls a user-supplied callback for each item as it passes through, then yields the item unchanged. Useful for reporting progress, logging, or collecting metrics on any stage in a composed pipeline.

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

Type Parameters

T

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

Inheritance

object

ProgressReportingTransformer<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>)

Remarks

ProgressReportingTransformer<T> is a pure pass-through: it does not filter, project, or reorder items. The callback is the only observable side-effect.

Insert it between any two stages in a chain to observe items at that point:

long count = 0;
var reporter = new ProgressReportingTransformer<Order>(item =>
    myProgress.Report(Interlocked.Increment(ref count)));
var pipeline = parse
.Then(reporter)       // taps the stream after parsing
.Then(validate);

Because it implements Wolfgang.Etl.Abstractions.ITransformAsync<TSource, TDestination>, it composes naturally with TransformerExtensions.

Callbacks are invoked on the caller's thread (or task) for each item before it is yielded to the downstream stage. Long-running synchronous callbacks will block the pipeline; use the async overload if the callback itself performs I/O.

Exceptions thrown by the callback propagate to the consumer through the normal IAsyncEnumerable<T> pull contract.

Constructors

ProgressReportingTransformer(Action<T>)

Initializes a new instance that invokes a synchronous callback for each item.

public ProgressReportingTransformer(Action<T> callback)

Parameters

callback Action<T>

The action to invoke for each item before it is yielded downstream. Must not be null.

Exceptions

ArgumentNullException

callback is null.

ProgressReportingTransformer(Func<T, ValueTask>)

Initializes a new instance that invokes an asynchronous callback for each item.

public ProgressReportingTransformer(Func<T, ValueTask> callback)

Parameters

callback Func<T, ValueTask>

The async function to await for each item before it is yielded downstream. Must not be null.

Exceptions

ArgumentNullException

callback is null.

Methods

TransformAsync(IAsyncEnumerable<T>)

Asynchronously yields each item from items, invoking the callback once per item before yielding it downstream.

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.