From 6efc06900a6c350e105d97c4071198b2a141b4b9 Mon Sep 17 00:00:00 2001 From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com> Date: Mon, 20 Jul 2026 19:30:10 -0400 Subject: [PATCH] docs: update EtlPipeline docs for Create()/DisposingOwned + add Example9 - Fix stale getting-started/introduction: EtlPipeline.From(...) is now EtlPipeline.Create().From(...) (From is an instance extension), and the removed EtlPipeline.Source sentinel is replaced with the extension-method description. - Document IEtlPipelineSink.DisposingOwned for factory-owned resource cleanup. - Add Example9-DisposingOwned (net4.8 + net8.0): a pipeline whose sink writes to a caller-opened file, with DisposingOwned closing it after RunAsync. Verified running on both TFMs. Co-Authored-By: Claude Opus 4.8 --- docfx_project/docs/getting-started.md | 51 ++++++++++++--- docfx_project/docs/introduction.md | 2 +- .../Example9-DisposingOwned/ConsoleColors.cs | 11 ++++ .../ETL/FileLineLoader.cs | 41 ++++++++++++ .../Example9-DisposingOwned.csproj | 26 ++++++++ .../Net4.8/Example9-DisposingOwned/Program.cs | 62 +++++++++++++++++++ .../Properties/AssemblyInfo.cs | 32 ++++++++++ .../Example9-DisposingOwned/ConsoleColors.cs | 8 +++ .../ETL/FileLineLoader.cs | 28 +++++++++ .../Example9-DisposingOwned.csproj | 19 ++++++ .../Net8.0/Example9-DisposingOwned/Program.cs | 56 +++++++++++++++++ 11 files changed, 327 insertions(+), 9 deletions(-) create mode 100644 examples/Net4.8/Example9-DisposingOwned/ConsoleColors.cs create mode 100644 examples/Net4.8/Example9-DisposingOwned/ETL/FileLineLoader.cs create mode 100644 examples/Net4.8/Example9-DisposingOwned/Example9-DisposingOwned.csproj create mode 100644 examples/Net4.8/Example9-DisposingOwned/Program.cs create mode 100644 examples/Net4.8/Example9-DisposingOwned/Properties/AssemblyInfo.cs create mode 100644 examples/Net8.0/Example9-DisposingOwned/ConsoleColors.cs create mode 100644 examples/Net8.0/Example9-DisposingOwned/ETL/FileLineLoader.cs create mode 100644 examples/Net8.0/Example9-DisposingOwned/Example9-DisposingOwned.csproj create mode 100644 examples/Net8.0/Example9-DisposingOwned/Program.cs diff --git a/docfx_project/docs/getting-started.md b/docfx_project/docs/getting-started.md index b381aaf5..cad5c731 100644 --- a/docfx_project/docs/getting-started.md +++ b/docfx_project/docs/getting-started.md @@ -122,7 +122,9 @@ appends transformer stages with `Through`, and terminates with a loader: ```csharp using Wolfgang.Etl.Abstractions; -await EtlPipeline.From(source) // IAsyncEnumerable or an ExtractorBase +await EtlPipeline + .Create() // a fresh pipeline builder + .From(source) // IAsyncEnumerable or an ExtractorBase .Through(new ParseTransformer()) // ITransformAsync .Through(new EnrichTransformer(lookup)) // ITransformAsync .To(sqlLoader) // LoaderBase @@ -143,7 +145,9 @@ or pass a **stream-to-stream delegate** to define a one-off stage inline, withou declaring a class: ```csharp -await EtlPipeline.From(orders) +await EtlPipeline + .Create() + .From(orders) .Through(s => s.Where(o => o.Amount > 0)) // Func, IAsyncEnumerable> .Through(Enrich) // a method: IAsyncEnumerable -> IAsyncEnumerable .To(sqlLoader) @@ -162,7 +166,11 @@ And if the source already produces what the loader consumes, skip `Through` enti the compiler requires the loader's input type to match the source's output: ```csharp -await EtlPipeline.From(orders).To(orderLoader).RunAsync(); +await EtlPipeline + .Create() + .From(orders) + .To(orderLoader) + .RunAsync(); ``` ### Progress, cancellation, and the escape hatch @@ -179,7 +187,9 @@ await EtlPipeline.From(orders).To(orderLoader).RunAsync(); var progress = new Progress(p => Console.WriteLine($"extracted {p.RecordsExtracted}, loaded {p.RecordsLoaded}")); -await EtlPipeline.From(extractor) +await EtlPipeline + .Create() + .From(extractor) .Through(enrich) .To(loader) .RunAsync(progress, cancellationToken); @@ -193,10 +203,35 @@ The core deliberately ships only the plumbing — `From`, `Through`, `To`, and - **LINQ-flavored operators** (`Where`, `Select`, `Distinct`, `Take`, `Buffer`, …) are provided by the companion `Wolfgang.Etl.Transformers` package as extension methods over `Through`, reusing the transformers it already ships. With that package - referenced the chain reads `EtlPipeline.From(...).Where(...).Select(...).To(...)`. + referenced the chain reads `EtlPipeline.Create().From(...).Where(...).Select(...).To(...)`. - **Source factories and sink terminators** (for example `CsvExtractor(...)` or - `SqlBulkCopyLoader(...)`) are provided by the format packages, hung off the - `EtlPipeline.Source` sentinel: `EtlPipeline.Source.CsvExtractor("orders.csv")`. + `SqlBulkCopyLoader(...)`) are provided by the format packages as **extension methods + on the `EtlPipeline` instance** (sources) and on `IEtlPipeline` (sinks): + `EtlPipeline.Create().CsvExtractor("orders.csv")`. + +### Cleaning up factory-owned resources (`DisposingOwned`) + +A path-based factory opens a resource the caller never handed it — a file stream, a +connection. `IEtlPipelineSink.DisposingOwned(params object[])` wraps the terminal sink so +those resources are disposed after the run completes (whether it succeeds or throws), +preferring `IAsyncDisposable` and falling back to `IDisposable`, in reverse (LIFO) order. +Format-package sink terminators use it so callers don't have to: + +```csharp +// inside a format package's sink terminator +public static IEtlPipelineSink CsvLoader(this IEtlPipeline pipeline, string path) + where T : notnull +{ + var stream = File.Create(path); + return pipeline + .To(new CsvLoader(stream)) + .DisposingOwned(stream); // stream closed after RunAsync +} +``` + +Passing no resources returns the sink unchanged, so it is safe to call unconditionally. A +runnable version is in +[`examples/Net8.0/Example9-DisposingOwned`](https://github.com/Chris-Wolfgang/ETL-Abstractions/tree/main/examples/Net8.0/Example9-DisposingOwned). A complete, runnable version lives in [`examples/Net8.0/Example8-EtlPipeline`](https://github.com/Chris-Wolfgang/ETL-Abstractions/tree/main/examples/Net8.0/Example8-EtlPipeline) @@ -208,7 +243,7 @@ A complete, runnable version lives in |---|---|---| | Starts from | a typed extractor | any `IAsyncEnumerable` or `ExtractorBase` | | Appends stages | `.Transform(...)` | `.Through(...)` (plus operators via `Wolfgang.Etl.Transformers`) | -| Extended by packages | no | yes — source factories via the `Source` sentinel | +| Extended by packages | no | yes — source factories are extension methods on the `EtlPipeline` instance | | Progress | per-stage `IProgress` | pipeline-level `EtlPipelineProgress` | Reach for `Pipeline` when you already hold discrete extractor/transformer/loader diff --git a/docfx_project/docs/introduction.md b/docfx_project/docs/introduction.md index 76203508..d27c3ddd 100644 --- a/docfx_project/docs/introduction.md +++ b/docfx_project/docs/introduction.md @@ -14,7 +14,7 @@ single strongly-typed, streaming pipeline. - **Fluent, type-safe pipeline** — `Pipeline.Extract(...).Transform(...).Load(...).RunAsync()` composes stages into one runnable flow; the compiler enforces that each stage's output type matches the next stage's input. -- **Generic, format-agnostic pipeline** — `EtlPipeline.From(...).Through(...).To(...).RunAsync()` +- **Generic, format-agnostic pipeline** — `EtlPipeline.Create().From(...).Through(...).To(...).RunAsync()` starts from any `IAsyncEnumerable` or extractor, chains transformer stages, and is the extension point that operator (`Wolfgang.Etl.Transformers`) and format packages build on. diff --git a/examples/Net4.8/Example9-DisposingOwned/ConsoleColors.cs b/examples/Net4.8/Example9-DisposingOwned/ConsoleColors.cs new file mode 100644 index 00000000..81a4eddb --- /dev/null +++ b/examples/Net4.8/Example9-DisposingOwned/ConsoleColors.cs @@ -0,0 +1,11 @@ +namespace Example9_DisposingOwned +{ + internal class ConsoleColors + { + private const char Escape = (char)27; + + public static readonly string Green = Escape + "[32m"; + public static readonly string Yellow = Escape + "[33m"; + public static readonly string Reset = Escape + "[0m"; + } +} diff --git a/examples/Net4.8/Example9-DisposingOwned/ETL/FileLineLoader.cs b/examples/Net4.8/Example9-DisposingOwned/ETL/FileLineLoader.cs new file mode 100644 index 00000000..4ca48135 --- /dev/null +++ b/examples/Net4.8/Example9-DisposingOwned/ETL/FileLineLoader.cs @@ -0,0 +1,41 @@ +using System.Collections.Generic; +using System.IO; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using Wolfgang.Etl.Abstractions; + +namespace Example9_DisposingOwned.ETL +{ + // A sink that writes each record as a line to a stream it was GIVEN — it does not own or + // dispose the stream (note leaveOpen: true). That ownership belongs to whoever opened it. + // In a real format package this is what a path-based sink factory wraps, handing stream + // cleanup to DisposingOwned. + internal sealed class FileLineLoader : LoaderBase + { + private readonly Stream _stream; + + public FileLineLoader(Stream stream) + { + _stream = stream; + } + + protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token) + { + using (var writer = new StreamWriter(_stream, new UTF8Encoding(encoderShouldEmitUTF8Identifier: false), 1024, leaveOpen: true)) + { + await foreach (var item in items.WithCancellation(token)) + { + token.ThrowIfCancellationRequested(); + await writer.WriteLineAsync(item); + IncrementCurrentItemCount(); + } + } + } + + protected override Report CreateProgressReport() + { + return new Report(CurrentItemCount); + } + } +} diff --git a/examples/Net4.8/Example9-DisposingOwned/Example9-DisposingOwned.csproj b/examples/Net4.8/Example9-DisposingOwned/Example9-DisposingOwned.csproj new file mode 100644 index 00000000..2c314561 --- /dev/null +++ b/examples/Net4.8/Example9-DisposingOwned/Example9-DisposingOwned.csproj @@ -0,0 +1,26 @@ + + + + Exe + net48 + Example9_DisposingOwned + Example9-DisposingOwned + 8 + true + true + true + false + $(NoWarn);CA2007 + + + + + + + + + + + + + diff --git a/examples/Net4.8/Example9-DisposingOwned/Program.cs b/examples/Net4.8/Example9-DisposingOwned/Program.cs new file mode 100644 index 00000000..1d95d46b --- /dev/null +++ b/examples/Net4.8/Example9-DisposingOwned/Program.cs @@ -0,0 +1,62 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading.Tasks; +using Example9_DisposingOwned.ETL; +using Wolfgang.Etl.Abstractions; + +namespace Example9_DisposingOwned +{ + internal class Program + { + private static async Task Main() + { + Console.WriteLine(ConsoleColors.Green + ".NET Version: " + Environment.Version + ConsoleColors.Reset + "\n"); + + Console.WriteLine( + ConsoleColors.Yellow + + "Running a pipeline whose sink writes to a file the caller opened, " + + "then letting DisposingOwned close it..." + ConsoleColors.Reset + "\n\n"); + + var path = Path.Combine(Path.GetTempPath(), "example9-" + Guid.NewGuid().ToString("N") + ".txt"); + + // The pipeline builder — not the loader — opened this stream, so the builder is + // responsible for closing it. A format package's path-based sink factory does exactly + // this internally; here we spell it out. + var stream = File.Create(path); + + // .To(loader) terminates the pipeline into an IEtlPipelineSink. + // .DisposingOwned(stream) wraps that sink so the factory-opened stream is disposed + // after RunAsync finishes — whether it succeeds or throws. On .NET Framework the + // FileStream implements only IDisposable, so it is disposed synchronously. + await EtlPipeline + .Create() + .From(Lines()) + .To(new FileLineLoader(stream)) + .DisposingOwned(stream) + .RunAsync(); + + // Proof the stream was closed: if it were still open, File.ReadAllLines would throw + // IOException (the file would still be locked for writing). + Console.WriteLine(ConsoleColors.Green + "Wrote and closed:" + ConsoleColors.Reset + " " + path + "\n"); + foreach (var line in File.ReadAllLines(path)) + { + Console.WriteLine(" " + line); + } + + File.Delete(path); + Console.WriteLine("\n" + ConsoleColors.Yellow + + "Pipeline completed; the file stream was disposed by DisposingOwned." + ConsoleColors.Reset); + } + + private static async IAsyncEnumerable Lines() + { + var fruit = new[] { "apple", "banana", "cherry", "date" }; + foreach (var item in fruit) + { + await Task.Delay(25); + yield return item; + } + } + } +} diff --git a/examples/Net4.8/Example9-DisposingOwned/Properties/AssemblyInfo.cs b/examples/Net4.8/Example9-DisposingOwned/Properties/AssemblyInfo.cs new file mode 100644 index 00000000..8c839b13 --- /dev/null +++ b/examples/Net4.8/Example9-DisposingOwned/Properties/AssemblyInfo.cs @@ -0,0 +1,32 @@ +using System.Reflection; +using System.Runtime.InteropServices; + +// General Information about an assembly is controlled through the following +// set of attributes. Change these attribute values to modify the information +// associated with an assembly. +[assembly: AssemblyTitle("Example9-DisposingOwned")] +[assembly: AssemblyDescription("")] +[assembly: AssemblyConfiguration("")] +[assembly: AssemblyCompany("")] +[assembly: AssemblyProduct("Example9-DisposingOwned")] +[assembly: AssemblyCopyright("Copyright © 2026")] +[assembly: AssemblyTrademark("")] +[assembly: AssemblyCulture("")] + +// Setting ComVisible to false makes the types in this assembly not visible +// to COM components. If you need to access a type in this assembly from +// COM, set the ComVisible attribute to true on that type. +[assembly: ComVisible(false)] + +// The following GUID is for the ID of the typelib if this project is exposed to COM +[assembly: Guid("3a9f2c14-7b6e-4d21-9f80-1c2e3a4b5c6d")] + +// Version information for an assembly consists of the following four values: +// +// Major Version +// Minor Version +// Build Number +// Revision +// +[assembly: AssemblyVersion("1.0.0.0")] +[assembly: AssemblyFileVersion("1.0.0.0")] diff --git a/examples/Net8.0/Example9-DisposingOwned/ConsoleColors.cs b/examples/Net8.0/Example9-DisposingOwned/ConsoleColors.cs new file mode 100644 index 00000000..f16b50a0 --- /dev/null +++ b/examples/Net8.0/Example9-DisposingOwned/ConsoleColors.cs @@ -0,0 +1,8 @@ +namespace Example9_DisposingOwned; + +internal class ConsoleColors +{ + public const string Green = "\e[32m"; + public const string Yellow = "\e[33m"; + public const string Reset = "\e[0m"; +} diff --git a/examples/Net8.0/Example9-DisposingOwned/ETL/FileLineLoader.cs b/examples/Net8.0/Example9-DisposingOwned/ETL/FileLineLoader.cs new file mode 100644 index 00000000..594da6af --- /dev/null +++ b/examples/Net8.0/Example9-DisposingOwned/ETL/FileLineLoader.cs @@ -0,0 +1,28 @@ +using Wolfgang.Etl.Abstractions; + +namespace Example9_DisposingOwned.ETL; + +// A sink that writes each record as a line to a stream it was GIVEN — it does not own or +// dispose the stream (note leaveOpen: true). That ownership belongs to whoever opened it. +// In a real format package this is what a path-based sink factory wraps, handing stream +// cleanup to DisposingOwned. +internal sealed class FileLineLoader(Stream stream) : LoaderBase +{ + protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token) + { + await using var writer = new StreamWriter(stream, leaveOpen: true); + + await foreach (var item in items.WithCancellation(token)) + { + token.ThrowIfCancellationRequested(); + await writer.WriteLineAsync(item.AsMemory(), token); + IncrementCurrentItemCount(); + } + } + + + protected override Report CreateProgressReport() + { + return new Report(CurrentItemCount); + } +} diff --git a/examples/Net8.0/Example9-DisposingOwned/Example9-DisposingOwned.csproj b/examples/Net8.0/Example9-DisposingOwned/Example9-DisposingOwned.csproj new file mode 100644 index 00000000..332adb5d --- /dev/null +++ b/examples/Net8.0/Example9-DisposingOwned/Example9-DisposingOwned.csproj @@ -0,0 +1,19 @@ + + + + Exe + net8.0 + Example9_DisposingOwned + enable + CA2007 + 1.0.0 + Copyright {copyright year} {author} + + + + + + + + + diff --git a/examples/Net8.0/Example9-DisposingOwned/Program.cs b/examples/Net8.0/Example9-DisposingOwned/Program.cs new file mode 100644 index 00000000..b3515ef3 --- /dev/null +++ b/examples/Net8.0/Example9-DisposingOwned/Program.cs @@ -0,0 +1,56 @@ +using Example9_DisposingOwned.ETL; +using Wolfgang.Etl.Abstractions; + +namespace Example9_DisposingOwned; + +internal class Program +{ + private static async Task Main() + { + Console.WriteLine($"{ConsoleColors.Green}.NET Version: {Environment.Version}{ConsoleColors.Reset}\n"); + + Console.WriteLine( + $"{ConsoleColors.Yellow}Running a pipeline whose sink writes to a file the caller opened, " + + $"then letting DisposingOwned close it...{ConsoleColors.Reset}\n\n"); + + var path = Path.Combine(Path.GetTempPath(), $"example9-{Guid.NewGuid():N}.txt"); + + // The pipeline builder — not the loader — opened this stream, so the builder is + // responsible for closing it. A format package's path-based sink factory does exactly + // this internally; here we spell it out. + var stream = File.Create(path); + + // .To(loader) terminates the pipeline into an IEtlPipelineSink. + // .DisposingOwned(stream) wraps that sink so the factory-opened stream is disposed + // after RunAsync finishes — whether it succeeds or throws. IAsyncDisposable is preferred + // (FileStream is disposed asynchronously on net8.0), falling back to IDisposable. + await EtlPipeline + .Create() + .From(Lines()) + .To(new FileLineLoader(stream)) + .DisposingOwned(stream) + .RunAsync(); + + // Proof the stream was closed: if it were still open, File.OpenRead would throw + // IOException (the file would still be locked for writing). + Console.WriteLine($"{ConsoleColors.Green}Wrote and closed:{ConsoleColors.Reset} {path}\n"); + foreach (var line in await File.ReadAllLinesAsync(path)) + { + Console.WriteLine($" {line}"); + } + + File.Delete(path); + Console.WriteLine($"\n{ConsoleColors.Yellow}Pipeline completed; the file stream was disposed by DisposingOwned.{ConsoleColors.Reset}"); + } + + + private static async IAsyncEnumerable Lines() + { + string[] fruit = ["apple", "banana", "cherry", "date"]; + foreach (var item in fruit) + { + await Task.Delay(25); + yield return item; + } + } +}