Class EtlPipelineOperatorExtensions

Namespace: Wolfgang.Etl.Transformers

Assembly: Wolfgang.Etl.Transformers.dll

LINQ-flavored operator extension methods on Wolfgang.Etl.Abstractions.IEtlPipeline<T>. Each operator is a thin wrapper that appends one of this package's transformers to the pipeline via the core's Wolfgang.Etl.Abstractions.IEtlPipeline<T>.Through<TOut>(Wolfgang.Etl.Abstractions.ITransformAsync<T, TOut>) primitive — no iteration logic is re-implemented here.

public static class EtlPipelineOperatorExtensions

Inheritance: object

EtlPipelineOperatorExtensions

Inherited Members: object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.MemberwiseClone()

object.ReferenceEquals(object, object)

object.ToString()

Remarks

The pipeline core in Wolfgang.Etl.Abstractions is deliberately minimal and does not depend on this package, so the operator surface lives here. Add a using Wolfgang.Etl.Transformers; to light these up between the source (From(...)) and sink (To(...)) stages of a pipeline:

await EtlPipeline
    .Create()
    .From(records)
    .Where(r => r.Amount > 0)
    .Select(r => r.Id)
    .To(loader)
    .RunAsync();

Methods

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

Inserts a decoupling buffer so the upstream producer and downstream consumer run concurrently. The item type is unchanged; this batches nothing.

public static IEtlPipeline<T> Buffered<T>(this IEtlPipeline<T> pipeline, int capacity) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to buffer.
capacity int: The maximum number of items held in the buffer. Must be at least 1.

Returns

IEtlPipeline<T>: A pipeline of the same items, produced concurrently with consumption.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline is null.

Cast<TSource, TDestination>(IEtlPipeline<TSource>)

Casts each item to TDestination, throwing on an incompatible item.

public static IEtlPipeline<TDestination> Cast<TSource, TDestination>(this IEtlPipeline<TSource> pipeline) where TSource : notnull where TDestination : notnull

Parameters

pipeline IEtlPipeline<TSource>: The pipeline to cast.

Returns

IEtlPipeline<TDestination>: A pipeline of the cast items.

Type Parameters

TSource: The input item type. Must be non-null.
TDestination: The target type. Must be non-null.

Exceptions

ArgumentNullException: pipeline is null.

Chunk<T>(IEtlPipeline<T>, int)

Batches items into fixed-size chunks.

public static IEtlPipeline<IReadOnlyList<T>> Chunk<T>(this IEtlPipeline<T> pipeline, int size) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to batch.
size int: The number of items per chunk. Must be at least 1.

Returns

IEtlPipeline<IReadOnlyList<T>>: A pipeline of chunks; the final chunk may hold fewer than size items.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline is null.

DistinctBy<TSource, TKey>(IEtlPipeline<TSource>, Func<TSource, TKey>, IEqualityComparer<TKey>?)

Removes items with duplicate keys, optionally using a supplied key comparer.

public static IEtlPipeline<TSource> DistinctBy<TSource, TKey>(this IEtlPipeline<TSource> pipeline, Func<TSource, TKey> keySelector, IEqualityComparer<TKey>? comparer = null) where TSource : notnull where TKey : notnull

Parameters

pipeline IEtlPipeline<TSource>: The pipeline to deduplicate.
keySelector Func<TSource, TKey>: Extracts the key each item is deduplicated by.
comparer IEqualityComparer<TKey>: The comparer used to determine key equality, or null to use Default.

Returns

IEtlPipeline<TSource>: A pipeline keeping the first item seen for each distinct key.

Type Parameters

TSource: The item type. Must be non-null.
TKey: The key type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or keySelector is null.

Distinct<T>(IEtlPipeline<T>, IEqualityComparer<T>?)

Removes duplicate items, optionally using a supplied comparer.

public static IEtlPipeline<T> Distinct<T>(this IEtlPipeline<T> pipeline, IEqualityComparer<T>? comparer = null) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to deduplicate.
comparer IEqualityComparer<T>: The comparer used to determine equality, or null to use Default.

Returns

IEtlPipeline<T>: A pipeline with duplicate items removed, preserving first-seen order.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline is null.

Log<T>(IEtlPipeline<T>, Func<T, string>, Action<string>)

Writes a log line per item and passes the item through unchanged — a Tap<T>(IEtlPipeline<T>, Action<T>) bound to a formatter and a sink. Deliberately dependency-free: it takes a delegate sink rather than referencing Microsoft.Extensions.Logging, so bridge to an ILogger at the call site if desired.

public static IEtlPipeline<T> Log<T>(this IEtlPipeline<T> pipeline, Func<T, string> format, Action<string> sink) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to observe.
format Func<T, string>: Produces the log message for an item.
sink Action<string>: Receives each formatted message (e.g. Console.WriteLine or msg => logger.LogInformation(msg)).

Returns

IEtlPipeline<T>: A pipeline yielding the same items, in the same order, logging one message per item.

Type Parameters

T: The item type. Must be non-null.

Examples

