feat: generic EtlPipeline core (#147)

ETL-Abstractions.sln

@@ -87,6 +87,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example7-FluentPipeline", "
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example7-FluentPipeline", "examples\Net4.8\Example7-FluentPipeline\Example7-FluentPipeline.csproj", "{F26156B5-9FDC-46CA-9152-F888932238E1}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example8-EtlPipeline", "examples\Net8.0\Example8-EtlPipeline\Example8-EtlPipeline.csproj", "{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example8-EtlPipeline", "examples\Net4.8\Example8-EtlPipeline\Example8-EtlPipeline.csproj", "{B59B4926-9CDD-41C2-A367-42F030DFAF54}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -349,6 +353,30 @@ Global
 		{F26156B5-9FDC-46CA-9152-F888932238E1}.Release|x64.Build.0 = Release|Any CPU
 		{F26156B5-9FDC-46CA-9152-F888932238E1}.Release|x86.ActiveCfg = Release|Any CPU
 		{F26156B5-9FDC-46CA-9152-F888932238E1}.Release|x86.Build.0 = Release|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Debug|x64.Build.0 = Debug|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Debug|x86.Build.0 = Debug|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Release|Any CPU.Build.0 = Release|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Release|x64.ActiveCfg = Release|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Release|x64.Build.0 = Release|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Release|x86.ActiveCfg = Release|Any CPU
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74}.Release|x86.Build.0 = Release|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Debug|x64.Build.0 = Debug|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Debug|x86.Build.0 = Debug|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Release|x64.ActiveCfg = Release|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Release|x64.Build.0 = Release|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Release|x86.ActiveCfg = Release|Any CPU
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -379,6 +407,8 @@ Global
 		{7855D034-57F4-4CF0-8EB4-5677EF80EC57} = {AE206253-B766-4B6A-8C08-9E70605A2B27}
 		{A39682A7-CD1A-40A2-9F5B-F1E045C11F8D} = {3C48157B-5E90-489E-9444-E01F51D59F86}
 		{F26156B5-9FDC-46CA-9152-F888932238E1} = {336D72A1-8E5E-49DE-83D9-DF6BE458BA24}
+		{F97582DD-F8D7-4BA0-B46E-20B7200ADC74} = {3C48157B-5E90-489E-9444-E01F51D59F86}
+		{B59B4926-9CDD-41C2-A367-42F030DFAF54} = {336D72A1-8E5E-49DE-83D9-DF6BE458BA24}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {F673635D-58CE-48A5-9AE4-31F4484BED9E}

docfx_project/docs/getting-started.md

@@ -111,6 +111,110 @@ A complete, runnable version of the basic walkthrough lives in
 [`examples/Net8.0/Example7-FluentPipeline`](https://github.com/Chris-Wolfgang/ETL-Abstractions/tree/main/examples/Net8.0/Example7-FluentPipeline)
 (the `Net4.8` folder has the same example for .NET Framework).
 
+## Generic pipeline (`EtlPipeline`)
+
+Alongside the fluent `Pipeline`, `EtlPipeline` is a **format-agnostic** pipeline
+built for cross-format flows (CSV → JSON, JSON → SQL, …) and for being extended by
+other packages. Where `Pipeline` starts from a typed extractor, `EtlPipeline` starts
+from *any* source — an `IAsyncEnumerable<T>` or an `ExtractorBase<T, TProgress>` —
+appends transformer stages with `Through`, and terminates with a loader:
+
+```csharp
+using Wolfgang.Etl.Abstractions;
+
+await EtlPipeline.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>
+    .RunAsync(progress, cancellationToken);
+```
+
+`Through` returns `IEtlPipeline<TOut>`, so you can chain as many stages as you like;
+the element type flows from one stage's output to the next and the compiler enforces
+the match (a same-type `ITransformAsync<T, T>` stage is fine). Nothing runs until
+`To(...).RunAsync()` — records are then pulled through the whole chain one at a time,
+with no buffering between stages.
+
+### Supplying a stage: a transformer class or an inline delegate
+
+`Through` has two shapes. Pass an `ITransformAsync<T, TOut>` (or the cancellation-aware
+`ITransformWithCancellationAsync<T, TOut>`) when you have a reusable transformer class —
+or pass a **stream-to-stream delegate** to define a one-off stage inline, without
+declaring a class:
+
+```csharp
+await EtlPipeline.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)
+    .RunAsync();
+```
+
+The delegate is the same stream-to-stream contract as `ITransformAsync.TransformAsync`
+(the lambda body typically composes `System.Linq.Async` operators), so it stays at the
+"append a stage" level — it is *not* a per-element projection. (A per-element
+`Where`/`Select` over individual records is an operator, provided by
+`Wolfgang.Etl.Transformers`; see below.) A cancellation-aware delegate overload,
+`Func<IAsyncEnumerable<T>, CancellationToken, IAsyncEnumerable<TOut>>`, receives the
+run's token.
+
+And if the source already produces what the loader consumes, skip `Through` entirely —
+the compiler requires the loader's input type to match the source's output:
+
+```csharp
+await EtlPipeline.From(orders).To(orderLoader).RunAsync();
+```
+
+### Progress, cancellation, and the escape hatch
+
+- **Progress** — pass an `IProgress<EtlPipelineProgress>` to `RunAsync`. The snapshot
+  reports `RecordsExtracted` (counted at the source) and `RecordsLoaded` (counted at
+  the sink) plus `Elapsed`, regardless of how many stages sit in between.
+- **Cancellation** — the token passed to `RunAsync` is observed while pulling from the
+  source and is forwarded into any cancellation-aware transformer stage.
+- **Escape hatch** — `AsAsyncEnumerable(token)` drops to the raw `IAsyncEnumerable<T>`
+  so you can apply `System.Linq.Async` operators directly.
+
+```csharp
+var progress = new Progress<EtlPipelineProgress>(p =>
+    Console.WriteLine($"extracted {p.RecordsExtracted}, loaded {p.RecordsLoaded}"));
+
+await EtlPipeline.From(extractor)
+    .Through(enrich)
+    .To(loader)
+    .RunAsync(progress, cancellationToken);
+```
+
+### Operators and source factories
+
+The core deliberately ships only the plumbing — `From`, `Through`, `To`, and
+`AsAsyncEnumerable`. Two layers build on it:
+
+- **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(...)`.
+- **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")`.
+
+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)
+(the `Net4.8` folder has the same example for .NET Framework).
+
+### `Pipeline` vs `EtlPipeline`
+
+| | `Pipeline` (fluent) | `EtlPipeline` (generic) |
+|---|---|---|
+| 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 |
+| Progress | per-stage `IProgress<T>` | pipeline-level `EtlPipelineProgress` |
+
+Reach for `Pipeline` when you already hold discrete extractor/transformer/loader
+objects and want per-stage progress; reach for `EtlPipeline` for cross-format flows or
+when you want the operator and format-package extensions.
+
 ## Next Steps
 
 - Explore the [API Reference](../api/index.md) for detailed documentation

docfx_project/docs/introduction.md

@@ -14,6 +14,10 @@ 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()`
+  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.
 - **Async streaming** — built on `IAsyncEnumerable<T>`, so items flow through the
   pipeline without buffering the whole set in memory.
 - **Opt-in progress reporting** — progress-capable stages surface `IProgress<T>`

examples/Net4.8/Example8-EtlPipeline/ConsoleColors.cs

@@ -0,0 +1,9 @@
+namespace Example8_EtlPipeline
+{
+    internal class ConsoleColors
+    {
+        public const string Green = "\u001b[32m";
+        public const string Yellow = "\u001b[33m";
+        public const string Reset = "\u001b[0m";
+    }
+}

examples/Net4.8/Example8-EtlPipeline/ETL/ConsoleLoader.cs

@@ -0,0 +1,28 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Wolfgang.Etl.Abstractions;
+
+namespace Example8_EtlPipeline.ETL
+{
+    // Sink: the pipeline's To(...) terminator requires a LoaderBase<T, TProgress>.
+    internal sealed class ConsoleLoader : LoaderBase<string, Report>
+    {
+        protected override async Task LoadWorkerAsync(IAsyncEnumerable<string> items, CancellationToken token)
+        {
+            await foreach (var item in items.WithCancellation(token))
+            {
+                token.ThrowIfCancellationRequested();
+                Console.WriteLine($"{ConsoleColors.Green}load  {ConsoleColors.Reset} {item}");
+                IncrementCurrentItemCount();
+            }
+        }
+
+
+        protected override Report CreateProgressReport()
+        {
+            return new Report(CurrentItemCount);
+        }
+    }
+}

examples/Net4.8/Example8-EtlPipeline/ETL/DoubleTransformer.cs

@@ -0,0 +1,20 @@
+using System;
+using System.Collections.Generic;
+using Wolfgang.Etl.Abstractions;
+
+namespace Example8_EtlPipeline.ETL
+{
+    // Stage 2: int -> double
+    internal sealed class DoubleTransformer : ITransformAsync<int, double>
+    {
+        public async IAsyncEnumerable<double> TransformAsync(IAsyncEnumerable<int> items)
+        {
+            await foreach (var item in items)
+            {
+                var doubled = item * 2.0;
+                Console.WriteLine($"{ConsoleColors.Green}double{ConsoleColors.Reset} {item} -> {doubled:F1}");
+                yield return doubled;
+            }
+        }
+    }
+}

examples/Net4.8/Example8-EtlPipeline/ETL/FormatTransformer.cs

@@ -0,0 +1,17 @@
+using System.Collections.Generic;
+using Wolfgang.Etl.Abstractions;
+
+namespace Example8_EtlPipeline.ETL
+{
+    // Stage 3: double -> string
+    internal sealed class FormatTransformer : ITransformAsync<double, string>
+    {
+        public async IAsyncEnumerable<string> TransformAsync(IAsyncEnumerable<double> items)
+        {
+            await foreach (var item in items)
+            {
+                yield return $"value = {item:F1}";
+            }
+        }
+    }
+}

examples/Net4.8/Example8-EtlPipeline/ETL/ParseIntTransformer.cs

@@ -0,0 +1,20 @@
+using System;
+using System.Collections.Generic;
+using Wolfgang.Etl.Abstractions;
+
+namespace Example8_EtlPipeline.ETL
+{
+    // Stage 1: string -> int
+    internal sealed class ParseIntTransformer : ITransformAsync<string, int>
+    {
+        public async IAsyncEnumerable<int> TransformAsync(IAsyncEnumerable<string> items)
+        {
+            await foreach (var item in items)
+            {
+                var value = int.Parse(item);
+                Console.WriteLine($"{ConsoleColors.Green}parse {ConsoleColors.Reset} \"{item}\" -> {value}");
+                yield return value;
+            }
+        }
+    }
+}

examples/Net4.8/Example8-EtlPipeline/Example8-EtlPipeline.csproj

@@ -0,0 +1,26 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net48</TargetFramework>
+    <RootNamespace>Example8_EtlPipeline</RootNamespace>
+    <AssemblyName>Example8-EtlPipeline</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/Example8-EtlPipeline/Program.cs

@@ -0,0 +1,52 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Example8_EtlPipeline.ETL;
+using Wolfgang.Etl.Abstractions;
+
+namespace Example8_EtlPipeline
+{
+    internal class Program
+    {
+        private static async Task Main()
+        {
+            Console.WriteLine($"{ConsoleColors.Green}.NET Version: {Environment.Version}{ConsoleColors.Reset}\n");
+
+            Console.WriteLine($"{ConsoleColors.Yellow}Building a generic EtlPipeline with three chained Through stages...{ConsoleColors.Reset}\n\n");
+
+            // Report EtlPipelineProgress as records flow through. The core counts the
+            // two ends of the pipeline: RecordsExtracted at the source, RecordsLoaded
+            // at the sink — regardless of how many stages sit in between.
+            var progress = new Progress<EtlPipelineProgress>(p =>
+                Console.WriteLine(
+                    $"{ConsoleColors.Green}[progress]{ConsoleColors.Reset} " +
+                    $"extracted={p.RecordsExtracted} loaded={p.RecordsLoaded} " +
+                    $"elapsed={p.Elapsed.TotalMilliseconds:F0}ms"));
+
+            // From a raw async stream, pipe through three transformer stages, then load.
+            // Each Through returns IEtlPipeline<TOut>, so the element type flows
+            // string -> int -> double -> string across the chain and the compiler enforces
+            // that each stage's output matches the next stage's input.
+            await EtlPipeline.From(RawNumbers())
+                .Through(new ParseIntTransformer())   // string -> int
+                .Through(new DoubleTransformer())     // int    -> double
+                .Through(new FormatTransformer())     // double -> string
+                .To(new ConsoleLoader())              // LoaderBase<string, Report>
+                .RunAsync(progress);
+
+            Console.WriteLine($"\n\n{ConsoleColors.Yellow}Pipeline completed.{ConsoleColors.Reset}");
+        }
+
+
+        // Any IAsyncEnumerable<T> is a valid source via EtlPipeline.From(...).
+        // An ExtractorBase<T, TProgress> works too: EtlPipeline.From(myExtractor).
+        private static async IAsyncEnumerable<string> RawNumbers()
+        {
+            for (var i = 1; i <= 8; i++)
+            {
+                await Task.Delay(50);
+                yield return i.ToString();
+            }
+        }
+    }
+}

examples/Net4.8/Example8-EtlPipeline/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("Example8-EtlPipeline")]
+[assembly: AssemblyDescription("")]
+[assembly: AssemblyConfiguration("")]
+[assembly: AssemblyCompany("")]
+[assembly: AssemblyProduct("Example8-EtlPipeline")]
+[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("24f8deda-6cc9-4a05-8588-eee0ec2d041b")]
+
+// 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/Example8-EtlPipeline/ConsoleColors.cs

@@ -0,0 +1,8 @@
+namespace Example8_EtlPipeline;
+
+internal class ConsoleColors
+{
+    public const string Green = "\e[32m";
+    public const string Yellow = "\e[33m";
+    public const string Reset = "\e[0m";
+}