Class TransformerExtensions

Namespace: Wolfgang.Etl.Transformers

Assembly: Wolfgang.Etl.Transformers.dll

Extension methods that compose transformers and wrap sequences with pipeline infrastructure.

public static class TransformerExtensions

Inheritance: object

TransformerExtensions

Inherited Members: object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.MemberwiseClone()

object.ReferenceEquals(object, object)

object.ToString()

Methods

Buffered<T>(IAsyncEnumerable<T>, int)

Inserts a BufferedTransformer<T> into a sequence, decoupling the upstream producer from the downstream consumer so both stages can run concurrently.

public static IAsyncEnumerable<T> Buffered<T>(this IAsyncEnumerable<T> source, int capacity) where T : notnull

Parameters

source IAsyncEnumerable<T>: The source sequence to buffer. Must not be null.
capacity int: The maximum number of items the internal buffer holds. Must be at least 1.

Returns

IAsyncEnumerable<T>: An IAsyncEnumerable<T> containing the same items as source, in the same order, but produced concurrently with consumption.

Type Parameters

T: The type of items in the sequence. Must be non-null.

Examples

var select = new SelectTransformer<RawRecord, ParsedRecord>(Parse);
var results = select.TransformAsync(extractor.ExtractAsync(token).Buffered(capacity: 500));
await foreach (var item in results) { ... }

Remarks

Sugar for new BufferedTransformer<T>(capacity).TransformAsync(source). See BufferedTransformer<T> for a full description of the buffering semantics, cancellation handling, and error propagation.

Exceptions

ArgumentNullException: source is null.
ArgumentOutOfRangeException: capacity is less than 1.

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

Composes two transformers into a single one: items flow through first, then through next. Equivalent to constructing new ChainTransformer<TSource, TIntermediate, TDestination>(first, next) but with all type parameters inferred at the call site.

public static ITransformAsync<TSource, TDestination> Then<TSource, TIntermediate, TDestination>(this ITransformAsync<TSource, TIntermediate> first, ITransformAsync<TIntermediate, TDestination> next) where TSource : notnull where TIntermediate : notnull where TDestination : notnull

Parameters

first ITransformAsync<TSource, TIntermediate>: The transformer that runs first.
next ITransformAsync<TIntermediate, TDestination>: The transformer that runs after first.

Returns

ITransformAsync<TSource, TDestination>: An Wolfgang.Etl.Abstractions.ITransformAsync<TSource, TDestination> representing the composed pipeline.

Type Parameters

TSource: The input type of the chain. Must be non-null.
TIntermediate: The intermediate type between the two transformers. Must be non-null.
TDestination: The output type of the chain. Must be non-null.

Remarks

Multiple stages compose by chaining successive .Then(...) calls. Each call returns a new ChainTransformer<TSource, TIntermediate, TDestination> with the previous chain as its first member, so the resulting structure is a left-leaning linked list of two-stage chains. The C# compiler infers all type parameters from the receiver and argument.

var pipeline = parseRaw       // string  -> DataRow
    .Then(normalize)          // DataRow -> DataRow
    .Then(lookupCustomer)     // DataRow -> CustomerRow
    .Then(validate)           // CustomerRow -> CustomerRow
    .Then(formatForLoad);     // CustomerRow -> LoadRow
// pipeline is ITransformAsync<string, LoadRow>

Exceptions

ArgumentNullException: first or next is null.

Then<TSource, TIntermediate, TDestination>(ITransformWithCancellationAsync<TSource, TIntermediate>, ITransformWithCancellationAsync<TIntermediate, TDestination>)

Composes two cancellation-aware transformers into a single one: items flow through first, then through next, and any CancellationToken supplied to the resulting chain is propagated to both stages.

public static ITransformWithCancellationAsync<TSource, TDestination> Then<TSource, TIntermediate, TDestination>(this ITransformWithCancellationAsync<TSource, TIntermediate> first, ITransformWithCancellationAsync<TIntermediate, TDestination> next) where TSource : notnull where TIntermediate : notnull where TDestination : notnull

Parameters

first ITransformWithCancellationAsync<TSource, TIntermediate>: The cancellation-aware transformer that runs first.
next ITransformWithCancellationAsync<TIntermediate, TDestination>: The cancellation-aware transformer that runs after first.

Returns

ITransformWithCancellationAsync<TSource, TDestination>: An Wolfgang.Etl.Abstractions.ITransformWithCancellationAsync<TSource, TDestination> representing the composed pipeline.

Type Parameters

TSource: The input type of the chain. Must be non-null.
TIntermediate: The intermediate type between the two transformers. Must be non-null.
TDestination: The output type of the chain. Must be non-null.

Remarks

This overload is selected by the C# compiler when both arguments implement Wolfgang.Etl.Abstractions.ITransformWithCancellationAsync<TSource, TDestination> (more specific than the base Wolfgang.Etl.Abstractions.ITransformAsync<TSource, TDestination> overload). The returned chain is itself Wolfgang.Etl.Abstractions.ITransformWithCancellationAsync<TSource, TDestination>, so subsequent .Then(...) calls in a longer chain also pick this overload, allowing arbitrary-length cancellation-aware chains to compose without ceremony.

Exceptions

ArgumentNullException: first or next is null.