From 2f2795d6190c3b98ae3cfb23a96a37b4f9f25cb6 Mon Sep 17 00:00:00 2001 From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com> Date: Mon, 27 Jul 2026 21:23:54 -0400 Subject: [PATCH 1/7] feat: dependency-free WrapWorkerExecution retry seam on the base classes (#94) Add a protected virtual WrapWorkerExecution hook to ExtractorBase, LoaderBase, and TransformerBase, wrapped around every worker invocation (both the no-progress and with-progress paths). The default implementation is a no-op, so behaviour is unchanged; an override receives a re-invocable worker factory (call it again to retry a transient failure) and stream-level semantics are documented on the method. Kept dependency-free per the 0.20.0 decision (Option 1): Abstractions stays zero-dep and roots 8 downstream packages, so no Polly dependency here. A ready-made Polly integration will ship as a separate opt-in Wolfgang.Etl.Polly package (#332). - 13 new RetrySeamTests (default no-op, seam-invoked-once on both paths, factory re-invocability, genuine transient-failure recovery, null guard) across all 3 bases. 426 unit tests pass; Stryker 100.00% (0 survivors) on the full project. - PublicAPI.Unshipped updated with the 3 protected virtual entries (RS0017-validated). - CHANGELOG [Unreleased] Added entry. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 7 + .../ExtractorBase.cs | 41 +- src/Wolfgang.Etl.Abstractions/LoaderBase.cs | 41 +- .../PublicAPI.Unshipped.txt | 3 + .../TransformerBase.cs | 42 +- .../BaseClassTests/RetrySeamTests.cs | 460 ++++++++++++++++++ 6 files changed, 588 insertions(+), 6 deletions(-) create mode 100644 tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/RetrySeamTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index c1aff664..60948d37 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **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 - **Breaking (#285):** `EtlPipelineProgress`'s counters — `ExtractedItemCount`, `LoadedItemCount`, and diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs index 40aff614..66f08364 100644 --- a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs +++ b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs @@ -312,7 +312,7 @@ [EnumeratorCancellation] CancellationToken token { ResetRunState(); - await foreach (var item in ExtractWorkerAsync(token)) + await foreach (var item in WrapWorkerExecution(ExtractWorkerAsync, token)) { yield return item; } @@ -330,7 +330,7 @@ private async IAsyncEnumerable ExtractWithProgressAsync( try { - await foreach (var item in ExtractWorkerAsync(token)) + await foreach (var item in WrapWorkerExecution(ExtractWorkerAsync, token)) { yield return item; } @@ -372,6 +372,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/LoaderBase.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs index 184d9fa1..ca2c40e5 100644 --- a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs +++ b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs @@ -313,7 +313,7 @@ CancellationToken token ) { ResetRunState(); - return LoadWorkerAsync(items, token); + return WrapWorkerExecution(ct => LoadWorkerAsync(items, ct), token); } @@ -329,7 +329,7 @@ private async Task LoadWithProgressAsync( try { - await LoadWorkerAsync(items, token).ConfigureAwait(false); + await WrapWorkerExecution(ct => LoadWorkerAsync(items, ct), token).ConfigureAwait(false); } finally { @@ -370,6 +370,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/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt index bf967293..fa63bcc8 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt +++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt @@ -1,2 +1,5 @@ #nullable enable +virtual Wolfgang.Etl.Abstractions.ExtractorBase.WrapWorkerExecution(System.Func!>! workerFactory, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! +virtual Wolfgang.Etl.Abstractions.LoaderBase.WrapWorkerExecution(System.Func! workerFactory, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! +virtual Wolfgang.Etl.Abstractions.TransformerBase.WrapWorkerExecution(System.Func!>! workerFactory, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! Wolfgang.Etl.Abstractions.Report.Report(int currentItemCount, System.DateTimeOffset? startedAt, System.TimeSpan elapsed, int? totalItemCount = null) -> void diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs index 993f0901..1928435b 100644 --- a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs +++ b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs @@ -320,7 +320,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; } @@ -338,7 +338,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; } @@ -379,6 +379,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/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/RetrySeamTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/RetrySeamTests.cs new file mode 100644 index 00000000..24505ead --- /dev/null +++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/RetrySeamTests.cs @@ -0,0 +1,460 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Wolfgang.Etl.Abstractions; +using Wolfgang.Etl.Abstractions.Tests.Unit.Models; +using Xunit; + +namespace Wolfgang.Etl.Abstractions.Tests.Unit.BaseClassTests; + +/// +/// Covers the #94 retry seam: the WrapWorkerExecution hook on each base class. The default +/// implementation is a no-op (behaviour is unchanged), and an override receives a re-invocable +/// worker factory — calling it again produces a fresh worker run, which is what lets a resilience +/// strategy retry a transient failure. The seam is exercised on both the no-progress and the +/// with-progress worker paths of all three base classes. +/// +public class RetrySeamTests +{ + // ---------- Extractor ---------- + + [Fact] + public async Task Extractor_default_seam_yields_all_items_and_runs_the_worker_once() + { + var sut = new SeamExtractor(new[] { 1, 2, 3 }); + + var items = await Drain(sut.ExtractAsync(CancellationToken.None)); + + Assert.Equal(new[] { 1, 2, 3 }, items); + Assert.Equal(1, sut.WorkerStartCount); // default no-op invokes the factory exactly once + Assert.Equal(1, sut.SeamCallCount); + } + + + [Fact] + public async Task Extractor_override_re_invokes_the_factory_producing_a_fresh_run_each_time() + { + var sut = new SeamExtractor(new[] { 1, 2, 3 }) { WorkerRuns = 3 }; + + var items = await Drain(sut.ExtractAsync(CancellationToken.None)); + + Assert.Equal(new[] { 1, 2, 3 }, items); // only the final run is yielded + Assert.Equal(3, sut.WorkerStartCount); // factory re-invoked -> worker restarted 3 times + Assert.Equal(1, sut.SeamCallCount); // seam wraps the run once + } + + + [Fact] + public async Task Extractor_seam_wraps_the_with_progress_path_too() + { + var sut = new SeamExtractor(new[] { 7, 8 }) { WorkerRuns = 2 }; + var progress = new SynchronousProgress(_ => { }); + + var items = await Drain(sut.ExtractAsync(progress, CancellationToken.None)); + + Assert.Equal(new[] { 7, 8 }, items); + Assert.Equal(2, sut.WorkerStartCount); + Assert.Equal(1, sut.SeamCallCount); + } + + + [Fact] + public async Task Extractor_seam_can_retry_a_transient_failure() + { + // Worker throws on its first start, succeeds on the second; the retrying seam recovers. + var sut = new RetryingExtractor(new[] { 5, 6 }, failuresBeforeSuccess: 1); + + var items = await Drain(sut.ExtractAsync(CancellationToken.None)); + + Assert.Equal(new[] { 5, 6 }, items); + Assert.Equal(2, sut.WorkerStartCount); + } + + + [Fact] + public void Extractor_seam_throws_when_the_factory_is_null() + { + var sut = new SeamExtractor(Array.Empty()); + + Assert.Throws(() => sut.InvokeSeamWithNull()); + } + + + // ---------- Loader ---------- + + [Fact] + public async Task Loader_default_seam_loads_all_items_and_runs_the_worker_once() + { + var sut = new SeamLoader(); + + await sut.LoadAsync(AsyncSource(1, 2, 3), CancellationToken.None); + + Assert.Equal(new[] { 1, 2, 3 }, sut.Loaded); + Assert.Equal(1, sut.WorkerStartCount); + Assert.Equal(1, sut.SeamCallCount); + } + + + [Fact] + public async Task Loader_override_re_invokes_the_factory_producing_a_fresh_run_each_time() + { + var sut = new SeamLoader { WorkerRuns = 3 }; + + await sut.LoadAsync(AsyncSource(1, 2), CancellationToken.None); + + Assert.Equal(3, sut.WorkerStartCount); + Assert.Equal(1, sut.SeamCallCount); + } + + + [Fact] + public async Task Loader_seam_wraps_the_with_progress_path_too() + { + var sut = new SeamLoader { WorkerRuns = 2 }; + var progress = new SynchronousProgress(_ => { }); + + await sut.LoadAsync(AsyncSource(9), progress, CancellationToken.None); + + Assert.Equal(2, sut.WorkerStartCount); + Assert.Equal(1, sut.SeamCallCount); + } + + + [Fact] + public void Loader_seam_throws_when_the_factory_is_null() + { + var sut = new SeamLoader(); + + Assert.Throws(() => { _ = sut.InvokeSeamWithNull(); }); + } + + + // ---------- Transformer ---------- + + [Fact] + public async Task Transformer_default_seam_yields_all_items_and_runs_the_worker_once() + { + var sut = new SeamTransformer(); + + var items = await Drain(sut.TransformAsync(AsyncSource(1, 2, 3), CancellationToken.None)); + + Assert.Equal(new[] { 10, 20, 30 }, items); + Assert.Equal(1, sut.WorkerStartCount); + Assert.Equal(1, sut.SeamCallCount); + } + + + [Fact] + public async Task Transformer_override_re_invokes_the_factory_producing_a_fresh_run_each_time() + { + var sut = new SeamTransformer { WorkerRuns = 3 }; + + var items = await Drain(sut.TransformAsync(AsyncSource(4), CancellationToken.None)); + + Assert.Equal(new[] { 40 }, items); // only the final run is yielded + Assert.Equal(3, sut.WorkerStartCount); + Assert.Equal(1, sut.SeamCallCount); + } + + + [Fact] + public async Task Transformer_seam_wraps_the_with_progress_path_too() + { + var sut = new SeamTransformer { WorkerRuns = 2 }; + var progress = new SynchronousProgress(_ => { }); + + var items = await Drain(sut.TransformAsync(AsyncSource(5), progress, CancellationToken.None)); + + Assert.Equal(new[] { 50 }, items); + Assert.Equal(2, sut.WorkerStartCount); + Assert.Equal(1, sut.SeamCallCount); + } + + + [Fact] + public void Transformer_seam_throws_when_the_factory_is_null() + { + var sut = new SeamTransformer(); + + Assert.Throws(() => sut.InvokeSeamWithNull()); + } + + + // ---------- helpers ---------- + + private static async IAsyncEnumerable AsyncSource(params int[] items) + { + foreach (var item in items) + { + await Task.Yield(); + yield return item; + } + } + + + private static async Task> Drain(IAsyncEnumerable source) + { + var result = new List(); + await foreach (var item in source.ConfigureAwait(false)) + { + result.Add(item); + } + + return result; + } + + + // ---------- doubles ---------- + + // Extractor whose overridden seam invokes the factory (WorkerRuns - 1) discarded times then a + // final real run, proving the factory produces a fresh worker run on each call. WorkerRuns == 1 + // (the default) leaves the seam a pass-through, exercising the base default no-op. + [ExcludeFromCodeCoverage] + private sealed class SeamExtractor : ExtractorBase + { + private readonly int[] _items; + private int _workerStarts; + + public SeamExtractor(int[] items) => _items = items; + + public int WorkerRuns { get; init; } = 1; + + public int WorkerStartCount => Volatile.Read(ref _workerStarts); + + public int SeamCallCount { get; private set; } + + public IAsyncEnumerable InvokeSeamWithNull() => base.WrapWorkerExecution(null!, CancellationToken.None); + + protected override async IAsyncEnumerable ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token) + { + Interlocked.Increment(ref _workerStarts); + foreach (var item in _items) + { + await Task.Yield(); + IncrementCurrentItemCount(); + yield return item; + } + } + + protected override async IAsyncEnumerable WrapWorkerExecution( + Func> workerFactory, + [EnumeratorCancellation] CancellationToken token) + { + SeamCallCount++; + if (WorkerRuns == 1) + { + await foreach (var item in base.WrapWorkerExecution(workerFactory, token).WithCancellation(token)) + { + yield return item; + } + + yield break; + } + + for (var run = 1; run < WorkerRuns; run++) + { + await foreach (var _ in workerFactory(token).WithCancellation(token)) + { + // discard – simulates a failed attempt being retried from scratch + } + } + + await foreach (var item in workerFactory(token).WithCancellation(token)) + { + yield return item; + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + // Extractor whose seam retries a genuine transient failure by re-invoking the factory. + [ExcludeFromCodeCoverage] + private sealed class RetryingExtractor : ExtractorBase + { + private readonly int[] _items; + private readonly int _failuresBeforeSuccess; + private int _workerStarts; + + public RetryingExtractor(int[] items, int failuresBeforeSuccess) + { + _items = items; + _failuresBeforeSuccess = failuresBeforeSuccess; + } + + public int WorkerStartCount => Volatile.Read(ref _workerStarts); + + protected override async IAsyncEnumerable ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token) + { + var attempt = Interlocked.Increment(ref _workerStarts); + await Task.Yield(); + if (attempt <= _failuresBeforeSuccess) + { + throw new InvalidOperationException("transient"); + } + + foreach (var item in _items) + { + yield return item; + } + } + + protected override async IAsyncEnumerable WrapWorkerExecution( + Func> workerFactory, + [EnumeratorCancellation] CancellationToken token) + { + for (var attempt = 1; ; attempt++) + { + var buffer = new List(); + var enumerator = workerFactory(token).GetAsyncEnumerator(token); + var failed = false; + try + { + while (true) + { + try + { + if (!await enumerator.MoveNextAsync().ConfigureAwait(false)) + { + break; + } + } + catch (InvalidOperationException) when (attempt <= _failuresBeforeSuccess) + { + failed = true; + break; + } + + buffer.Add(enumerator.Current); + } + } + finally + { + await enumerator.DisposeAsync().ConfigureAwait(false); + } + + if (failed) + { + continue; + } + + foreach (var item in buffer) + { + yield return item; + } + + yield break; + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + [ExcludeFromCodeCoverage] + private sealed class SeamLoader : LoaderBase + { + private int _workerStarts; + + public List Loaded { get; } = new(); + + public int WorkerRuns { get; init; } = 1; + + public int WorkerStartCount => Volatile.Read(ref _workerStarts); + + public int SeamCallCount { get; private set; } + + public Task InvokeSeamWithNull() => base.WrapWorkerExecution(null!, CancellationToken.None); + + protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token) + { + Interlocked.Increment(ref _workerStarts); + Loaded.Clear(); + await foreach (var item in items.WithCancellation(token).ConfigureAwait(false)) + { + Loaded.Add(item); + IncrementCurrentItemCount(); + } + } + + protected override async Task WrapWorkerExecution( + Func workerFactory, + CancellationToken token) + { + SeamCallCount++; + if (WorkerRuns == 1) + { + await base.WrapWorkerExecution(workerFactory, token).ConfigureAwait(false); + return; + } + + for (var run = 1; run <= WorkerRuns; run++) + { + await workerFactory(token).ConfigureAwait(false); + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + [ExcludeFromCodeCoverage] + private sealed class SeamTransformer : TransformerBase + { + private int _workerStarts; + + public int WorkerRuns { get; init; } = 1; + + public int WorkerStartCount => Volatile.Read(ref _workerStarts); + + public int SeamCallCount { get; private set; } + + public IAsyncEnumerable InvokeSeamWithNull() => base.WrapWorkerExecution(null!, CancellationToken.None); + + protected override async IAsyncEnumerable TransformWorkerAsync( + IAsyncEnumerable items, [EnumeratorCancellation] CancellationToken token) + { + Interlocked.Increment(ref _workerStarts); + await foreach (var item in items.WithCancellation(token).ConfigureAwait(false)) + { + IncrementCurrentItemCount(); + yield return item * 10; + } + } + + protected override async IAsyncEnumerable WrapWorkerExecution( + Func> workerFactory, + [EnumeratorCancellation] CancellationToken token) + { + SeamCallCount++; + if (WorkerRuns == 1) + { + await foreach (var item in base.WrapWorkerExecution(workerFactory, token).WithCancellation(token)) + { + yield return item; + } + + yield break; + } + + for (var run = 1; run < WorkerRuns; run++) + { + await foreach (var _ in workerFactory(token).WithCancellation(token)) + { + // discard + } + } + + await foreach (var item in workerFactory(token).WithCancellation(token)) + { + yield return item; + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } +} From 9db359dbb160b7a2980274ead96498465e9649b7 Mon Sep 17 00:00:00 2001 From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com> Date: Mon, 27 Jul 2026 21:51:41 -0400 Subject: [PATCH 2/7] feat: composable per-item middleware / interceptor (#93) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add IItemMiddleware + MiddlewareResult (Continue/Drop) and WithMiddleware(...) extensions that attach cross-cutting per-item behaviour (logging, validation, metrics, throttling, dedup) to any IAsyncEnumerable stream — extractor/transformer output or loader input — without changing the component, and compose inside an EtlPipeline via Through(s => s.WithMiddleware(...)). Single and ordered-chain overloads; a dropped item short-circuits the rest of the chain. Dependency-free. - 15 MiddlewareTests (transform, drop, chain order, stop-at-drop, empty chain, token flow, pipeline composition, null guards) + MiddlewareResult value semantics. 441 unit tests pass; Stryker 100.00% (0 survivors). - PublicAPI.Unshipped updated (RS0016 completeness + RS0017 correctness validated). - CHANGELOG [Unreleased] Added entry. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 6 + .../IItemMiddleware.cs | 25 ++ .../MiddlewareExtensions.cs | 122 ++++++++ .../MiddlewareResult.Factory.cs | 25 ++ .../MiddlewareResult.cs | 64 +++++ .../PublicAPI.Unshipped.txt | 17 ++ .../MiddlewareTests.cs | 266 ++++++++++++++++++ 7 files changed, 525 insertions(+) create mode 100644 src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs create mode 100644 src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs create mode 100644 src/Wolfgang.Etl.Abstractions/MiddlewareResult.Factory.cs create mode 100644 src/Wolfgang.Etl.Abstractions/MiddlewareResult.cs create mode 100644 tests/Wolfgang.Etl.Abstractions.Tests.Unit/MiddlewareTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 60948d37..4920f5a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **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 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/MiddlewareExtensions.cs b/src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs new file mode 100644 index 00000000..fb4565b1 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs @@ -0,0 +1,122 @@ +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 async IAsyncEnumerable WithMiddleware + ( + this IAsyncEnumerable source, + IItemMiddleware middleware, + [EnumeratorCancellation] CancellationToken token = default + ) + { + if (source is null) + { + throw new ArgumentNullException(nameof(source)); + } + + if (middleware is null) + { + throw new ArgumentNullException(nameof(middleware)); + } + + await foreach (var item in source.WithCancellation(token)) + { + // 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 async IAsyncEnumerable WithMiddleware + ( + this IAsyncEnumerable source, + IEnumerable> middlewares, + [EnumeratorCancellation] CancellationToken token = default + ) + { + if (source is null) + { + throw new ArgumentNullException(nameof(source)); + } + + if (middlewares is null) + { + throw new ArgumentNullException(nameof(middlewares)); + } + + // Snapshot the chain once so the same ordered set runs for every item. + 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."); + } + } + + await foreach (var item in source.WithCancellation(token)) + { + 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.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt index fa63bcc8..e53b6f35 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt +++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt @@ -3,3 +3,20 @@ virtual Wolfgang.Etl.Abstractions.ExtractorBase.WrapWorkerEx virtual Wolfgang.Etl.Abstractions.LoaderBase.WrapWorkerExecution(System.Func! workerFactory, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! virtual Wolfgang.Etl.Abstractions.TransformerBase.WrapWorkerExecution(System.Func!>! workerFactory, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! Wolfgang.Etl.Abstractions.Report.Report(int currentItemCount, System.DateTimeOffset? startedAt, System.TimeSpan elapsed, int? totalItemCount = null) -> void +Wolfgang.Etl.Abstractions.IItemMiddleware +Wolfgang.Etl.Abstractions.IItemMiddleware.OnItemAsync(T item, System.Threading.CancellationToken token) -> System.Threading.Tasks.ValueTask> +Wolfgang.Etl.Abstractions.MiddlewareResult +static Wolfgang.Etl.Abstractions.MiddlewareResult.Continue(T item) -> Wolfgang.Etl.Abstractions.MiddlewareResult +static Wolfgang.Etl.Abstractions.MiddlewareResult.Drop() -> Wolfgang.Etl.Abstractions.MiddlewareResult +Wolfgang.Etl.Abstractions.MiddlewareResult +Wolfgang.Etl.Abstractions.MiddlewareResult.MiddlewareResult() -> void +Wolfgang.Etl.Abstractions.MiddlewareResult.Item.get -> T +Wolfgang.Etl.Abstractions.MiddlewareResult.Skip.get -> bool +Wolfgang.Etl.Abstractions.MiddlewareResult.Equals(Wolfgang.Etl.Abstractions.MiddlewareResult other) -> bool +override Wolfgang.Etl.Abstractions.MiddlewareResult.Equals(object? obj) -> bool +override Wolfgang.Etl.Abstractions.MiddlewareResult.GetHashCode() -> int +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 +Wolfgang.Etl.Abstractions.MiddlewareExtensions +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.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! diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/MiddlewareTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/MiddlewareTests.cs new file mode 100644 index 00000000..92ae8342 --- /dev/null +++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/MiddlewareTests.cs @@ -0,0 +1,266 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Wolfgang.Etl.Abstractions; +using Xunit; + +namespace Wolfgang.Etl.Abstractions.Tests.Unit; + +/// +/// Covers the #93 middleware mechanism: , +/// , and the WithMiddleware stream decorators — including +/// composition inside an Through stage. +/// +public class MiddlewareTests +{ + // ---------- MiddlewareResult value type ---------- + + [Fact] + public void MiddlewareResult_Continue_keeps_the_item() + { + var result = MiddlewareResult.Continue(42); + + Assert.False(result.Skip); + Assert.Equal(42, result.Item); + } + + + [Fact] + public void MiddlewareResult_Drop_marks_the_item_skipped() + { + var result = MiddlewareResult.Drop(); + + Assert.True(result.Skip); + } + + + [Fact] + public void MiddlewareResult_has_value_equality() + { + var a = MiddlewareResult.Continue(7); + var b = MiddlewareResult.Continue(7); + var different = MiddlewareResult.Continue(8); + var dropped = MiddlewareResult.Drop(); + + Assert.Equal(a, b); + Assert.True(a == b); + Assert.False(a != b); + Assert.Equal(a.GetHashCode(), b.GetHashCode()); + Assert.NotEqual(a, different); + Assert.True(a != different); + Assert.NotEqual(a, dropped); + } + + + // ---------- single middleware ---------- + + [Fact] + public async Task WithMiddleware_transforms_each_item() + { + var items = await Drain(AsyncSource(1, 2, 3).WithMiddleware(new TimesTenMiddleware())); + + Assert.Equal(new[] { 10, 20, 30 }, items); + } + + + [Fact] + public async Task WithMiddleware_drops_items_the_middleware_skips() + { + var items = await Drain(AsyncSource(1, 2, 3, 4).WithMiddleware(new DropOddMiddleware())); + + Assert.Equal(new[] { 2, 4 }, items); + } + + + [Fact] + public async Task WithMiddleware_flows_the_cancellation_token_to_the_middleware() + { + using var cts = new CancellationTokenSource(); + var capturing = new TokenCapturingMiddleware(); + + await Drain(AsyncSource(1).WithMiddleware(capturing), cts.Token); + + Assert.Equal(cts.Token, capturing.LastToken); + } + + + [Fact] + public async Task WithMiddleware_composes_inside_an_EtlPipeline_Through_stage() + { + var stream = EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3, 4)) + .Through(s => s.WithMiddleware(new DropOddMiddleware())) + .Through(s => s.WithMiddleware(new TimesTenMiddleware())) + .AsAsyncEnumerable(); + + var items = await Drain(stream); + + Assert.Equal(new[] { 20, 40 }, items); + } + + + [Fact] + public async Task WithMiddleware_when_source_is_null_throws_ArgumentNullException() + { + await Assert.ThrowsAsync( + () => Drain(((IAsyncEnumerable)null!).WithMiddleware(new TimesTenMiddleware()))); + } + + + [Fact] + public async Task WithMiddleware_when_middleware_is_null_throws_ArgumentNullException() + { + await Assert.ThrowsAsync( + () => Drain(AsyncSource(1).WithMiddleware((IItemMiddleware)null!))); + } + + + // ---------- middleware chain ---------- + + [Fact] + public async Task WithMiddleware_chain_runs_in_registration_order_each_seeing_the_previous_output() + { + var log = new List(); + var chain = new IItemMiddleware[] + { + new RecordingMiddleware(log, "A", add: 10), + new RecordingMiddleware(log, "B", add: 100), + }; + + var items = await Drain(AsyncSource(1).WithMiddleware(chain)); + + Assert.Equal(new[] { 111 }, items); // 1 -> +10 -> +100 + Assert.Equal(new[] { "A:1", "B:11" }, log); // B saw A's output + } + + + [Fact] + public async Task WithMiddleware_chain_stops_at_the_first_drop() + { + var log = new List(); + var chain = new IItemMiddleware[] + { + new DropOddMiddleware(), + new RecordingMiddleware(log, "R", add: 0), + }; + + var items = await Drain(AsyncSource(1, 2, 3).WithMiddleware(chain)); + + Assert.Equal(new[] { 2 }, items); // odds dropped before reaching R + Assert.Equal(new[] { "R:2" }, log); // R only ran for the surviving even item + } + + + [Fact] + public async Task WithMiddleware_empty_chain_passes_items_through() + { + var items = await Drain(AsyncSource(1, 2, 3).WithMiddleware(Array.Empty>())); + + Assert.Equal(new[] { 1, 2, 3 }, items); + } + + + [Fact] + public async Task WithMiddleware_chain_when_source_is_null_throws_ArgumentNullException() + { + await Assert.ThrowsAsync( + () => Drain(((IAsyncEnumerable)null!).WithMiddleware(new IItemMiddleware[] { new TimesTenMiddleware() }))); + } + + + [Fact] + public async Task WithMiddleware_chain_when_middlewares_is_null_throws_ArgumentNullException() + { + var ex = await Assert.ThrowsAsync( + () => Drain(AsyncSource(1).WithMiddleware((IEnumerable>)null!))); + + // The explicit guard names "middlewares"; without it the fallback List ctor would name "collection". + Assert.Equal("middlewares", ex.ParamName); + } + + + [Fact] + public async Task WithMiddleware_chain_when_a_member_is_null_throws_ArgumentNullException() + { + var chain = new IItemMiddleware[] { new TimesTenMiddleware(), null! }; + + await Assert.ThrowsAsync( + () => Drain(AsyncSource(1).WithMiddleware(chain))); + } + + + // ---------- helpers ---------- + + private static async IAsyncEnumerable AsyncSource(params int[] items) + { + foreach (var item in items) + { + await Task.Yield(); + yield return item; + } + } + + + private static async Task> Drain(IAsyncEnumerable source, CancellationToken token = default) + { + var result = new List(); + await foreach (var item in source.WithCancellation(token).ConfigureAwait(false)) + { + result.Add(item); + } + + return result; + } + + + // ---------- doubles ---------- + + private sealed class TimesTenMiddleware : IItemMiddleware + { + public ValueTask> OnItemAsync(int item, CancellationToken token) => + new(MiddlewareResult.Continue(item * 10)); + } + + + private sealed class DropOddMiddleware : IItemMiddleware + { + public ValueTask> OnItemAsync(int item, CancellationToken token) => + new(item % 2 == 0 ? MiddlewareResult.Continue(item) : MiddlewareResult.Drop()); + } + + + private sealed class TokenCapturingMiddleware : IItemMiddleware + { + public CancellationToken LastToken { get; private set; } + + public ValueTask> OnItemAsync(int item, CancellationToken token) + { + LastToken = token; + return new(MiddlewareResult.Continue(item)); + } + } + + + private sealed class RecordingMiddleware : IItemMiddleware + { + private readonly List _log; + private readonly string _name; + private readonly int _add; + + public RecordingMiddleware(List log, string name, int add) + { + _log = log; + _name = name; + _add = add; + } + + public ValueTask> OnItemAsync(int item, CancellationToken token) + { + _log.Add($"{_name}:{item}"); + return new(MiddlewareResult.Continue(item + _add)); + } + } +} From 31a25f0adb081d269d33817a93c8e57cf918a4b9 Mon Sep 17 00:00:00 2001 From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:14:17 -0400 Subject: [PATCH 3/7] feat: aggregate per-item error counts across all pipeline stages (#335) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix the base-class asymmetry: CurrentErrorItemCount lived only on the concrete base classes, so EtlPipelineProgress.ErrorItemCount could only surface the extractor's errors (Through takes ITransformAsync, an interface that didn't expose it). - Add IReportsItemErrors { int CurrentErrorItemCount { get; } }, implemented by ExtractorBase, LoaderBase, TransformerBase (property already existed — additive). - EtlRunState now holds a list of error-count readers and sums them in Snapshot(). - Register a reader from every stage that reports errors: the extractor (From), each transformer (Through, via 'is IReportsItemErrors'), and the loader (sink). ErrorItemCount now reflects items ANY stage's error policy discarded. Pre-1.0 behaviour change. - 5 AggregateErrorsTests (sum across all three stages, delegate stage contributes 0, zero when nothing reports, plain-ITransformAsync overload, interface implemented by all bases). 446 unit tests pass; Stryker 100.00% (0 survivors). - PublicAPI.Unshipped updated (RS0016 completeness + RS0017 correctness validated). - CHANGELOG [Unreleased] Added + Changed entries. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 7 + .../EtlPipeline/EtlPipelineImpl.cs | 24 +- .../EtlPipeline/EtlPipelineSink.cs | 5 + .../EtlPipelineSourceExtensions.cs | 3 +- .../EtlPipeline/EtlRunState.cs | 20 +- .../ExtractorBase.cs | 1 + .../IReportsItemErrors.cs | 17 ++ src/Wolfgang.Etl.Abstractions/LoaderBase.cs | 1 + .../PublicAPI.Unshipped.txt | 2 + .../TransformerBase.cs | 1 + .../EtlPipelineTests/AggregateErrorsTests.cs | 251 ++++++++++++++++++ 11 files changed, 325 insertions(+), 7 deletions(-) create mode 100644 src/Wolfgang.Etl.Abstractions/IReportsItemErrors.cs create mode 100644 tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/AggregateErrorsTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 4920f5a2..e94c45fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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 @@ -24,6 +27,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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. - **Breaking (#285):** `EtlPipelineProgress`'s counters — `ExtractedItemCount`, `LoadedItemCount`, and `ErrorItemCount` — are now `long` instead of `int`, so a long-running pipeline can report more than `int.MaxValue` (~2.1 billion) records without overflow. This changes the record's getters, positional 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 9c8c7a42..c07c4bde 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; using System.Diagnostics; @@ -18,16 +19,27 @@ 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 EtlPipelineProgress Snapshot() { + long errorItemCount = 0; + foreach (var reader in _errorCountReaders) + { + errorItemCount += reader(); + } + return new EtlPipelineProgress(ExtractedItemCount, LoadedItemCount, _stopwatch.Elapsed) { - ErrorItemCount = ErrorCountReader?.Invoke() ?? 0, + ErrorItemCount = errorItemCount, }; } } diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs index 66f08364..7fb1b76e 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 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 ca2c40e5..3b1e8ad6 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 diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt index e53b6f35..77924326 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt +++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt @@ -20,3 +20,5 @@ static Wolfgang.Etl.Abstractions.MiddlewareResult.operator !=(Wolfgang.Etl.Ab Wolfgang.Etl.Abstractions.MiddlewareExtensions 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.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! +Wolfgang.Etl.Abstractions.IReportsItemErrors +Wolfgang.Etl.Abstractions.IReportsItemErrors.CurrentErrorItemCount.get -> int diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs index 1928435b..9c92e7e4 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 diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/AggregateErrorsTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/AggregateErrorsTests.cs new file mode 100644 index 00000000..cf58f441 --- /dev/null +++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/AggregateErrorsTests.cs @@ -0,0 +1,251 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Wolfgang.Etl.Abstractions; +using Wolfgang.Etl.Abstractions.Tests.Unit.Models; +using Xunit; + +namespace Wolfgang.Etl.Abstractions.Tests.Unit.EtlPipelineTests; + +/// +/// Covers #335: aggregates the per-item error counts +/// of every stage that reports them (source, transformers, sink) via , +/// not just the extractor. +/// +public class AggregateErrorsTests +{ + [Fact] + public void All_three_base_classes_implement_IReportsItemErrors() + { + Assert.IsAssignableFrom(new ErroringExtractor(good: 0, errors: 0)); + Assert.IsAssignableFrom(new ErroringTransformer(errors: 0)); + Assert.IsAssignableFrom(new ErroringLoader(errors: 0)); + } + + + [Fact] + public async Task ErrorItemCount_sums_extractor_transformer_and_loader_errors() + { + var reports = new List(); + var extractor = new ErroringExtractor(good: 3, errors: 2); + var transformer = new ErroringTransformer(errors: 4); + var loader = new ErroringLoader(errors: 1); + + await EtlPipeline + .Create() + .From(extractor) + .Through(transformer) + .To(loader) + .RunAsync(new SyncProgress(reports.Add)); + + var final = reports[^1]; + Assert.Equal(3, final.ExtractedItemCount); + Assert.Equal(3, final.LoadedItemCount); + Assert.Equal(7, final.ErrorItemCount); // 2 (extract) + 4 (transform) + 1 (load) + } + + + [Fact] + public async Task ErrorItemCount_counts_only_stages_that_report_errors() + { + // A delegate transform stage does not implement IReportsItemErrors, so it contributes nothing. + var reports = new List(); + var extractor = new ErroringExtractor(good: 2, errors: 5); + var loader = new ErroringLoader(errors: 3); + + await EtlPipeline + .Create() + .From(extractor) + .Through(s => s) + .To(loader) + .RunAsync(new SyncProgress(reports.Add)); + + Assert.Equal(8, reports[^1].ErrorItemCount); // 5 (extract) + 3 (load); delegate adds 0 + } + + + [Fact] + public async Task ErrorItemCount_includes_a_plain_ITransformAsync_stage_that_reports_errors() + { + // Binds the non-cancellation Through overload (a bare ITransformAsync, not a TransformerBase). + var reports = new List(); + var extractor = new ErroringExtractor(good: 2, errors: 1); + var transform = new ReportingPassThroughTransform(errors: 5); + var loader = new ErroringLoader(errors: 0); + + await EtlPipeline + .Create() + .From(extractor) + .Through(transform) + .To(loader) + .RunAsync(new SyncProgress(reports.Add)); + + Assert.Equal(6, reports[^1].ErrorItemCount); // 1 (extract) + 5 (plain transform) + } + + + [Fact] + public async Task ErrorItemCount_is_zero_when_no_stage_reports_errors() + { + var reports = new List(); + var loader = new ErroringLoader(errors: 0); + + await EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3)) // raw IAsyncEnumerable — not an IReportsItemErrors stage + .To(loader) + .RunAsync(new SyncProgress(reports.Add)); + + Assert.Equal(0, reports[^1].ErrorItemCount); + } + + + // ---------- helpers ---------- + + private static async IAsyncEnumerable AsyncSource(params int[] items) + { + foreach (var item in items) + { + await Task.Yield(); + yield return item; + } + } + + + // A bare ITransformAsync (not a TransformerBase) that still reports an error count — exercises the + // non-cancellation Through overload's error-reader registration. + private sealed class ReportingPassThroughTransform : ITransformAsync, IReportsItemErrors + { + public ReportingPassThroughTransform(int errors) => CurrentErrorItemCount = errors; + + public int CurrentErrorItemCount { get; } + + public IAsyncEnumerable TransformAsync(IAsyncEnumerable items) => items; + } + + + private sealed class SyncProgress : IProgress + { + private readonly Action _report; + + public SyncProgress(Action report) => _report = report; + + public void Report(EtlPipelineProgress value) => _report(value); + } + + + // ---------- error-reporting doubles (route real #84 errors through HandleItemError) ---------- + + private sealed class ErroringExtractor : ExtractorBase + { + private readonly int _good; + private readonly int _errors; + + public ErroringExtractor(int good, int errors) + { + _good = good; + _errors = errors; + } + + protected override ItemErrorAction OnItemError(ItemErrorContext context) => ItemErrorAction.Skip; + + protected override async IAsyncEnumerable ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token) + { + for (var i = 0; i < _good; i++) + { + await Task.Yield(); + IncrementCurrentItemCount(); + yield return i; + } + + for (var e = 0; e < _errors; e++) + { + // Route a synthetic failure through the base #84 hook: with OnItemError => Skip, + // HandleItemError increments CurrentErrorItemCount and returns without rethrowing. + try + { + throw new InvalidOperationException("bad item"); + } + catch (InvalidOperationException ex) + { + HandleItemError(new ItemErrorContext(e, ex)); + } + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + private sealed class ErroringTransformer : TransformerBase + { + private readonly int _errors; + + public ErroringTransformer(int errors) => _errors = errors; + + protected override ItemErrorAction OnItemError(ItemErrorContext context) => ItemErrorAction.Skip; + + protected override async IAsyncEnumerable TransformWorkerAsync( + IAsyncEnumerable items, [EnumeratorCancellation] CancellationToken token) + { + await foreach (var item in items.WithCancellation(token)) + { + IncrementCurrentItemCount(); + yield return item; + } + + for (var e = 0; e < _errors; e++) + { + // Route a synthetic failure through the base #84 hook: with OnItemError => Skip, + // HandleItemError increments CurrentErrorItemCount and returns without rethrowing. + try + { + throw new InvalidOperationException("bad item"); + } + catch (InvalidOperationException ex) + { + HandleItemError(new ItemErrorContext(e, ex)); + } + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + private sealed class ErroringLoader : LoaderBase + { + private readonly int _errors; + + public ErroringLoader(int errors) => _errors = errors; + + protected override ItemErrorAction OnItemError(ItemErrorContext context) => ItemErrorAction.Skip; + + protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token) + { + await foreach (var item in items.WithCancellation(token)) + { + IncrementCurrentItemCount(); + } + + for (var e = 0; e < _errors; e++) + { + // Route a synthetic failure through the base #84 hook: with OnItemError => Skip, + // HandleItemError increments CurrentErrorItemCount and returns without rethrowing. + try + { + throw new InvalidOperationException("bad item"); + } + catch (InvalidOperationException ex) + { + HandleItemError(new ItemErrorContext(e, ex)); + } + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } +} From bd3ae7027834e4eb49f5462d372ec9c6f6ca4630 Mon Sep 17 00:00:00 2001 From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com> Date: Tue, 28 Jul 2026 12:33:21 -0400 Subject: [PATCH 4/7] chore: bump vNext-plus-one to 0.20.0 Reconcile complete: the 0.20 feature stack (#94 retry seam / #93 middleware / #335 error aggregation) merged in, plus vNext's 0.19.0 prep + #338 clock seam merged down. CHANGELOG [Unreleased] holds the 0.20 features; [0.19.0] section carried in. PublicAPI merged (features in Unshipped, Report ctor promoted to Shipped). 452 tests pass; all 11 TFMs clean; pack + Package Validation green vs 0.18.1. Co-Authored-By: Claude Opus 4.8 --- src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj index 6bb45565..d8b0bdf3 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 - - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net10.0/Wolfgang.Etl.Abstractions.dll - lib/net10.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net10.0/Wolfgang.Etl.Abstractions.dll - lib/net10.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net10.0/Wolfgang.Etl.Abstractions.dll - lib/net10.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net10.0/Wolfgang.Etl.Abstractions.dll - lib/net10.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net10.0/Wolfgang.Etl.Abstractions.dll - lib/net10.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net462/Wolfgang.Etl.Abstractions.dll - lib/net462/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net462/Wolfgang.Etl.Abstractions.dll - lib/net462/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net462/Wolfgang.Etl.Abstractions.dll - lib/net462/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net462/Wolfgang.Etl.Abstractions.dll - lib/net462/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net462/Wolfgang.Etl.Abstractions.dll - lib/net462/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net472/Wolfgang.Etl.Abstractions.dll - lib/net472/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net472/Wolfgang.Etl.Abstractions.dll - lib/net472/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net472/Wolfgang.Etl.Abstractions.dll - lib/net472/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net472/Wolfgang.Etl.Abstractions.dll - lib/net472/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net472/Wolfgang.Etl.Abstractions.dll - lib/net472/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net48/Wolfgang.Etl.Abstractions.dll - lib/net48/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net48/Wolfgang.Etl.Abstractions.dll - lib/net48/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net48/Wolfgang.Etl.Abstractions.dll - lib/net48/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net48/Wolfgang.Etl.Abstractions.dll - lib/net48/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net48/Wolfgang.Etl.Abstractions.dll - lib/net48/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net481/Wolfgang.Etl.Abstractions.dll - lib/net481/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net481/Wolfgang.Etl.Abstractions.dll - lib/net481/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net481/Wolfgang.Etl.Abstractions.dll - lib/net481/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net481/Wolfgang.Etl.Abstractions.dll - lib/net481/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net481/Wolfgang.Etl.Abstractions.dll - lib/net481/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net5.0/Wolfgang.Etl.Abstractions.dll - lib/net5.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net5.0/Wolfgang.Etl.Abstractions.dll - lib/net5.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net5.0/Wolfgang.Etl.Abstractions.dll - lib/net5.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net5.0/Wolfgang.Etl.Abstractions.dll - lib/net5.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net5.0/Wolfgang.Etl.Abstractions.dll - lib/net5.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net6.0/Wolfgang.Etl.Abstractions.dll - lib/net6.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net6.0/Wolfgang.Etl.Abstractions.dll - lib/net6.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net6.0/Wolfgang.Etl.Abstractions.dll - lib/net6.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net6.0/Wolfgang.Etl.Abstractions.dll - lib/net6.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net6.0/Wolfgang.Etl.Abstractions.dll - lib/net6.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net7.0/Wolfgang.Etl.Abstractions.dll - lib/net7.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net7.0/Wolfgang.Etl.Abstractions.dll - lib/net7.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net7.0/Wolfgang.Etl.Abstractions.dll - lib/net7.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net7.0/Wolfgang.Etl.Abstractions.dll - lib/net7.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net7.0/Wolfgang.Etl.Abstractions.dll - lib/net7.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net8.0/Wolfgang.Etl.Abstractions.dll - lib/net8.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net8.0/Wolfgang.Etl.Abstractions.dll - lib/net8.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net8.0/Wolfgang.Etl.Abstractions.dll - lib/net8.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net8.0/Wolfgang.Etl.Abstractions.dll - lib/net8.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net8.0/Wolfgang.Etl.Abstractions.dll - lib/net8.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/net9.0/Wolfgang.Etl.Abstractions.dll - lib/net9.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/net9.0/Wolfgang.Etl.Abstractions.dll - lib/net9.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/net9.0/Wolfgang.Etl.Abstractions.dll - lib/net9.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/net9.0/Wolfgang.Etl.Abstractions.dll - lib/net9.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/net9.0/Wolfgang.Etl.Abstractions.dll - lib/net9.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.#ctor(System.Int32,System.Int32,System.TimeSpan) - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.Deconstruct(System.Int32@,System.Int32@,System.TimeSpan@) - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ErrorItemCount - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_ExtractedItemCount - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - true - - - CP0002 - M:Wolfgang.Etl.Abstractions.EtlPipelineProgress.get_LoadedItemCount - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - lib/netstandard2.0/Wolfgang.Etl.Abstractions.dll - true - - \ No newline at end of file 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/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt index 59ddb488..7dc5c581 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt +++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt @@ -1,23 +1 @@ #nullable enable -virtual Wolfgang.Etl.Abstractions.ExtractorBase.WrapWorkerExecution(System.Func!>! workerFactory, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! -virtual Wolfgang.Etl.Abstractions.LoaderBase.WrapWorkerExecution(System.Func! workerFactory, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! -virtual Wolfgang.Etl.Abstractions.TransformerBase.WrapWorkerExecution(System.Func!>! workerFactory, System.Threading.CancellationToken token) -> 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.MiddlewareResult -static Wolfgang.Etl.Abstractions.MiddlewareResult.Continue(T item) -> Wolfgang.Etl.Abstractions.MiddlewareResult -static Wolfgang.Etl.Abstractions.MiddlewareResult.Drop() -> Wolfgang.Etl.Abstractions.MiddlewareResult -Wolfgang.Etl.Abstractions.MiddlewareResult -Wolfgang.Etl.Abstractions.MiddlewareResult.MiddlewareResult() -> void -Wolfgang.Etl.Abstractions.MiddlewareResult.Item.get -> T -Wolfgang.Etl.Abstractions.MiddlewareResult.Skip.get -> bool -Wolfgang.Etl.Abstractions.MiddlewareResult.Equals(Wolfgang.Etl.Abstractions.MiddlewareResult other) -> bool -override Wolfgang.Etl.Abstractions.MiddlewareResult.Equals(object? obj) -> bool -override Wolfgang.Etl.Abstractions.MiddlewareResult.GetHashCode() -> int -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 -Wolfgang.Etl.Abstractions.MiddlewareExtensions -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.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! -Wolfgang.Etl.Abstractions.IReportsItemErrors -Wolfgang.Etl.Abstractions.IReportsItemErrors.CurrentErrorItemCount.get -> int diff --git a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj index d8b0bdf3..d0fdbebd 100644 --- a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj +++ b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj @@ -20,7 +20,7 @@ recorded in CompatibilitySuppressions.xml, regenerated with `dotnet pack /p:GenerateCompatibilitySuppressionFile=true`. --> true - 0.18.1 + 0.19.0 False $(AssemblyName) Contains interfaces and base classes used to build ETL applications