await EtlPipeline
    .Create()
    .From(records)
    .Log(r => $"processing {r.Id}", Console.WriteLine)
    .To(loader)
    .RunAsync();

Exceptions

ArgumentNullException: pipeline, format, or sink is null.

OfType<TSource, TDestination>(IEtlPipeline<TSource>)

Passes through only the items assignable to TDestination.

public static IEtlPipeline<TDestination> OfType<TSource, TDestination>(this IEtlPipeline<TSource> pipeline) where TSource : notnull where TDestination : notnull

Parameters

pipeline IEtlPipeline<TSource>: The pipeline to filter by type.

Returns

IEtlPipeline<TDestination>: A pipeline of the items that are of type TDestination.

Type Parameters

TSource: The input item type. Must be non-null.
TDestination: The type to filter to. Must be non-null.

Exceptions

ArgumentNullException: pipeline is null.

SelectMany<TSource, TDestination>(IEtlPipeline<TSource>, Func<TSource, IAsyncEnumerable<TDestination>>)

Fans each item out to an asynchronous sequence and flattens the results.

public static IEtlPipeline<TDestination> SelectMany<TSource, TDestination>(this IEtlPipeline<TSource> pipeline, Func<TSource, IAsyncEnumerable<TDestination>> selector) where TSource : notnull where TDestination : notnull

Parameters

pipeline IEtlPipeline<TSource>: The pipeline to fan out.
selector Func<TSource, IAsyncEnumerable<TDestination>>: Maps each item to an asynchronous sequence of results.

Returns

IEtlPipeline<TDestination>: A pipeline of the flattened items.

Type Parameters

TSource: The input item type. Must be non-null.
TDestination: The flattened item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or selector is null.

SelectMany<TSource, TDestination>(IEtlPipeline<TSource>, Func<TSource, IEnumerable<TDestination>>)

Fans each item out to a synchronous sequence and flattens the results.

public static IEtlPipeline<TDestination> SelectMany<TSource, TDestination>(this IEtlPipeline<TSource> pipeline, Func<TSource, IEnumerable<TDestination>> selector) where TSource : notnull where TDestination : notnull

Parameters

pipeline IEtlPipeline<TSource>: The pipeline to fan out.
selector Func<TSource, IEnumerable<TDestination>>: Maps each item to a sequence of results.

Returns

IEtlPipeline<TDestination>: A pipeline of the flattened items.

Type Parameters

TSource: The input item type. Must be non-null.
TDestination: The flattened item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or selector is null.

Select<TSource, TDestination>(IEtlPipeline<TSource>, Func<TSource, ValueTask<TDestination>>)

Projects each item with an asynchronous selector.

public static IEtlPipeline<TDestination> Select<TSource, TDestination>(this IEtlPipeline<TSource> pipeline, Func<TSource, ValueTask<TDestination>> selector) where TSource : notnull where TDestination : notnull

Parameters

pipeline IEtlPipeline<TSource>: The pipeline to project.
selector Func<TSource, ValueTask<TDestination>>: The asynchronous projection applied to each item.

Returns

IEtlPipeline<TDestination>: A pipeline of the projected items.

Type Parameters

TSource: The input item type. Must be non-null.
TDestination: The projected item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or selector is null.

Select<TSource, TDestination>(IEtlPipeline<TSource>, Func<TSource, TDestination>)

Projects each item with a synchronous selector.

public static IEtlPipeline<TDestination> Select<TSource, TDestination>(this IEtlPipeline<TSource> pipeline, Func<TSource, TDestination> selector) where TSource : notnull where TDestination : notnull

Parameters

pipeline IEtlPipeline<TSource>: The pipeline to project.
selector Func<TSource, TDestination>: The projection applied to each item.

Returns

IEtlPipeline<TDestination>: A pipeline of the projected items.

Type Parameters

TSource: The input item type. Must be non-null.
TDestination: The projected item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or selector is null.

SkipWhile<T>(IEtlPipeline<T>, Func<T, bool>)

Skips items while a synchronous predicate holds, then yields the rest.

public static IEtlPipeline<T> SkipWhile<T>(this IEtlPipeline<T> pipeline, Func<T, bool> predicate) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to skip within.
predicate Func<T, bool>: Skipping continues while this returns true.

Returns

IEtlPipeline<T>: A pipeline yielding everything from the first item that fails predicate onward.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or predicate is null.

SkipWhile<T>(IEtlPipeline<T>, Func<T, ValueTask<bool>>)

Skips items while an asynchronous predicate holds, then yields the rest.

public static IEtlPipeline<T> SkipWhile<T>(this IEtlPipeline<T> pipeline, Func<T, ValueTask<bool>> predicate) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to skip within.
predicate Func<T, ValueTask<bool>>: Skipping continues while this returns true.

Returns

IEtlPipeline<T>: A pipeline yielding everything from the first item that fails predicate onward.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or predicate is null.

Skip<T>(IEtlPipeline<T>, int)

Skips the first count items, then yields the rest.

public static IEtlPipeline<T> Skip<T>(this IEtlPipeline<T> pipeline, int count) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to skip within.
count int: The number of leading items to skip. Must be non-negative.

