From 28ac55033324f45d92eab42620a54a86fe234ad7 Mon Sep 17 00:00:00 2001 From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com> Date: Mon, 20 Jul 2026 13:48:26 -0400 Subject: [PATCH] feat: add IEtlPipelineSink.DisposingOwned for factory-owned resource cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Class-named source/sink factories in format packages open files/connections the caller did not hand them. DisposingOwned wraps a sink to dispose those factory-owned resources after RunAsync (success or failure), preferring IAsyncDisposable and falling back to IDisposable, in reverse (LIFO) order — so packages don't each reimplement the decorator. Co-Authored-By: Claude Opus 4.8 --- .../EtlPipeline/EtlPipelineSinkExtensions.cs | 98 ++++++++++++++ .../PublicAPI.Shipped.txt | 2 + .../EtlPipelineSinkExtensionsTests.cs | 121 ++++++++++++++++++ 3 files changed, 221 insertions(+) create mode 100644 src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSinkExtensions.cs create mode 100644 tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineSinkExtensionsTests.cs diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSinkExtensions.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSinkExtensions.cs new file mode 100644 index 00000000..ed496fc0 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSinkExtensions.cs @@ -0,0 +1,98 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + + +namespace Wolfgang.Etl.Abstractions; + +/// +/// Extensions for terminal pipelines. +/// +public static class EtlPipelineSinkExtensions +{ + /// + /// Wraps a sink so that the given resources — opened by the code that built the pipeline (for + /// example a file stream a path-based loader factory created) — are disposed after the run + /// completes, whether it succeeds or throws. + /// + /// The sink to wrap. + /// + /// The resources to dispose after the run. Each is disposed via + /// when supported, otherwise ; anything else (or ) + /// is skipped. Resources are disposed in reverse of the order supplied (LIFO). + /// + /// + /// A sink that disposes after running, or the original + /// unchanged when no resources are supplied. + /// + /// is . + public static IEtlPipelineSink DisposingOwned + ( + this IEtlPipelineSink sink, + params object[] ownedResources + ) + { + if (sink is null) + { + throw new ArgumentNullException(nameof(sink)); + } + + if (ownedResources is null || ownedResources.Length == 0) + { + return sink; + } + + return new DisposingSink(sink, ownedResources); + } + + + private sealed class DisposingSink : IEtlPipelineSink + { + private readonly IEtlPipelineSink _inner; + private readonly object[] _ownedResources; + + + internal DisposingSink(IEtlPipelineSink inner, object[] ownedResources) + { + _inner = inner; + _ownedResources = ownedResources; + } + + + public async Task RunAsync + ( + IProgress? progress = null, + CancellationToken token = default + ) + { + try + { + await _inner.RunAsync(progress, token).ConfigureAwait(false); + } + finally + { + await DisposeOwnedAsync().ConfigureAwait(false); + } + } + + + // Reverse (LIFO) order, matching nested using / DI-scope disposal. IAsyncDisposable is + // preferred where the concrete resource implements it (e.g. FileStream on net5.0+); on the + // older frameworks the same FileStream matches only IDisposable, so it disposes synchronously. + private async Task DisposeOwnedAsync() + { + for (var i = _ownedResources.Length - 1; i >= 0; i--) + { + switch (_ownedResources[i]) + { + case IAsyncDisposable asyncDisposable: + await asyncDisposable.DisposeAsync().ConfigureAwait(false); + break; + case IDisposable disposable: + disposable.Dispose(); + break; + } + } + } + } +} diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt index 71f7efaa..88c5115f 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt +++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt @@ -170,3 +170,5 @@ 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! +Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions +static Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions.DisposingOwned(this Wolfgang.Etl.Abstractions.IEtlPipelineSink! sink, params object![]! ownedResources) -> Wolfgang.Etl.Abstractions.IEtlPipelineSink! diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineSinkExtensionsTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineSinkExtensionsTests.cs new file mode 100644 index 00000000..88864d08 --- /dev/null +++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineSinkExtensionsTests.cs @@ -0,0 +1,121 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace Wolfgang.Etl.Abstractions.Tests.Unit.EtlPipelineTests; + +public sealed class EtlPipelineSinkExtensionsTests +{ + private sealed class FakeSink : IEtlPipelineSink + { + private readonly Func? _onRun; + + public FakeSink(Func? onRun = null) => _onRun = onRun; + + public bool Ran { get; private set; } + + public async Task RunAsync(IProgress? progress = null, CancellationToken token = default) + { + Ran = true; + if (_onRun is not null) + { + await _onRun().ConfigureAwait(false); + } + } + } + + + private sealed class TrackingDisposable : IDisposable + { + private readonly List _log; + private readonly string _name; + + public TrackingDisposable(List log, string name) + { + _log = log; + _name = name; + } + + public void Dispose() => _log.Add(_name); + } + + + private sealed class TrackingAsyncDisposable : IAsyncDisposable + { + private readonly List _log; + private readonly string _name; + + public TrackingAsyncDisposable(List log, string name) + { + _log = log; + _name = name; + } + + public ValueTask DisposeAsync() + { + _log.Add(_name); + return default; + } + } + + + [Fact] + public async Task DisposingOwned_disposes_resources_after_a_successful_run() + { + var log = new List(); + var sink = new FakeSink(); + + await sink.DisposingOwned(new TrackingDisposable(log, "a")).RunAsync(); + + Assert.True(sink.Ran); + Assert.Equal(new[] { "a" }, log); + } + + + [Fact] + public async Task DisposingOwned_disposes_in_reverse_order() + { + var log = new List(); + var sink = new FakeSink(); + + await sink + .DisposingOwned( + new TrackingDisposable(log, "first"), + new TrackingAsyncDisposable(log, "second")) + .RunAsync(); + + Assert.Equal(new[] { "second", "first" }, log); + } + + + [Fact] + public async Task DisposingOwned_disposes_even_when_the_run_throws() + { + var log = new List(); + var sink = new FakeSink(() => throw new InvalidOperationException("boom")); + + await Assert.ThrowsAsync( + () => sink.DisposingOwned(new TrackingDisposable(log, "a")).RunAsync()); + + Assert.Equal(new[] { "a" }, log); + } + + + [Fact] + public void DisposingOwned_with_no_resources_returns_the_same_sink() + { + var sink = new FakeSink(); + + Assert.Same(sink, sink.DisposingOwned()); + } + + + [Fact] + public void DisposingOwned_null_sink_throws() + { + Assert.Throws( + () => ((IEtlPipelineSink)null!).DisposingOwned(new object())); + } +}