diff --git a/CHANGELOG.md b/CHANGELOG.md index 7dddd854..86e35911 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +## [0.20.0] - 2026-07-30 + +Minor release: a dependency-free retry seam, composable per-item middleware, and pipeline-wide +error aggregation on the base stages. Purely additive — no breaking change (validates against the +0.19.0 baseline). + +### Added + +- **`IReportsItemErrors` (#335):** a small interface (`int CurrentErrorItemCount { get; }`) implemented + by `ExtractorBase`, `LoaderBase`, and `TransformerBase`, letting a pipeline read any stage's + error-item count uniformly regardless of concrete type. +- **Middleware / interceptor (#93):** a composable per-item hook — `IItemMiddleware` returning + `MiddlewareResult` (`Continue` to keep/replace an item, `Drop` to remove it) — attached to any + stream with the `WithMiddleware(...)` extensions (single or ordered chain). Lets cross-cutting + concerns (logging, validation, metrics, throttling, dedup) decorate an extractor/transformer output + or loader input without changing the component, and composes inside an `EtlPipeline` via + `Through(s => s.WithMiddleware(...))`. Dependency-free. +- **Retry seam (#94):** `ExtractorBase`, `LoaderBase`, and `TransformerBase` gained a + `protected virtual WrapWorkerExecution(...)` hook wrapped around every worker invocation. The + default implementation is a no-op, so behaviour is unchanged; override it to run the worker through + a retry / resilience strategy. The override receives a re-invocable worker factory (call it again to + retry) and stream-level semantics are documented on the method. Kept dependency-free — a ready-made + Polly integration will ship as a separate opt-in `Wolfgang.Etl.Polly` package (#332). + +### Changed + +- **`EtlPipelineProgress.ErrorItemCount` now aggregates every stage (#335).** Previously it reported + only the extractor's error-item count; it now sums the error-item counts of the source, every + transformer, and the loader (each stage that implements `IReportsItemErrors`), so an item any + stage's error policy discarded is reflected in the total. Pre-1.0 behaviour change. +- _Internal (no public API change):_ the base classes' `StartedAt` / `Elapsed` timing now reads + through an injectable time source (#338), so downstream test kits can drive the `Report` + throughput / ETA metrics from a fake clock (unblocks ETL-Test-Kit#262). + ## [0.19.0] - 2026-07-29 Minor release: makes the `EtlPipelineProgress` record counters overflow-safe. diff --git a/README.md b/README.md index 9046ea96..07ff3ae4 100644 --- a/README.md +++ b/README.md @@ -156,6 +156,9 @@ await Pipeline | Throughput & ETA | `Report` exposes `StartedAt`, `Elapsed`, `ItemsPerSecond`, `PercentComplete`, and `EstimatedRemaining` derived from the item count and elapsed time | | Resource Disposal | Base classes implement `IDisposable` / `IAsyncDisposable`; override `Dispose(bool)` or `DisposeAsync()` to release resources deterministically | | Per-run State | Item counts and timing reset at the start of each enumeration, so a reused component reports the current run rather than cumulative totals | +| Per-item Error Handling | Opt-in `OnItemError` policy (`Abort` / `Skip`) on each base stage with a distinct `CurrentErrorItemCount`; the pipeline sums error counts across every stage (source, transformers, sink) into `EtlPipelineProgress.ErrorItemCount` | +| Middleware / Interceptors | Attach composable per-item behaviour (logging, validation, metrics, throttling, dedup) to any stream with `WithMiddleware(...)` — an `IItemMiddleware` returning `MiddlewareResult` (`Continue` / `Drop`) | +| Retry Seam | `WrapWorkerExecution` hook on each base stage to run the worker through a retry / resilience strategy (for example Polly); default no-op, dependency-free | | Cancellation | Full `CancellationToken` support across all operations | | Multi-TFM | Targets .NET Framework 4.6.2–4.8.1, .NET Standard 2.0, and .NET 5.0–10.0 | | Skip & Limit | `SkipItemCount` and `MaximumItemCount` for partial extraction/loading | diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs index aceca4a7..53dc8dc8 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs @@ -46,7 +46,11 @@ public IEtlPipeline Through(ITransformAsync transformer) throw new ArgumentNullException(nameof(transformer)); } - return new EtlPipelineImpl((state, token) => transformer.TransformAsync(_factory(state, token))); + return new EtlPipelineImpl((state, token) => + { + RegisterErrorReader(transformer, state); + return transformer.TransformAsync(_factory(state, token)); + }); } @@ -59,7 +63,11 @@ public IEtlPipeline Through(ITransformWithCancellationAsync throw new ArgumentNullException(nameof(transformer)); } - return new EtlPipelineImpl((state, token) => transformer.TransformAsync(_factory(state, token), token)); + return new EtlPipelineImpl((state, token) => + { + RegisterErrorReader(transformer, state); + return transformer.TransformAsync(_factory(state, token), token); + }); } @@ -109,6 +117,18 @@ public IAsyncEnumerable AsAsyncEnumerable(CancellationToken token = default) } + // Registers a transformer stage's error-item count with the run so + // EtlPipelineProgress.ErrorItemCount sums it alongside the source and sink. A transformer that + // does not report errors (for example a plain delegate transform) is skipped. + private static void RegisterErrorReader(object transformer, EtlRunState state) + { + if (transformer is IReportsItemErrors reporter) + { + state.AddErrorCountReader(() => reporter.CurrentErrorItemCount); + } + } + + // The head of every pipeline: pulls from the raw source, honours cancellation, and counts each // record as extracted. WithCancellation covers sources that observe the token via // [EnumeratorCancellation]; the explicit throw covers sources that ignore it. diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs index 25675a23..c806c9d7 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs @@ -35,6 +35,11 @@ LoaderBase loader public async Task RunAsync(IProgress? progress = null, CancellationToken token = default) { var state = new EtlRunState(); + + // Surface the loader's error-item count into the snapshot too, summed with the source and any + // transformer error counts so the pipeline total reflects every stage, not just the source. + state.AddErrorCountReader(() => _loader.CurrentErrorItemCount); + var stream = CountLoaded(_factory(state, token), state, progress, token); await _loader.LoadAsync(stream, token).ConfigureAwait(false); diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs index 7b36f8a3..b4ea611d 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs @@ -56,7 +56,8 @@ public static IEtlPipeline From(this EtlPipeline pipeline, Extr { // Surface the extractor's error-item count into the pipeline snapshot so a bad record // the extractor's error policy discarded is reported, not silently absent from the totals. - state.ErrorCountReader = () => extractor.CurrentErrorItemCount; + // Summed with any transformer/loader error counts (see EtlRunState.AddErrorCountReader). + state.AddErrorCountReader(() => extractor.CurrentErrorItemCount); return extractor.ExtractAsync(token); }); } diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs index 4c7fd8a9..6265d889 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; namespace Wolfgang.Etl.Abstractions; @@ -18,9 +19,14 @@ internal sealed class EtlRunState public long LoadedItemCount; - // Optional reader that surfaces an error-reporting source's error-item count into the snapshot. - // Left null for sources that don't report errors (e.g. a raw IAsyncEnumerable), which reads as 0. - public Func? ErrorCountReader; + // Error-item-count readers, one per stage that reports errors (source, transformers, sink). Their + // values are summed into the snapshot so an item any stage's error policy discarded is reported, + // not just the source's. Empty for a pipeline of stages that don't report errors, which reads as 0. + private readonly List> _errorCountReaders = new(); + + + // Registers a stage's error-item-count reader. Called once per stage as the factory chain runs. + public void AddErrorCountReader(Func reader) => _errorCountReaders.Add(reader); public EtlRunState() @@ -49,9 +55,15 @@ private TimeSpan Elapsed public EtlPipelineProgress Snapshot() { + long errorItemCount = 0; + foreach (var reader in _errorCountReaders) + { + errorItemCount += reader(); + } + return new EtlPipelineProgress(ExtractedItemCount, LoadedItemCount, Elapsed) { - ErrorItemCount = ErrorCountReader?.Invoke() ?? 0, + ErrorItemCount = errorItemCount, }; } } diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs index 2b795481..1876d3b7 100644 --- a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs +++ b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs @@ -19,6 +19,7 @@ namespace Wolfgang.Etl.Abstractions; /// The type of the progress object public abstract class ExtractorBase : IExtractWithProgressAndCancellationAsync, + IReportsItemErrors, IAsyncDisposable, IDisposable where TSource : notnull @@ -322,7 +323,7 @@ [EnumeratorCancellation] CancellationToken token { ResetRunState(); - await foreach (var item in ExtractWorkerAsync(token)) + await foreach (var item in WrapWorkerExecution(ExtractWorkerAsync, token)) { yield return item; } @@ -340,7 +341,7 @@ private async IAsyncEnumerable ExtractWithProgressAsync( try { - await foreach (var item in ExtractWorkerAsync(token)) + await foreach (var item in WrapWorkerExecution(ExtractWorkerAsync, token)) { yield return item; } @@ -382,6 +383,43 @@ private void ResetRunState() + /// + /// A resilience seam wrapped around every invocation of . The + /// default implementation simply invokes once, so extraction + /// behaves exactly as if the seam were absent. Override it to run the worker through a retry / + /// resilience strategy (for example a Polly ResiliencePipeline): the strategy can invoke + /// more than once, each call producing a fresh stream, to retry + /// a transient failure. + /// + /// + /// This is stream-level resilience: a retry re-runs the whole worker from the start, so a + /// failure part-way through re-yields items already seen. The per-run counters + /// (, , + /// ) are reset once at the start of the run, not on each + /// retry, so they accumulate across attempts unless the override resets them. Any delay the + /// override introduces must observe . Kept dependency-free by design — a + /// concrete Polly integration lives in a separate opt-in package rather than in this library. + /// + /// A factory that produces a fresh worker stream for the supplied token. Re-invocable — call it again to retry. + /// A to observe, including during any retry delay. + /// is . + /// The (possibly resilience-wrapped) stream of extracted items. + protected virtual IAsyncEnumerable WrapWorkerExecution + ( + Func> workerFactory, + CancellationToken token + ) + { + if (workerFactory is null) + { + throw new ArgumentNullException(nameof(workerFactory)); + } + + return workerFactory(token); + } + + + /// /// Creates a progress report of type TProgress. This gives the derived class the opportunity to /// implement a custom progress report that is specific to the extraction process. diff --git a/src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs b/src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs new file mode 100644 index 00000000..511f51df --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs @@ -0,0 +1,25 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace Wolfgang.Etl.Abstractions; + +/// +/// A composable, reusable hook for cross-cutting per-item behaviour (logging, validation, metrics, +/// throttling, deduplication) that can be attached to any stream without modifying the extractor, +/// transformer, or loader that produced it. Attach one or more with +/// ; +/// they run in the order attached, each seeing the item the previous one passed on. +/// +/// The item type flowing through the pipeline. +public interface IItemMiddleware +{ + /// + /// Invoked once per item. Return to keep the item + /// flowing (optionally replacing it), or to remove it from + /// the stream. + /// + /// The item to process. + /// A to observe. + /// The outcome describing whether to keep or drop the item. + ValueTask> OnItemAsync(T item, CancellationToken token); +} diff --git a/src/Wolfgang.Etl.Abstractions/IReportsItemErrors.cs b/src/Wolfgang.Etl.Abstractions/IReportsItemErrors.cs new file mode 100644 index 00000000..3b160c2d --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/IReportsItemErrors.cs @@ -0,0 +1,17 @@ +namespace Wolfgang.Etl.Abstractions; + +/// +/// Implemented by an ETL stage that counts items its error policy discarded, so a pipeline can read +/// that count uniformly regardless of the stage's concrete type. , +/// , and +/// all implement it, and EtlPipeline sums across every stage +/// that reports it into . +/// +public interface IReportsItemErrors +{ + /// + /// The number of items this stage's error policy has discarded (OnItemError returned + /// ) so far in the current run. + /// + int CurrentErrorItemCount { get; } +} diff --git a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs index a2971fd9..348fedf2 100644 --- a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs +++ b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs @@ -18,6 +18,7 @@ namespace Wolfgang.Etl.Abstractions; /// The type of the progress object public abstract class LoaderBase : ILoadWithProgressAndCancellationAsync, + IReportsItemErrors, IAsyncDisposable, IDisposable where TDestination : notnull @@ -323,7 +324,7 @@ CancellationToken token ) { ResetRunState(); - return LoadWorkerAsync(items, token); + return WrapWorkerExecution(ct => LoadWorkerAsync(items, ct), token); } @@ -339,7 +340,7 @@ private async Task LoadWithProgressAsync( try { - await LoadWorkerAsync(items, token).ConfigureAwait(false); + await WrapWorkerExecution(ct => LoadWorkerAsync(items, ct), token).ConfigureAwait(false); } finally { @@ -380,6 +381,43 @@ private void ResetRunState() + /// + /// A resilience seam wrapped around every invocation of . The + /// default implementation simply invokes once, so loading + /// behaves exactly as if the seam were absent. Override it to run the worker through a retry / + /// resilience strategy (for example a Polly ResiliencePipeline): the strategy can invoke + /// more than once to retry a transient failure. + /// + /// + /// This is stream-level resilience: a retry re-runs the whole worker, which re-enumerates + /// the source items from the start — so retry is only safe when that source can be + /// enumerated more than once. The per-run counters (, + /// , ) are reset once at + /// the start of the run, not on each retry, so they accumulate across attempts unless the + /// override resets them. Any delay the override introduces must observe . + /// Kept dependency-free by design — a concrete Polly integration lives in a separate opt-in + /// package rather than in this library. + /// + /// A factory that runs the worker for the supplied token. Re-invocable — call it again to retry. + /// A to observe, including during any retry delay. + /// is . + /// A task representing the (possibly resilience-wrapped) load operation. + protected virtual Task WrapWorkerExecution + ( + Func workerFactory, + CancellationToken token + ) + { + if (workerFactory is null) + { + throw new ArgumentNullException(nameof(workerFactory)); + } + + return workerFactory(token); + } + + + /// /// Creates a progress report of type TProgress. This gives the derived class the opportunity to /// implement a custom progress report that is specific to the loading process. diff --git a/src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs b/src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs new file mode 100644 index 00000000..7ef380cc --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs @@ -0,0 +1,146 @@ +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; + +namespace Wolfgang.Etl.Abstractions; + +/// +/// Extension methods that attach to an +/// stream, so cross-cutting per-item behaviour composes onto any +/// extractor / transformer output or loader input — and inside an EtlPipeline via +/// Through(stream => stream.WithMiddleware(...)) — without changing the component itself. +/// +public static class MiddlewareExtensions +{ + /// + /// Pipes every item of through . Items the + /// middleware drops () are removed from the stream; otherwise + /// the (possibly replaced) item is yielded. + /// + /// The item type. + /// The stream to decorate. + /// The middleware to run for each item. + /// A to observe. + /// The decorated stream. + /// or is . + public static IAsyncEnumerable WithMiddleware + ( + this IAsyncEnumerable source, + IItemMiddleware middleware, + CancellationToken token = default + ) + { + // Validate eagerly (at the call site), then delegate to the iterator, so a null argument fails + // fast rather than only when the returned stream is first enumerated. + if (source is null) + { + throw new ArgumentNullException(nameof(source)); + } + + if (middleware is null) + { + throw new ArgumentNullException(nameof(middleware)); + } + + return Iterate(source, middleware, token); + + + static async IAsyncEnumerable Iterate( + IAsyncEnumerable source, + IItemMiddleware middleware, + [EnumeratorCancellation] CancellationToken token) + { + // Stryker disable once Boolean: equivalent — with no synchronization context in play, ConfigureAwait(false) and (true) are indistinguishable. + await foreach (var item in source.WithCancellation(token).ConfigureAwait(false)) + { + // Stryker disable once Boolean: equivalent — with no synchronization context in play, ConfigureAwait(false) and (true) are indistinguishable. + var result = await middleware.OnItemAsync(item, token).ConfigureAwait(false); + if (!result.Skip) + { + yield return result.Item; + } + } + } + } + + + + /// + /// Pipes every item of through in order: + /// each middleware sees the item the previous one passed on. If any middleware drops the item + /// (), the remaining middleware is not run and the item is + /// removed from the stream. + /// + /// The item type. + /// The stream to decorate. + /// The middleware chain, applied in enumeration order. + /// A to observe. + /// The decorated stream. + /// or is , or a member of is . + public static IAsyncEnumerable WithMiddleware + ( + this IAsyncEnumerable source, + IEnumerable> middlewares, + CancellationToken token = default + ) + { + // Validate — and snapshot the chain — eagerly at the call site, so a null argument (or null + // member) fails fast rather than only when the returned stream is first enumerated. The + // snapshot also fixes the ordered set that runs for every item. + if (source is null) + { + throw new ArgumentNullException(nameof(source)); + } + + if (middlewares is null) + { + throw new ArgumentNullException(nameof(middlewares)); + } + + var chain = new List>(middlewares); + foreach (var middleware in chain) + { + if (middleware is null) + { + // Stryker disable once String: the exception message is diagnostic-only, not a behavioural contract asserted by tests. + throw new ArgumentNullException(nameof(middlewares), "A middleware in the chain is null."); + } + } + + return Iterate(source, chain, token); + + + static async IAsyncEnumerable Iterate( + IAsyncEnumerable source, + List> chain, + [EnumeratorCancellation] CancellationToken token) + { + // Stryker disable once Boolean: equivalent — with no synchronization context in play, ConfigureAwait(false) and (true) are indistinguishable. + await foreach (var item in source.WithCancellation(token).ConfigureAwait(false)) + { + var current = item; + var dropped = false; + + foreach (var middleware in chain) + { + // Stryker disable once Boolean: equivalent — with no synchronization context in play, ConfigureAwait(false) and (true) are indistinguishable. + var result = await middleware.OnItemAsync(current, token).ConfigureAwait(false); + if (result.Skip) + { + dropped = true; + break; + } + + current = result.Item; + } + + if (!dropped) + { + yield return current; + } + } + } + } +} diff --git a/src/Wolfgang.Etl.Abstractions/MiddlewareResult.Factory.cs b/src/Wolfgang.Etl.Abstractions/MiddlewareResult.Factory.cs new file mode 100644 index 00000000..b8f9a1c6 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/MiddlewareResult.Factory.cs @@ -0,0 +1,25 @@ +namespace Wolfgang.Etl.Abstractions; + +/// +/// Factory methods for creating values from an +/// implementation. +/// +public static class MiddlewareResult +{ + /// + /// Keeps the item in the stream, optionally replacing it with a transformed value. + /// + /// The item type. + /// The item to pass on. + /// A result that keeps flowing. + public static MiddlewareResult Continue(T item) => new(item, skip: false); + + + + /// + /// Drops the current item from the stream. + /// + /// The item type. + /// A result that discards the current item. + public static MiddlewareResult Drop() => new(default!, skip: true); +} diff --git a/src/Wolfgang.Etl.Abstractions/MiddlewareResult.cs b/src/Wolfgang.Etl.Abstractions/MiddlewareResult.cs new file mode 100644 index 00000000..ec86e59b --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/MiddlewareResult.cs @@ -0,0 +1,64 @@ +using System; +using System.Collections.Generic; + +namespace Wolfgang.Etl.Abstractions; + +/// +/// The outcome of running a single item through an : the +/// (possibly replaced) item to pass on, and whether the item should be dropped from the stream. +/// Create one with to keep an item flowing or +/// to discard it. +/// +/// The item type flowing through the pipeline. +public readonly struct MiddlewareResult : IEquatable> +{ + internal MiddlewareResult(T item, bool skip) + { + Item = item; + Skip = skip; + } + + + + /// + /// The item to pass on to the next middleware (or to the stream). Meaningful only when + /// is . + /// + public T Item { get; } + + + + /// + /// to drop the item from the stream (later middleware is not run and the + /// item is not yielded); to keep it. + /// + public bool Skip { get; } + + + + /// + public bool Equals(MiddlewareResult other) => + Skip == other.Skip && EqualityComparer.Default.Equals(Item, other.Item); + + + + /// + public override bool Equals(object? obj) => obj is MiddlewareResult other && Equals(other); + + + + /// + // Stryker disable once all: equivalent — any change to the hash formula still yields equal hash codes for equal values (the only GetHashCode contract), so no behavioural test can distinguish it. + public override int GetHashCode() => + unchecked(((Skip ? 1 : 0) * 397) ^ (Item is null ? 0 : EqualityComparer.Default.GetHashCode(Item))); + + + + /// Indicates whether two results are equal. + public static bool operator ==(MiddlewareResult left, MiddlewareResult right) => left.Equals(right); + + + + /// Indicates whether two results are not equal. + public static bool operator !=(MiddlewareResult left, MiddlewareResult right) => !left.Equals(right); +} diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt index 10094f57..98ca92ef 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt +++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt @@ -57,6 +57,8 @@ Wolfgang.Etl.Abstractions.IExtractWithProgressAndCancellationAsync.ExtractAsync(System.IProgress! progress, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! Wolfgang.Etl.Abstractions.IExtractWithProgressAsync Wolfgang.Etl.Abstractions.IExtractWithProgressAsync.ExtractAsync(System.IProgress! progress) -> System.Collections.Generic.IAsyncEnumerable! +Wolfgang.Etl.Abstractions.IItemMiddleware +Wolfgang.Etl.Abstractions.IItemMiddleware.OnItemAsync(T item, System.Threading.CancellationToken token) -> System.Threading.Tasks.ValueTask> Wolfgang.Etl.Abstractions.ILoadAsync Wolfgang.Etl.Abstractions.ILoadAsync.LoadAsync(System.Collections.Generic.IAsyncEnumerable! items) -> System.Threading.Tasks.Task! Wolfgang.Etl.Abstractions.ILoadWithCancellationAsync @@ -77,6 +79,8 @@ Wolfgang.Etl.Abstractions.IProgressTimer Wolfgang.Etl.Abstractions.IProgressTimer.Elapsed -> System.Action? Wolfgang.Etl.Abstractions.IProgressTimer.Start(int intervalMilliseconds) -> void Wolfgang.Etl.Abstractions.IProgressTimer.StopTimer() -> void +Wolfgang.Etl.Abstractions.IReportsItemErrors +Wolfgang.Etl.Abstractions.IReportsItemErrors.CurrentErrorItemCount.get -> int Wolfgang.Etl.Abstractions.ISupportDryRun Wolfgang.Etl.Abstractions.ISupportDryRun.IsDryRun.get -> bool Wolfgang.Etl.Abstractions.ISupportDryRun.IsDryRun.set -> void @@ -124,6 +128,13 @@ Wolfgang.Etl.Abstractions.LoaderBase.ReportingInterval. Wolfgang.Etl.Abstractions.LoaderBase.SkipItemCount.get -> int Wolfgang.Etl.Abstractions.LoaderBase.SkipItemCount.set -> void Wolfgang.Etl.Abstractions.LoaderBase.StartedAt.get -> System.DateTimeOffset? +Wolfgang.Etl.Abstractions.MiddlewareExtensions +Wolfgang.Etl.Abstractions.MiddlewareResult +Wolfgang.Etl.Abstractions.MiddlewareResult +Wolfgang.Etl.Abstractions.MiddlewareResult.Equals(Wolfgang.Etl.Abstractions.MiddlewareResult other) -> bool +Wolfgang.Etl.Abstractions.MiddlewareResult.Item.get -> T +Wolfgang.Etl.Abstractions.MiddlewareResult.MiddlewareResult() -> void +Wolfgang.Etl.Abstractions.MiddlewareResult.Skip.get -> bool Wolfgang.Etl.Abstractions.Pipeline Wolfgang.Etl.Abstractions.Report Wolfgang.Etl.Abstractions.Report.CurrentItemCount.get -> int @@ -161,10 +172,18 @@ abstract Wolfgang.Etl.Abstractions.LoaderBase.CreatePro abstract Wolfgang.Etl.Abstractions.LoaderBase.LoadWorkerAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! abstract Wolfgang.Etl.Abstractions.TransformerBase.CreateProgressReport() -> TProgress abstract Wolfgang.Etl.Abstractions.TransformerBase.TransformWorkerAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! +override Wolfgang.Etl.Abstractions.MiddlewareResult.Equals(object? obj) -> bool +override Wolfgang.Etl.Abstractions.MiddlewareResult.GetHashCode() -> int static Wolfgang.Etl.Abstractions.EtlPipeline.Create() -> Wolfgang.Etl.Abstractions.EtlPipeline! static Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions.DisposingOwned(this Wolfgang.Etl.Abstractions.IEtlPipelineSink! sink, params object![]! ownedResources) -> Wolfgang.Etl.Abstractions.IEtlPipelineSink! static Wolfgang.Etl.Abstractions.EtlPipelineSourceExtensions.From(this Wolfgang.Etl.Abstractions.EtlPipeline! pipeline, Wolfgang.Etl.Abstractions.ExtractorBase! extractor) -> Wolfgang.Etl.Abstractions.IEtlPipeline! static Wolfgang.Etl.Abstractions.EtlPipelineSourceExtensions.From(this Wolfgang.Etl.Abstractions.EtlPipeline! pipeline, System.Collections.Generic.IAsyncEnumerable! source) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware(this System.Collections.Generic.IAsyncEnumerable! source, System.Collections.Generic.IEnumerable!>! middlewares, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable! +static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware(this System.Collections.Generic.IAsyncEnumerable! source, Wolfgang.Etl.Abstractions.IItemMiddleware! middleware, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable! +static Wolfgang.Etl.Abstractions.MiddlewareResult.Continue(T item) -> Wolfgang.Etl.Abstractions.MiddlewareResult +static Wolfgang.Etl.Abstractions.MiddlewareResult.Drop() -> Wolfgang.Etl.Abstractions.MiddlewareResult +static Wolfgang.Etl.Abstractions.MiddlewareResult.operator !=(Wolfgang.Etl.Abstractions.MiddlewareResult left, Wolfgang.Etl.Abstractions.MiddlewareResult right) -> bool +static Wolfgang.Etl.Abstractions.MiddlewareResult.operator ==(Wolfgang.Etl.Abstractions.MiddlewareResult left, Wolfgang.Etl.Abstractions.MiddlewareResult right) -> bool static Wolfgang.Etl.Abstractions.Pipeline.Extract(Wolfgang.Etl.Abstractions.IExtractWithProgressAndCancellationAsync! extractor) -> Wolfgang.Etl.Abstractions.IExtractStageWithProgress! static Wolfgang.Etl.Abstractions.Pipeline.Extract(Wolfgang.Etl.Abstractions.IExtractWithProgressAsync! extractor) -> Wolfgang.Etl.Abstractions.IExtractStageWithProgress! static Wolfgang.Etl.Abstractions.Pipeline.Extract(Wolfgang.Etl.Abstractions.IExtractAsync! extractor) -> Wolfgang.Etl.Abstractions.IExtractStage! @@ -177,6 +196,7 @@ virtual Wolfgang.Etl.Abstractions.ExtractorBase.ExtractAsync virtual Wolfgang.Etl.Abstractions.ExtractorBase.ExtractAsync(System.IProgress! progress, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! virtual Wolfgang.Etl.Abstractions.ExtractorBase.ExtractAsync(System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! virtual Wolfgang.Etl.Abstractions.ExtractorBase.OnItemError(Wolfgang.Etl.Abstractions.ItemErrorContext! context) -> Wolfgang.Etl.Abstractions.ItemErrorAction +virtual Wolfgang.Etl.Abstractions.ExtractorBase.WrapWorkerExecution(System.Func!>! workerFactory, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! virtual Wolfgang.Etl.Abstractions.LoaderBase.CreateProgressTimer(System.IProgress! progress) -> Wolfgang.Etl.Abstractions.IProgressTimer! virtual Wolfgang.Etl.Abstractions.LoaderBase.Dispose(bool disposing) -> void virtual Wolfgang.Etl.Abstractions.LoaderBase.DisposeAsync() -> System.Threading.Tasks.ValueTask @@ -185,6 +205,7 @@ virtual Wolfgang.Etl.Abstractions.LoaderBase.LoadAsync( virtual Wolfgang.Etl.Abstractions.LoaderBase.LoadAsync(System.Collections.Generic.IAsyncEnumerable! items, System.IProgress! progress, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! virtual Wolfgang.Etl.Abstractions.LoaderBase.LoadAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! virtual Wolfgang.Etl.Abstractions.LoaderBase.OnItemError(Wolfgang.Etl.Abstractions.ItemErrorContext! context) -> Wolfgang.Etl.Abstractions.ItemErrorAction +virtual Wolfgang.Etl.Abstractions.LoaderBase.WrapWorkerExecution(System.Func! workerFactory, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! virtual Wolfgang.Etl.Abstractions.TransformerBase.CreateProgressTimer(System.IProgress! progress) -> Wolfgang.Etl.Abstractions.IProgressTimer! virtual Wolfgang.Etl.Abstractions.TransformerBase.Dispose(bool disposing) -> void virtual Wolfgang.Etl.Abstractions.TransformerBase.DisposeAsync() -> System.Threading.Tasks.ValueTask @@ -193,3 +214,4 @@ virtual Wolfgang.Etl.Abstractions.TransformerBase.TransformAsync(System.Collections.Generic.IAsyncEnumerable! items, System.IProgress! progress) -> System.Collections.Generic.IAsyncEnumerable! virtual Wolfgang.Etl.Abstractions.TransformerBase.TransformAsync(System.Collections.Generic.IAsyncEnumerable! items, System.IProgress! progress, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! virtual Wolfgang.Etl.Abstractions.TransformerBase.TransformAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! +virtual Wolfgang.Etl.Abstractions.TransformerBase.WrapWorkerExecution(System.Func!>! workerFactory, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs index 045eaafd..e6c86668 100644 --- a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs +++ b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs @@ -20,6 +20,7 @@ namespace Wolfgang.Etl.Abstractions; /// The type of the progress object public abstract class TransformerBase : ITransformWithProgressAndCancellationAsync, + IReportsItemErrors, IAsyncDisposable, IDisposable where TSource : notnull @@ -330,7 +331,7 @@ [EnumeratorCancellation] CancellationToken token { ResetRunState(); - await foreach (var item in TransformWorkerAsync(items, token)) + await foreach (var item in WrapWorkerExecution(ct => TransformWorkerAsync(items, ct), token)) { yield return item; } @@ -348,7 +349,7 @@ private async IAsyncEnumerable TransformWithProgressAsync( try { - await foreach (var item in TransformWorkerAsync(items, token)) + await foreach (var item in WrapWorkerExecution(ct => TransformWorkerAsync(items, ct), token)) { yield return item; } @@ -389,6 +390,44 @@ private void ResetRunState() + /// + /// A resilience seam wrapped around every invocation of . The + /// default implementation simply invokes once, so transformation + /// behaves exactly as if the seam were absent. Override it to run the worker through a retry / + /// resilience strategy (for example a Polly ResiliencePipeline): the strategy can invoke + /// more than once, each call producing a fresh stream, to retry + /// a transient failure. + /// + /// + /// This is stream-level resilience: a retry re-runs the whole worker, which re-enumerates + /// the source items from the start — so retry is only safe when that source can be + /// enumerated more than once, and a failure part-way through re-yields items already seen. The + /// per-run counters (, , + /// ) are reset once at the start of the run, not on each + /// retry, so they accumulate across attempts unless the override resets them. Any delay the + /// override introduces must observe . Kept dependency-free by design — a + /// concrete Polly integration lives in a separate opt-in package rather than in this library. + /// + /// A factory that produces a fresh worker stream for the supplied token. Re-invocable — call it again to retry. + /// A to observe, including during any retry delay. + /// is . + /// The (possibly resilience-wrapped) stream of transformed items. + protected virtual IAsyncEnumerable WrapWorkerExecution + ( + Func> workerFactory, + CancellationToken token + ) + { + if (workerFactory is null) + { + throw new ArgumentNullException(nameof(workerFactory)); + } + + return workerFactory(token); + } + + + /// /// Creates a progress report object of type TProgress. /// diff --git a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj index 7a5100d8..d0fdbebd 100644 --- a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj +++ b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj @@ -2,7 +2,7 @@ net462;net472;net48;net481;netstandard2.0;net5.0;net6.0;net7.0;net8.0;net9.0;net10.0 latest - 0.19.0 + 0.20.0