Returns

IEtlPipeline<T>: A pipeline yielding the items after the first count.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline is null.

TakeWhile<T>(IEtlPipeline<T>, Func<T, bool>)

Yields items while a synchronous predicate holds, then stops.

public static IEtlPipeline<T> TakeWhile<T>(this IEtlPipeline<T> pipeline, Func<T, bool> predicate) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to truncate.
predicate Func<T, bool>: Yielding continues while this returns true.

Returns

IEtlPipeline<T>: A pipeline yielding the leading run of items satisfying predicate.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or predicate is null.

TakeWhile<T>(IEtlPipeline<T>, Func<T, ValueTask<bool>>)

Yields items while an asynchronous predicate holds, then stops.

public static IEtlPipeline<T> TakeWhile<T>(this IEtlPipeline<T> pipeline, Func<T, ValueTask<bool>> predicate) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to truncate.
predicate Func<T, ValueTask<bool>>: Yielding continues while this returns true.

Returns

IEtlPipeline<T>: A pipeline yielding the leading run of items satisfying predicate.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or predicate is null.

Take<T>(IEtlPipeline<T>, int)

Yields only the first count items, then stops.

public static IEtlPipeline<T> Take<T>(this IEtlPipeline<T> pipeline, int count) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to truncate.
count int: The maximum number of items to yield. Must be non-negative.

Returns

IEtlPipeline<T>: A pipeline yielding at most count items.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline is null.

Tap<T>(IEtlPipeline<T>, Action<T>)

Runs a synchronous side effect on each item and passes the item through unchanged — an observability "tap" point (logging, metrics, debugging) that does not alter the stream's shape.

public static IEtlPipeline<T> Tap<T>(this IEtlPipeline<T> pipeline, Action<T> onItem) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to observe.
onItem Action<T>: The side effect to run for each item before it flows on.

Returns

IEtlPipeline<T>: A pipeline yielding the same items, in the same order, with onItem invoked per item.

Type Parameters

T: The item type. Must be non-null.

Examples

await EtlPipeline
    .Create()
    .From(records)
    .Tap(r => Console.WriteLine($"seen {r.Id}"))
    .To(loader)
    .RunAsync();

Exceptions

ArgumentNullException: pipeline or onItem is null.

Tap<T>(IEtlPipeline<T>, Func<T, ValueTask>)

Runs an asynchronous side effect on each item and passes the item through unchanged — an observability "tap" point that does not alter the stream's shape.

public static IEtlPipeline<T> Tap<T>(this IEtlPipeline<T> pipeline, Func<T, ValueTask> onItem) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to observe.
onItem Func<T, ValueTask>: The asynchronous side effect to run for each item before it flows on.

Returns

IEtlPipeline<T>: A pipeline yielding the same items, in the same order, with onItem awaited per item.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or onItem is null.

Throttle<T>(IEtlPipeline<T>, TimeSpan)

Paces the pipeline so successive items are yielded at least minInterval apart, without changing the stream's shape or order — rate-limiting for a downstream sink that must not be hit too fast. The first item is not delayed; pacing is adaptive (a consumer that was already slow is not delayed further) and observes the pipeline's cancellation token.

public static IEtlPipeline<T> Throttle<T>(this IEtlPipeline<T> pipeline, TimeSpan minInterval) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to pace.
minInterval TimeSpan: The minimum time between successive items. Zero is a pass-through.

Returns

IEtlPipeline<T>: A pipeline yielding the same items, in the same order, paced apart.

Type Parameters

T: The item type. Must be non-null.

Examples

await EtlPipeline
    .Create()
    .From(records)
    .Throttle(TimeSpan.FromMilliseconds(200))   // at most ~5 items/second
    .To(loader)
    .RunAsync();

Exceptions

ArgumentNullException: pipeline is null.
ArgumentOutOfRangeException: minInterval is negative.

Where<T>(IEtlPipeline<T>, Func<T, bool>)

Filters the pipeline to items that satisfy a synchronous predicate.

public static IEtlPipeline<T> Where<T>(this IEtlPipeline<T> pipeline, Func<T, bool> predicate) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to filter.
predicate Func<T, bool>: The predicate an item must satisfy to be kept.

Returns

IEtlPipeline<T>: A pipeline yielding only the items for which predicate returns true.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or predicate is null.

Where<T>(IEtlPipeline<T>, Func<T, ValueTask<bool>>)

Filters the pipeline to items that satisfy an asynchronous predicate.

public static IEtlPipeline<T> Where<T>(this IEtlPipeline<T> pipeline, Func<T, ValueTask<bool>> predicate) where T : notnull

Parameters

pipeline IEtlPipeline<T>: The pipeline to filter.
predicate Func<T, ValueTask<bool>>: The asynchronous predicate an item must satisfy to be kept.

Returns

IEtlPipeline<T>: A pipeline yielding only the items for which predicate returns true.

Type Parameters

T: The item type. Must be non-null.

Exceptions

ArgumentNullException: pipeline or predicate is null.