diff --git a/ETL-Abstractions.sln b/ETL-Abstractions.sln index de0e11f5..73eb4adf 100644 --- a/ETL-Abstractions.sln +++ b/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} diff --git a/docfx_project/docs/getting-started.md b/docfx_project/docs/getting-started.md index 6a8d5280..b381aaf5 100644 --- a/docfx_project/docs/getting-started.md +++ b/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` or an `ExtractorBase` — +appends transformer stages with `Through`, and terminates with a loader: + +```csharp +using Wolfgang.Etl.Abstractions; + +await EtlPipeline.From(source) // IAsyncEnumerable or an ExtractorBase + .Through(new ParseTransformer()) // ITransformAsync + .Through(new EnrichTransformer(lookup)) // ITransformAsync + .To(sqlLoader) // LoaderBase + .RunAsync(progress, cancellationToken); +``` + +`Through` returns `IEtlPipeline`, 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` 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` (or the cancellation-aware +`ITransformWithCancellationAsync`) 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> + .Through(Enrich) // a method: IAsyncEnumerable -> IAsyncEnumerable + .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, CancellationToken, IAsyncEnumerable>`, 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` 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` + so you can apply `System.Linq.Async` operators directly. + +```csharp +var progress = new Progress(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(...)` or + `SqlBulkCopyLoader(...)`) are provided by the format packages, hung off the + `EtlPipeline.Source` sentinel: `EtlPipeline.Source.CsvExtractor("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` or `ExtractorBase` | +| 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` | 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 diff --git a/docfx_project/docs/introduction.md b/docfx_project/docs/introduction.md index ffd8feec..76203508 100644 --- a/docfx_project/docs/introduction.md +++ b/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` 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`, so items flow through the pipeline without buffering the whole set in memory. - **Opt-in progress reporting** — progress-capable stages surface `IProgress` diff --git a/examples/Net4.8/Example8-EtlPipeline/ConsoleColors.cs b/examples/Net4.8/Example8-EtlPipeline/ConsoleColors.cs new file mode 100644 index 00000000..52184544 --- /dev/null +++ b/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"; + } +} diff --git a/examples/Net4.8/Example8-EtlPipeline/ETL/ConsoleLoader.cs b/examples/Net4.8/Example8-EtlPipeline/ETL/ConsoleLoader.cs new file mode 100644 index 00000000..2ee5b21b --- /dev/null +++ b/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. + internal sealed class ConsoleLoader : LoaderBase + { + protected override async Task LoadWorkerAsync(IAsyncEnumerable 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); + } + } +} diff --git a/examples/Net4.8/Example8-EtlPipeline/ETL/DoubleTransformer.cs b/examples/Net4.8/Example8-EtlPipeline/ETL/DoubleTransformer.cs new file mode 100644 index 00000000..4d95e856 --- /dev/null +++ b/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 + { + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable 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; + } + } + } +} diff --git a/examples/Net4.8/Example8-EtlPipeline/ETL/FormatTransformer.cs b/examples/Net4.8/Example8-EtlPipeline/ETL/FormatTransformer.cs new file mode 100644 index 00000000..4325c95f --- /dev/null +++ b/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 + { + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable items) + { + await foreach (var item in items) + { + yield return $"value = {item:F1}"; + } + } + } +} diff --git a/examples/Net4.8/Example8-EtlPipeline/ETL/ParseIntTransformer.cs b/examples/Net4.8/Example8-EtlPipeline/ETL/ParseIntTransformer.cs new file mode 100644 index 00000000..4d3c5746 --- /dev/null +++ b/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 + { + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable items) + { + await foreach (var item in items) + { + var value = int.Parse(item); + Console.WriteLine($"{ConsoleColors.Green}parse {ConsoleColors.Reset} \"{item}\" -> {value}"); + yield return value; + } + } + } +} diff --git a/examples/Net4.8/Example8-EtlPipeline/Example8-EtlPipeline.csproj b/examples/Net4.8/Example8-EtlPipeline/Example8-EtlPipeline.csproj new file mode 100644 index 00000000..4707a8a9 --- /dev/null +++ b/examples/Net4.8/Example8-EtlPipeline/Example8-EtlPipeline.csproj @@ -0,0 +1,26 @@ + + + + Exe + net48 + Example8_EtlPipeline + Example8-EtlPipeline + 8 + true + true + true + false + $(NoWarn);CA2007 + + + + + + + + + + + + + diff --git a/examples/Net4.8/Example8-EtlPipeline/Program.cs b/examples/Net4.8/Example8-EtlPipeline/Program.cs new file mode 100644 index 00000000..716ae1f3 --- /dev/null +++ b/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(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, 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 + .RunAsync(progress); + + Console.WriteLine($"\n\n{ConsoleColors.Yellow}Pipeline completed.{ConsoleColors.Reset}"); + } + + + // Any IAsyncEnumerable is a valid source via EtlPipeline.From(...). + // An ExtractorBase works too: EtlPipeline.From(myExtractor). + private static async IAsyncEnumerable RawNumbers() + { + for (var i = 1; i <= 8; i++) + { + await Task.Delay(50); + yield return i.ToString(); + } + } + } +} diff --git a/examples/Net4.8/Example8-EtlPipeline/Properties/AssemblyInfo.cs b/examples/Net4.8/Example8-EtlPipeline/Properties/AssemblyInfo.cs new file mode 100644 index 00000000..5522f2ac --- /dev/null +++ b/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")] diff --git a/examples/Net8.0/Example8-EtlPipeline/ConsoleColors.cs b/examples/Net8.0/Example8-EtlPipeline/ConsoleColors.cs new file mode 100644 index 00000000..d282de44 --- /dev/null +++ b/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"; +} diff --git a/examples/Net8.0/Example8-EtlPipeline/ETL/ConsoleLoader.cs b/examples/Net8.0/Example8-EtlPipeline/ETL/ConsoleLoader.cs new file mode 100644 index 00000000..ba9c672b --- /dev/null +++ b/examples/Net8.0/Example8-EtlPipeline/ETL/ConsoleLoader.cs @@ -0,0 +1,23 @@ +using Wolfgang.Etl.Abstractions; + +namespace Example8_EtlPipeline.ETL; + +// Sink: the pipeline's To(...) terminator requires a LoaderBase. +internal sealed class ConsoleLoader : LoaderBase +{ + protected override async Task LoadWorkerAsync(IAsyncEnumerable 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); + } +} diff --git a/examples/Net8.0/Example8-EtlPipeline/ETL/DoubleTransformer.cs b/examples/Net8.0/Example8-EtlPipeline/ETL/DoubleTransformer.cs new file mode 100644 index 00000000..581e5dda --- /dev/null +++ b/examples/Net8.0/Example8-EtlPipeline/ETL/DoubleTransformer.cs @@ -0,0 +1,17 @@ +using Wolfgang.Etl.Abstractions; + +namespace Example8_EtlPipeline.ETL; + +// Stage 2: int -> double +internal sealed class DoubleTransformer : ITransformAsync +{ + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable 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; + } + } +} diff --git a/examples/Net8.0/Example8-EtlPipeline/ETL/FormatTransformer.cs b/examples/Net8.0/Example8-EtlPipeline/ETL/FormatTransformer.cs new file mode 100644 index 00000000..caf16766 --- /dev/null +++ b/examples/Net8.0/Example8-EtlPipeline/ETL/FormatTransformer.cs @@ -0,0 +1,15 @@ +using Wolfgang.Etl.Abstractions; + +namespace Example8_EtlPipeline.ETL; + +// Stage 3: double -> string +internal sealed class FormatTransformer : ITransformAsync +{ + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable items) + { + await foreach (var item in items) + { + yield return $"value = {item:F1}"; + } + } +} diff --git a/examples/Net8.0/Example8-EtlPipeline/ETL/ParseIntTransformer.cs b/examples/Net8.0/Example8-EtlPipeline/ETL/ParseIntTransformer.cs new file mode 100644 index 00000000..33f151ba --- /dev/null +++ b/examples/Net8.0/Example8-EtlPipeline/ETL/ParseIntTransformer.cs @@ -0,0 +1,17 @@ +using Wolfgang.Etl.Abstractions; + +namespace Example8_EtlPipeline.ETL; + +// Stage 1: string -> int +internal sealed class ParseIntTransformer : ITransformAsync +{ + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable items) + { + await foreach (var item in items) + { + var value = int.Parse(item); + Console.WriteLine($"{ConsoleColors.Green}parse {ConsoleColors.Reset} \"{item}\" -> {value}"); + yield return value; + } + } +} diff --git a/examples/Net8.0/Example8-EtlPipeline/Example8-EtlPipeline.csproj b/examples/Net8.0/Example8-EtlPipeline/Example8-EtlPipeline.csproj new file mode 100644 index 00000000..f0fb74a8 --- /dev/null +++ b/examples/Net8.0/Example8-EtlPipeline/Example8-EtlPipeline.csproj @@ -0,0 +1,19 @@ + + + + Exe + net8.0 + Example8_EtlPipeline + enable + CA2007 + 1.0.0 + Copyright {copyright year} {author} + + + + + + + + + diff --git a/examples/Net8.0/Example8-EtlPipeline/Program.cs b/examples/Net8.0/Example8-EtlPipeline/Program.cs new file mode 100644 index 00000000..19b320da --- /dev/null +++ b/examples/Net8.0/Example8-EtlPipeline/Program.cs @@ -0,0 +1,48 @@ +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(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, 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 + .RunAsync(progress); + + Console.WriteLine($"\n\n{ConsoleColors.Yellow}Pipeline completed.{ConsoleColors.Reset}"); + } + + + // Any IAsyncEnumerable is a valid source via EtlPipeline.From(...). + // An ExtractorBase works too: EtlPipeline.From(myExtractor). + private static async IAsyncEnumerable RawNumbers() + { + for (var i = 1; i <= 8; i++) + { + await Task.Delay(50); + yield return i.ToString(); + } + } +} diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipeline.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipeline.cs new file mode 100644 index 00000000..d2d1a465 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipeline.cs @@ -0,0 +1,85 @@ +using System; +using System.Collections.Generic; +using System.Threading; + + +namespace Wolfgang.Etl.Abstractions; + +/// +/// Entry point for building a generic, format-agnostic ETL pipeline. Start from a source — either a +/// built-in / +/// factory, or a format-specific factory hung off by a format package — chain +/// append transformer stages on , terminate with a sink, then call +/// . +/// +/// +/// +/// The name EtlPipeline (rather than Pipeline) avoids clashing with the fluent +/// extract/transform/load builder in this same namespace and with +/// System.IO.Pipelines. +/// +/// +/// The core exposes only the plumbing — a source, +/// for appending transformer stages, and a sink. The LINQ-flavored operators (Where, +/// Select, Distinct, …) are extension methods shipped by Wolfgang.Etl.Transformers, +/// which already owns those transformers. +/// +/// +/// +/// +/// await EtlPipeline.From(records) +/// .Through(new WhereTransformer<Order>(r => r.Amount > 0)) +/// .To(sqlLoader) +/// .RunAsync(progress, token); +/// +/// +public static class EtlPipeline +{ + /// + /// The sentinel that format packages extend with source factories, enabling the + /// EtlPipeline.Source.CsvExtractor<T>(...) shape. See . + /// + public static EtlPipelineSource Source { get; } = new(); + + + /// + /// Begins a pipeline from an existing asynchronous stream — the generic escape hatch for any + /// source the caller already has as an . + /// + /// The type of item produced by the source. + /// The stream that seeds the pipeline. + /// An for chaining. + /// is . + public static IEtlPipeline From(IAsyncEnumerable source) + where T : notnull + { + if (source is null) + { + throw new ArgumentNullException(nameof(source)); + } + + return EtlPipelineImpl.FromStream((_, _) => source); + } + + + /// + /// Begins a pipeline from an . The pipeline's + /// cancellation token is forwarded to the extractor. + /// + /// The type of item produced by the extractor. + /// The extractor's progress-report type. + /// The extractor that seeds the pipeline. + /// An for chaining. + /// is . + public static IEtlPipeline From(ExtractorBase extractor) + where T : notnull + where TProgress : notnull + { + if (extractor is null) + { + throw new ArgumentNullException(nameof(extractor)); + } + + return EtlPipelineImpl.FromStream((_, token) => extractor.ExtractAsync(token)); + } +} diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs new file mode 100644 index 00000000..44e78535 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs @@ -0,0 +1,129 @@ +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; + + +namespace Wolfgang.Etl.Abstractions; + +/// +/// Internal implementation. A pipeline is a lazily-composed factory +/// that, given a per-run and a , produces +/// the record stream. Each appended stage returns a new instance wrapping the previous factory; +/// nothing runs until the terminal sink enumerates the stream. +/// +internal sealed class EtlPipelineImpl : IEtlPipeline + where T : notnull +{ + private readonly Func> _factory; + + + private EtlPipelineImpl(Func> factory) + { + _factory = factory; + } + + + /// + /// Wraps a raw source factory with the extracted-record counter, producing the head of a pipeline. + /// + internal static EtlPipelineImpl FromStream + ( + Func> rawSource + ) + { + return new EtlPipelineImpl((state, token) => CountExtracted(rawSource(state, token), state, token)); + } + + + /// + public IEtlPipeline Through(ITransformAsync transformer) + where TOut : notnull + { + if (transformer is null) + { + throw new ArgumentNullException(nameof(transformer)); + } + + return new EtlPipelineImpl((state, token) => transformer.TransformAsync(_factory(state, token))); + } + + + /// + public IEtlPipeline Through(ITransformWithCancellationAsync transformer) + where TOut : notnull + { + if (transformer is null) + { + throw new ArgumentNullException(nameof(transformer)); + } + + return new EtlPipelineImpl((state, token) => transformer.TransformAsync(_factory(state, token), token)); + } + + + /// + public IEtlPipeline Through(Func, IAsyncEnumerable> stage) + where TOut : notnull + { + if (stage is null) + { + throw new ArgumentNullException(nameof(stage)); + } + + return new EtlPipelineImpl((state, token) => stage(_factory(state, token))); + } + + + /// + public IEtlPipeline Through(Func, CancellationToken, IAsyncEnumerable> stage) + where TOut : notnull + { + if (stage is null) + { + throw new ArgumentNullException(nameof(stage)); + } + + return new EtlPipelineImpl((state, token) => stage(_factory(state, token), token)); + } + + + /// + public IEtlPipelineSink To(LoaderBase loader) + where TProgress : notnull + { + if (loader is null) + { + throw new ArgumentNullException(nameof(loader)); + } + + return new EtlPipelineSink(_factory, loader); + } + + + /// + public IAsyncEnumerable AsAsyncEnumerable(CancellationToken token = default) + { + return _factory(new EtlRunState(), token); + } + + + // The head of every pipeline: pulls from the raw source, honours cancellation, and counts each + // record as extracted. WithCancellation covers sources that observe the token via + // [EnumeratorCancellation]; the explicit throw covers sources that ignore it. + private static async IAsyncEnumerable CountExtracted + ( + IAsyncEnumerable source, + EtlRunState state, + [EnumeratorCancellation] CancellationToken token + ) + { + await foreach (var item in source.WithCancellation(token).ConfigureAwait(false)) + { + token.ThrowIfCancellationRequested(); + state.RecordsExtracted++; + yield return item; + } + } +} diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineProgress.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineProgress.cs new file mode 100644 index 00000000..ed6d2a65 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineProgress.cs @@ -0,0 +1,27 @@ +using System; + + +namespace Wolfgang.Etl.Abstractions; + +/// +/// An immutable snapshot of a generic run's throughput, reported through +/// the sink passed to +/// . +/// +/// +/// The counters describe the ends of the pipeline — records pulled from the source and records +/// delivered to the sink — because those are the only points the core pipeline observes. Operators +/// (filtering, projection, batching) live in higher-level packages and change the count between the +/// two ends without the core needing to know how. For rich, stage-specific progress attach an +/// observer directly to the extractor or loader before handing it to the +/// pipeline. +/// +/// The number of records pulled from the source so far. +/// The number of records delivered to the sink so far. +/// The wall-clock time elapsed since the run started. +public sealed record EtlPipelineProgress +( + int RecordsExtracted, + int RecordsLoaded, + TimeSpan Elapsed +); diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs new file mode 100644 index 00000000..3003a428 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs @@ -0,0 +1,62 @@ +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; + + +namespace Wolfgang.Etl.Abstractions; + +/// +/// Internal that feeds the composed record stream into a +/// , counting delivered records and reporting +/// along the way. +/// +internal sealed class EtlPipelineSink : IEtlPipelineSink + where T : notnull + where TProgress : notnull +{ + private readonly Func> _factory; + private readonly LoaderBase _loader; + + + internal EtlPipelineSink + ( + Func> factory, + LoaderBase loader + ) + { + _factory = factory; + _loader = loader; + } + + + /// + public async Task RunAsync(IProgress? progress = null, CancellationToken token = default) + { + var state = new EtlRunState(); + var stream = CountLoaded(_factory(state, token), state, progress, token); + + await _loader.LoadAsync(stream, token).ConfigureAwait(false); + + progress?.Report(state.Snapshot()); + } + + + private static async IAsyncEnumerable CountLoaded + ( + IAsyncEnumerable stream, + EtlRunState state, + IProgress? progress, + [EnumeratorCancellation] CancellationToken token + ) + { + await foreach (var item in stream.ConfigureAwait(false)) + { + token.ThrowIfCancellationRequested(); + state.RecordsLoaded++; + progress?.Report(state.Snapshot()); + yield return item; + } + } +} diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSource.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSource.cs new file mode 100644 index 00000000..193bb589 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSource.cs @@ -0,0 +1,15 @@ +namespace Wolfgang.Etl.Abstractions; + +/// +/// A sentinel target for source-factory extension methods. C# cannot attach extension methods to a +/// static class, so exposes an instance of this type for format +/// packages to extend — for example a CSV package adds +/// public static ICsvExtractorBuilder<T> CsvExtractor<T>(this EtlPipelineSource source, string path), +/// enabling EtlPipeline.Source.CsvExtractor<Order>("orders.csv"). +/// +public sealed class EtlPipelineSource +{ + internal EtlPipelineSource() + { + } +} diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs new file mode 100644 index 00000000..d92fd26c --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs @@ -0,0 +1,25 @@ +using System.Diagnostics; + + +namespace Wolfgang.Etl.Abstractions; + +/// +/// Per-run counter bag for a generic ETL pipeline. Created when a run starts and threaded through the +/// factory chain so the source can count extracted records and the sink can count loaded records. +/// A pipeline is enumerated by a single consumer one item at a time, so the counters are only ever +/// touched on one logical flow and need no synchronization. +/// +internal sealed class EtlRunState +{ + private readonly Stopwatch _stopwatch = Stopwatch.StartNew(); + + public int RecordsExtracted; + + public int RecordsLoaded; + + + public EtlPipelineProgress Snapshot() + { + return new EtlPipelineProgress(RecordsExtracted, RecordsLoaded, _stopwatch.Elapsed); + } +} diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/IEtlPipeline.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/IEtlPipeline.cs new file mode 100644 index 00000000..387a0b36 --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/IEtlPipeline.cs @@ -0,0 +1,102 @@ +using System; +using System.Collections.Generic; +using System.Threading; + + +namespace Wolfgang.Etl.Abstractions; + +/// +/// A generic, format-agnostic ETL pipeline carrying a stream of items of type +/// . Append transformer stages with , +/// then terminate with (or a format-specific +/// sink terminator) to obtain a runnable . +/// +/// The type of item currently flowing through the pipeline. +/// +/// +/// This is the minimal pipeline core. It knows how to pull from a source, pipe through transformers, +/// and push to a sink — nothing more. The LINQ-flavored operators (Where, Select, +/// Distinct, Take, Buffer, …) are extension methods that layer on top +/// of and are shipped by +/// Wolfgang.Etl.Transformers, which already owns those transformers — the core does not +/// re-implement them. +/// +/// +/// The pipeline is lazy: appending a stage builds up a description of the work and nothing executes +/// until +/// is called (or the stream is enumerated via ). +/// +/// +public interface IEtlPipeline + where T : notnull +{ + /// + /// Appends a transformer stage, piping the current stream through it. This is the primitive on + /// which higher-level operators are built. + /// + /// The type produced by the transformer. + /// The transformer to append. + /// A pipeline carrying . + /// is . + IEtlPipeline Through(ITransformAsync transformer) where TOut : notnull; + + + /// + /// Appends a cancellation-aware transformer stage. The pipeline's cancellation token is forwarded + /// into the transformer. + /// + /// The type produced by the transformer. + /// The transformer to append. + /// A pipeline carrying . + /// is . + IEtlPipeline Through(ITransformWithCancellationAsync transformer) where TOut : notnull; + + + /// + /// Appends a stream-to-stream transformer stage supplied as a delegate — the same contract as + /// , + /// but inline, without declaring a class. Note this transforms the whole stream; a + /// per-element projection (Select) is an operator provided by Wolfgang.Etl.Transformers, + /// not the core. + /// + /// The type produced by the stage. + /// The stream-to-stream transform to append. + /// A pipeline carrying . + /// is . + IEtlPipeline Through(Func, IAsyncEnumerable> stage) where TOut : notnull; + + + /// + /// Appends a cancellation-aware stream-to-stream transformer stage supplied as a delegate. The + /// pipeline's cancellation token is forwarded into the delegate. + /// + /// The type produced by the stage. + /// The stream-to-stream transform to append. + /// A pipeline carrying . + /// is . + IEtlPipeline Through(Func, CancellationToken, IAsyncEnumerable> stage) where TOut : notnull; + + + /// + /// Terminates the pipeline with a loader. Format-specific sink terminators (for example + /// CsvLoader or SqlBulkCopyLoader) are extension methods shipped by their own + /// packages; this generic overload is the catch-all for callers who already hold a + /// . + /// + /// The loader's progress-report type. + /// The loader that consumes the pipeline output. + /// A runnable . + /// is . + IEtlPipelineSink To(LoaderBase loader) where TProgress : notnull; + + + /// + /// Drops down to the raw — the escape hatch for consumers who + /// want to apply System.Linq.Async operators directly or otherwise leave the pipeline. + /// Enumerating the returned stream runs the pipeline (without a sink, so + /// is not tracked). + /// + /// A cancellation token observed while enumerating. + /// The composed record stream. + IAsyncEnumerable AsAsyncEnumerable(CancellationToken token = default); +} diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/IEtlPipelineSink.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/IEtlPipelineSink.cs new file mode 100644 index 00000000..540d324e --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/IEtlPipelineSink.cs @@ -0,0 +1,30 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + + +namespace Wolfgang.Etl.Abstractions; + +/// +/// A terminated, runnable generic ETL pipeline. Obtained from +/// or a format-specific sink +/// terminator (for example a CsvLoader extension shipped by a format package). +/// +public interface IEtlPipelineSink +{ + /// + /// Runs the pipeline to completion: pulls records from the source, threads them through every + /// operator, and delivers them to the sink. + /// + /// + /// An optional sink for snapshots. When supplied, a snapshot is + /// reported as records are delivered to the loader and once more when the run completes. Pass + /// to skip progress reporting. + /// + /// + /// A cancellation token observed while pulling from the source and threading records through the + /// pipeline. It is also forwarded to the sink. + /// + /// A task that completes when the loader has finished consuming the stream. + Task RunAsync(IProgress? progress = null, CancellationToken token = default); +} diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt index 31b06ed9..aa61c460 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt +++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt @@ -1,4 +1,26 @@ #nullable enable +Wolfgang.Etl.Abstractions.EtlPipeline +Wolfgang.Etl.Abstractions.EtlPipelineProgress +Wolfgang.Etl.Abstractions.EtlPipelineProgress.Elapsed.get -> System.TimeSpan +Wolfgang.Etl.Abstractions.EtlPipelineProgress.Elapsed.init -> void +Wolfgang.Etl.Abstractions.EtlPipelineProgress.EtlPipelineProgress(int RecordsExtracted, int RecordsLoaded, System.TimeSpan Elapsed) -> void +Wolfgang.Etl.Abstractions.EtlPipelineProgress.RecordsExtracted.get -> int +Wolfgang.Etl.Abstractions.EtlPipelineProgress.RecordsExtracted.init -> void +Wolfgang.Etl.Abstractions.EtlPipelineProgress.RecordsLoaded.get -> int +Wolfgang.Etl.Abstractions.EtlPipelineProgress.RecordsLoaded.init -> void +Wolfgang.Etl.Abstractions.EtlPipelineSource +Wolfgang.Etl.Abstractions.IEtlPipeline +Wolfgang.Etl.Abstractions.IEtlPipeline.AsAsyncEnumerable(System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable! +Wolfgang.Etl.Abstractions.IEtlPipeline.Through(System.Func!, System.Collections.Generic.IAsyncEnumerable!>! stage) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +Wolfgang.Etl.Abstractions.IEtlPipeline.Through(System.Func!, System.Threading.CancellationToken, System.Collections.Generic.IAsyncEnumerable!>! stage) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +Wolfgang.Etl.Abstractions.IEtlPipeline.Through(Wolfgang.Etl.Abstractions.ITransformAsync! transformer) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +Wolfgang.Etl.Abstractions.IEtlPipeline.Through(Wolfgang.Etl.Abstractions.ITransformWithCancellationAsync! transformer) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +Wolfgang.Etl.Abstractions.IEtlPipeline.To(Wolfgang.Etl.Abstractions.LoaderBase! loader) -> Wolfgang.Etl.Abstractions.IEtlPipelineSink! +Wolfgang.Etl.Abstractions.IEtlPipelineSink +Wolfgang.Etl.Abstractions.IEtlPipelineSink.RunAsync(System.IProgress? progress = null, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! Wolfgang.Etl.Abstractions.ISupportDryRun Wolfgang.Etl.Abstractions.ISupportDryRun.IsDryRun.get -> bool Wolfgang.Etl.Abstractions.ISupportDryRun.IsDryRun.set -> void +static Wolfgang.Etl.Abstractions.EtlPipeline.From(Wolfgang.Etl.Abstractions.ExtractorBase! extractor) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +static Wolfgang.Etl.Abstractions.EtlPipeline.From(System.Collections.Generic.IAsyncEnumerable! source) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +static Wolfgang.Etl.Abstractions.EtlPipeline.Source.get -> Wolfgang.Etl.Abstractions.EtlPipelineSource! diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs new file mode 100644 index 00000000..b385d11c --- /dev/null +++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs @@ -0,0 +1,278 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Wolfgang.Etl.Abstractions.Tests.Unit.BaseClassTests; +using Wolfgang.Etl.Abstractions.Tests.Unit.Models; +using Xunit; + +namespace Wolfgang.Etl.Abstractions.Tests.Unit.EtlPipelineTests; + +public class EtlPipelineTests +{ + private static async IAsyncEnumerable AsyncSource(params T[] items) + { + foreach (var item in items) + { + await Task.Yield(); + yield return item; + } + } + + + private static async Task> Collect(IAsyncEnumerable source, CancellationToken token = default) + { + var result = new List(); + await foreach (var item in source.WithCancellation(token)) + { + result.Add(item); + } + + return result; + } + + + // A stream-to-stream transform used to exercise the delegate Through overload + // (iterator lambdas aren't allowed in C#, so this is a method). + private static async IAsyncEnumerable Label(IAsyncEnumerable items) + { + await foreach (var item in items) + { + yield return $"#{item}"; + } + } + + + [Fact] + public async Task RunAsync_when_source_is_IAsyncEnumerable_delivers_all_records_to_the_loader() + { + var loader = new CollectingLoader(); + + await EtlPipeline.From(AsyncSource(1, 2, 3)) + .To(loader) + .RunAsync(); + + Assert.Equal(new[] { 1, 2, 3 }, loader.Loaded); + } + + + [Fact] + public async Task RunAsync_when_source_is_an_extractor_delivers_all_records_to_the_loader() + { + var loader = new CollectingLoader(); + var extractor = new SeededExtractor(new[] { 10, 20, 30 }); + + await EtlPipeline.From(extractor) + .To(loader) + .RunAsync(); + + Assert.Equal(new[] { 10, 20, 30 }, loader.Loaded); + } + + + [Fact] + public async Task Through_pipes_records_through_the_transformer() + { + var loader = new CollectingLoader(); + + await EtlPipeline.From(AsyncSource(1, 2, 3)) + .Through(new MapTransformer(x => $"n{x}")) + .To(loader) + .RunAsync(); + + Assert.Equal(new[] { "n1", "n2", "n3" }, loader.Loaded); + } + + + [Fact] + public async Task Through_can_be_chained() + { + var loader = new CollectingLoader(); + + await EtlPipeline.From(AsyncSource(1, 2, 3)) + .Through(new MapTransformer(x => x + 1)) + .Through(new MapTransformer(x => x * 10)) + .To(loader) + .RunAsync(); + + Assert.Equal(new[] { 20, 30, 40 }, loader.Loaded); + } + + + [Fact] + public async Task Through_forwards_the_cancellation_token_to_a_cancellation_aware_transformer() + { + using var cts = new CancellationTokenSource(); + var transformer = new TokenRecordingTransformer(); + var loader = new CollectingLoader(); + + await EtlPipeline.From(AsyncSource(1, 2, 3)) + .Through(transformer) + .To(loader) + .RunAsync(null, cts.Token); + + Assert.Equal(cts.Token, transformer.LastToken); + Assert.Equal(new[] { 1, 2, 3 }, loader.Loaded); + } + + + [Fact] + public async Task Through_delegate_pipes_records_through_the_stage() + { + var loader = new CollectingLoader(); + + await EtlPipeline.From(AsyncSource(1, 2, 3)) + .Through(Label) // Func, IAsyncEnumerable> + .To(loader) + .RunAsync(); + + Assert.Equal(new[] { "#1", "#2", "#3" }, loader.Loaded); + } + + + [Fact] + public async Task Through_cancellation_aware_delegate_forwards_the_token() + { + using var cts = new CancellationTokenSource(); + var captured = CancellationToken.None; + var loader = new CollectingLoader(); + + // Identity stage that records the token it is handed. + Func, CancellationToken, IAsyncEnumerable> stage = (items, token) => + { + captured = token; + return items; + }; + + await EtlPipeline.From(AsyncSource(1, 2, 3)) + .Through(stage) + .To(loader) + .RunAsync(null, cts.Token); + + Assert.Equal(cts.Token, captured); + Assert.Equal(new[] { 1, 2, 3 }, loader.Loaded); + } + + + [Fact] + public void Through_when_delegate_is_null_throws_ArgumentNullException() + { + var pipeline = EtlPipeline.From(AsyncSource(1)); + Assert.Throws( + () => pipeline.Through((Func, IAsyncEnumerable>)null!)); + } + + + [Fact] + public void Through_when_cancellation_aware_delegate_is_null_throws_ArgumentNullException() + { + var pipeline = EtlPipeline.From(AsyncSource(1)); + Assert.Throws( + () => pipeline.Through((Func, CancellationToken, IAsyncEnumerable>)null!)); + } + + + [Fact] + public async Task AsAsyncEnumerable_exposes_the_composed_stream() + { + var stream = EtlPipeline.From(AsyncSource(1, 2, 3)) + .Through(new MapTransformer(x => x * 100)) + .AsAsyncEnumerable(); + + var result = await Collect(stream); + + Assert.Equal(new[] { 100, 200, 300 }, result); + } + + + [Fact] + public async Task RunAsync_propagates_an_exception_thrown_by_a_transformer() + { + var loader = new CollectingLoader(); + + var sink = EtlPipeline.From(AsyncSource(1, 2, 3)) + .Through(new MapTransformer(_ => throw new InvalidOperationException("boom"))) + .To(loader); + + var ex = await Assert.ThrowsAsync(() => sink.RunAsync()); + Assert.Equal("boom", ex.Message); + } + + + [Fact] + public async Task RunAsync_observes_cancellation_mid_stream() + { + using var cts = new CancellationTokenSource(); + var loader = new CollectingLoader(); + + var sink = EtlPipeline.From(AsyncSource(1, 2, 3, 4, 5)) + .Through(new CancelingTransformer(cts, 2)) + .To(loader); + + await Assert.ThrowsAnyAsync(() => sink.RunAsync(null, cts.Token)); + Assert.True(loader.Loaded.Count < 5); + } + + + [Fact] + public async Task RunAsync_reports_extracted_and_loaded_counters() + { + var reports = new List(); + var progress = new SynchronousProgress(reports.Add); + var loader = new CollectingLoader(); + + await EtlPipeline.From(AsyncSource(1, 2, 3, 4, 5)) + .To(loader) + .RunAsync(progress); + + var final = reports.Last(); + Assert.Equal(5, final.RecordsExtracted); + Assert.Equal(5, final.RecordsLoaded); + } + + + [Fact] + public void Source_sentinel_is_available_for_format_package_extensions() + { + Assert.NotNull(EtlPipeline.Source); + } + + + [Fact] + public void From_when_stream_is_null_throws_ArgumentNullException() + { + Assert.Throws(() => EtlPipeline.From((IAsyncEnumerable)null!)); + } + + + [Fact] + public void From_when_extractor_is_null_throws_ArgumentNullException() + { + Assert.Throws(() => EtlPipeline.From(null!)); + } + + + [Fact] + public void Through_when_transformer_is_null_throws_ArgumentNullException() + { + var pipeline = EtlPipeline.From(AsyncSource(1)); + Assert.Throws(() => pipeline.Through((ITransformAsync)null!)); + } + + + [Fact] + public void Through_when_cancellation_aware_transformer_is_null_throws_ArgumentNullException() + { + var pipeline = EtlPipeline.From(AsyncSource(1)); + Assert.Throws(() => pipeline.Through((ITransformWithCancellationAsync)null!)); + } + + + [Fact] + public void To_when_loader_is_null_throws_ArgumentNullException() + { + var pipeline = EtlPipeline.From(AsyncSource(1)); + Assert.Throws(() => pipeline.To(null!)); + } +} diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/TestDoubles.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/TestDoubles.cs new file mode 100644 index 00000000..18e472c8 --- /dev/null +++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/TestDoubles.cs @@ -0,0 +1,161 @@ +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Wolfgang.Etl.Abstractions.Tests.Unit.Models; + +namespace Wolfgang.Etl.Abstractions.Tests.Unit.EtlPipelineTests; + +/// +/// Minimal that collects everything it is handed, +/// for exercising . +/// +internal sealed class CollectingLoader : LoaderBase + where T : notnull +{ + public List Loaded { get; } = new(); + + + protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token) + { + await foreach (var item in items.WithCancellation(token)) + { + token.ThrowIfCancellationRequested(); + Loaded.Add(item); + IncrementCurrentItemCount(); + } + } + + + protected override EtlProgress CreateProgressReport() + { + return new EtlProgress(CurrentItemCount); + } +} + + +/// +/// Minimal that yields a fixed sequence, for +/// exercising . +/// +internal sealed class SeededExtractor : ExtractorBase + where T : notnull +{ + private readonly IEnumerable _items; + + + public SeededExtractor(IEnumerable items) + { + _items = items; + } + + + protected override async IAsyncEnumerable ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token) + { + foreach (var item in _items) + { + token.ThrowIfCancellationRequested(); + yield return item; + IncrementCurrentItemCount(); + await Task.Yield(); + } + } + + + protected override EtlProgress CreateProgressReport() + { + return new EtlProgress(CurrentItemCount); + } +} + + +/// +/// A projecting for exercising +/// . +/// +internal sealed class MapTransformer : ITransformAsync + where TSource : notnull + where TDestination : notnull +{ + private readonly Func _map; + + + public MapTransformer(Func map) + { + _map = map; + } + + + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable items) + { + await foreach (var item in items) + { + yield return _map(item); + } + } +} + + +/// +/// A cancellation-aware pass-through transformer that records the token it was handed, for verifying +/// that forwards +/// the run's cancellation token. +/// +internal sealed class TokenRecordingTransformer : ITransformWithCancellationAsync + where T : notnull +{ + public CancellationToken LastToken { get; private set; } + + + public IAsyncEnumerable TransformAsync(IAsyncEnumerable items) + { + return TransformAsync(items, CancellationToken.None); + } + + + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable items, [EnumeratorCancellation] CancellationToken token) + { + LastToken = token; + await foreach (var item in items.WithCancellation(token)) + { + token.ThrowIfCancellationRequested(); + yield return item; + } + } +} + + +/// +/// A pass-through transformer that cancels the supplied source after emitting a set number of records, +/// for exercising mid-stream cancellation. +/// +internal sealed class CancelingTransformer : ITransformAsync + where T : notnull +{ + private readonly CancellationTokenSource _cts; + private readonly int _cancelAfter; + + + public CancelingTransformer(CancellationTokenSource cts, int cancelAfter) + { + _cts = cts; + _cancelAfter = cancelAfter; + } + + + public async IAsyncEnumerable TransformAsync(IAsyncEnumerable items) + { + var count = 0; + await foreach (var item in items) + { + count++; + if (count == _cancelAfter) + { + _cts.Cancel(); + } + + yield return item; + } + } +}