refactor: EtlPipeline.From as extension method + fix InspectCode (MA0049/S2325)

src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSinkExtensions.cs

@@ -0,0 +1,98 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// Extensions for terminal <see cref="IEtlPipelineSink"/> pipelines.
+/// </summary>
+public static class EtlPipelineSinkExtensions
+{
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <param name="sink">The sink to wrap.</param>
+    /// <param name="ownedResources">
+    /// The resources to dispose after the run. Each is disposed via <see cref="IAsyncDisposable"/>
+    /// when supported, otherwise <see cref="IDisposable"/>; anything else (or <see langword="null"/>)
+    /// is skipped. Resources are disposed in reverse of the order supplied (LIFO).
+    /// </param>
+    /// <returns>
+    /// A sink that disposes <paramref name="ownedResources"/> after running, or the original
+    /// <paramref name="sink"/> unchanged when no resources are supplied.
+    /// </returns>
+    /// <exception cref="ArgumentNullException"><paramref name="sink"/> is <see langword="null"/>.</exception>
+    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<EtlPipelineProgress>? 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;
+                }
+            }
+        }
+    }
+}

src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt

@@ -170,3 +170,5 @@ virtual Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgre
 virtual Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.TransformAsync(System.Collections.Generic.IAsyncEnumerable<TSource>! items, System.IProgress<TProgress>! progress) -> System.Collections.Generic.IAsyncEnumerable<TDestination>!
 virtual Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.TransformAsync(System.Collections.Generic.IAsyncEnumerable<TSource>! items, System.IProgress<TProgress>! progress, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable<TDestination>!
 virtual Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.TransformAsync(System.Collections.Generic.IAsyncEnumerable<TSource>! items, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable<TDestination>!
+Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions
+static Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions.DisposingOwned(this Wolfgang.Etl.Abstractions.IEtlPipelineSink! sink, params object![]! ownedResources) -> Wolfgang.Etl.Abstractions.IEtlPipelineSink!

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<Task>? _onRun;
+
+        public FakeSink(Func<Task>? onRun = null) => _onRun = onRun;
+
+        public bool Ran { get; private set; }
+
+        public async Task RunAsync(IProgress<EtlPipelineProgress>? progress = null, CancellationToken token = default)
+        {
+            Ran = true;
+            if (_onRun is not null)
+            {
+                await _onRun().ConfigureAwait(false);
+            }
+        }
+    }
+
+
+    private sealed class TrackingDisposable : IDisposable
+    {
+        private readonly List<string> _log;
+        private readonly string _name;
+
+        public TrackingDisposable(List<string> log, string name)
+        {
+            _log = log;
+            _name = name;
+        }
+
+        public void Dispose() => _log.Add(_name);
+    }
+
+
+    private sealed class TrackingAsyncDisposable : IAsyncDisposable
+    {
+        private readonly List<string> _log;
+        private readonly string _name;
+
+        public TrackingAsyncDisposable(List<string> 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<string>();
+        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<string>();
+        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<string>();
+        var sink = new FakeSink(() => throw new InvalidOperationException("boom"));
+
+        await Assert.ThrowsAsync<InvalidOperationException>(
+            () => 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<ArgumentNullException>(
+            () => ((IEtlPipelineSink)null!).DisposingOwned(new object()));
+    }
+}