docs: EtlPipeline Create()/DisposingOwned docs + Example9

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<string> or an ExtractorBase
+await EtlPipeline
+    .Create()                                // a fresh pipeline builder
+    .From(source)                            // IAsyncEnumerable<string> or an ExtractorBase
     .Through(new ParseTransformer())         // ITransformAsync<string, Order>
     .Through(new EnrichTransformer(lookup))  // ITransformAsync<Order, EnrichedOrder>
     .To(sqlLoader)                           // LoaderBase<EnrichedOrder, TProgress>
@@ -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<Order>, IAsyncEnumerable<Order>>
     .Through(Enrich)                            // a method: IAsyncEnumerable<Order> -> IAsyncEnumerable<EnrichedOrder>
     .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<EtlPipelineProgress>(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<T>(...)` or
-  `SqlBulkCopyLoader<T>(...)`) are provided by the format packages, hung off the
-  `EtlPipeline.Source` sentinel: `EtlPipeline.Source.CsvExtractor<Order>("orders.csv")`.
+  `SqlBulkCopyLoader<T>(...)`) are provided by the format packages as **extension methods
+  on the `EtlPipeline` instance** (sources) and on `IEtlPipeline<T>` (sinks):
+  `EtlPipeline.Create().CsvExtractor<Order>("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<T>(this IEtlPipeline<T> pipeline, string path)
+    where T : notnull
+{
+    var stream = File.Create(path);
+    return pipeline
+        .To(new CsvLoader<T>(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<T>` or `ExtractorBase<T, TProgress>` |
 | 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<T>` | pipeline-level `EtlPipelineProgress` |
 
 Reach for `Pipeline` when you already hold discrete extractor/transformer/loader

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<T>` or extractor, chains transformer stages, and is
   the extension point that operator (`Wolfgang.Etl.Transformers`) and format packages
   build on.

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";
+    }
+}

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<string, Report>
+    {
+        private readonly Stream _stream;
+
+        public FileLineLoader(Stream stream)
+        {
+            _stream = stream;
+        }
+
+        protected override async Task LoadWorkerAsync(IAsyncEnumerable<string> 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);
+        }
+    }
+}

examples/Net4.8/Example9-DisposingOwned/Example9-DisposingOwned.csproj

@@ -0,0 +1,26 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net48</TargetFramework>
+    <RootNamespace>Example9_DisposingOwned</RootNamespace>
+    <AssemblyName>Example9-DisposingOwned</AssemblyName>
+    <LangVersion>8</LangVersion>
+    <Deterministic>true</Deterministic>
+    <AutoGenerateBindingRedirects>true</AutoGenerateBindingRedirects>
+    <EnableWindowsTargeting>true</EnableWindowsTargeting>
+    <GenerateAssemblyInfo>false</GenerateAssemblyInfo>
+    <NoWarn>$(NoWarn);CA2007</NoWarn>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="10.0.5" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\src\Wolfgang.Etl.Abstractions\Wolfgang.Etl.Abstractions.csproj" />
+  </ItemGroup>
+
+  <!-- Analyzer PackageReferences are centralized in Directory.Build.props -->
+
+</Project>

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<string> Lines()
+        {
+            var fruit = new[] { "apple", "banana", "cherry", "date" };
+            foreach (var item in fruit)
+            {
+                await Task.Delay(25);
+                yield return item;
+            }
+        }
+    }
+}

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")]

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";
+}

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<string, Report>
+{
+    protected override async Task LoadWorkerAsync(IAsyncEnumerable<string> 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);
+    }
+}

examples/Net8.0/Example9-DisposingOwned/Example9-DisposingOwned.csproj

@@ -0,0 +1,19 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net8.0</TargetFramework>
+    <RootNamespace>Example9_DisposingOwned</RootNamespace>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <NoWarn>CA2007</NoWarn>
+    <Version>1.0.0</Version>
+    <Copyright>Copyright {copyright year} {author}</Copyright>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\src\Wolfgang.Etl.Abstractions\Wolfgang.Etl.Abstractions.csproj" />
+  </ItemGroup>
+
+  <!-- Analyzer PackageReferences are centralized in Directory.Build.props -->
+
+</Project>

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<string> Lines()
+    {
+        string[] fruit = ["apple", "banana", "cherry", "date"];
+        foreach (var item in fruit)
+        {
+            await Task.Delay(25);
+            yield return item;
+        }
+    }
+}