Composable per-item middleware / interceptor (#93)

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Middleware / interceptor (#93):** a composable per-item hook — `IItemMiddleware<T>` returning
+  `MiddlewareResult<T>` (`Continue` to keep/replace an item, `Drop` to remove it) — attached to any
+  stream with the `WithMiddleware(...)` extensions (single or ordered chain). Lets cross-cutting
+  concerns (logging, validation, metrics, throttling, dedup) decorate an extractor/transformer output
+  or loader input without changing the component, and composes inside an `EtlPipeline` via
+  `Through(s => s.WithMiddleware(...))`. Dependency-free.
 - **Retry seam (#94):** `ExtractorBase`, `LoaderBase`, and `TransformerBase` gained a
   `protected virtual WrapWorkerExecution(...)` hook wrapped around every worker invocation. The
   default implementation is a no-op, so behaviour is unchanged; override it to run the worker through

src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs

@@ -0,0 +1,25 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// A composable, reusable hook for cross-cutting per-item behaviour (logging, validation, metrics,
+/// throttling, deduplication) that can be attached to any stream without modifying the extractor,
+/// transformer, or loader that produced it. Attach one or more with
+/// <see cref="MiddlewareExtensions.WithMiddleware{T}(System.Collections.Generic.IAsyncEnumerable{T}, IItemMiddleware{T}, CancellationToken)"/>;
+/// they run in the order attached, each seeing the item the previous one passed on.
+/// </summary>
+/// <typeparam name="T">The item type flowing through the pipeline.</typeparam>
+public interface IItemMiddleware<T>
+{
+    /// <summary>
+    /// Invoked once per item. Return <see cref="MiddlewareResult.Continue{T}(T)"/> to keep the item
+    /// flowing (optionally replacing it), or <see cref="MiddlewareResult.Drop{T}"/> to remove it from
+    /// the stream.
+    /// </summary>
+    /// <param name="item">The item to process.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The outcome describing whether to keep or drop the item.</returns>
+    ValueTask<MiddlewareResult<T>> OnItemAsync(T item, CancellationToken token);
+}

src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs

@@ -0,0 +1,122 @@
+using System;
+using System.Collections.Generic;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// Extension methods that attach <see cref="IItemMiddleware{T}"/> to an
+/// <see cref="IAsyncEnumerable{T}"/> stream, so cross-cutting per-item behaviour composes onto any
+/// extractor / transformer output or loader input — and inside an <c>EtlPipeline</c> via
+/// <c>Through(stream =&gt; stream.WithMiddleware(...))</c> — without changing the component itself.
+/// </summary>
+public static class MiddlewareExtensions
+{
+    /// <summary>
+    /// Pipes every item of <paramref name="source"/> through <paramref name="middleware"/>. Items the
+    /// middleware drops (<see cref="MiddlewareResult.Drop{T}"/>) are removed from the stream; otherwise
+    /// the (possibly replaced) item is yielded.
+    /// </summary>
+    /// <typeparam name="T">The item type.</typeparam>
+    /// <param name="source">The stream to decorate.</param>
+    /// <param name="middleware">The middleware to run for each item.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The decorated stream.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="middleware"/> is <see langword="null"/>.</exception>
+    public static async IAsyncEnumerable<T> WithMiddleware<T>
+    (
+        this IAsyncEnumerable<T> source,
+        IItemMiddleware<T> middleware,
+        [EnumeratorCancellation] CancellationToken token = default
+    )
+    {
+        if (source is null)
+        {
+            throw new ArgumentNullException(nameof(source));
+        }
+
+        if (middleware is null)
+        {
+            throw new ArgumentNullException(nameof(middleware));
+        }
+
+        await foreach (var item in source.WithCancellation(token))
+        {
+            // Stryker disable once Boolean: equivalent — with no synchronization context in play, ConfigureAwait(false) and (true) are indistinguishable.
+            var result = await middleware.OnItemAsync(item, token).ConfigureAwait(false);
+            if (!result.Skip)
+            {
+                yield return result.Item;
+            }
+        }
+    }
+
+
+
+    /// <summary>
+    /// Pipes every item of <paramref name="source"/> through <paramref name="middlewares"/> in order:
+    /// each middleware sees the item the previous one passed on. If any middleware drops the item
+    /// (<see cref="MiddlewareResult.Drop{T}"/>), the remaining middleware is not run and the item is
+    /// removed from the stream.
+    /// </summary>
+    /// <typeparam name="T">The item type.</typeparam>
+    /// <param name="source">The stream to decorate.</param>
+    /// <param name="middlewares">The middleware chain, applied in enumeration order.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The decorated stream.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="middlewares"/> is <see langword="null"/>, or a member of <paramref name="middlewares"/> is <see langword="null"/>.</exception>
+    public static async IAsyncEnumerable<T> WithMiddleware<T>
+    (
+        this IAsyncEnumerable<T> source,
+        IEnumerable<IItemMiddleware<T>> middlewares,
+        [EnumeratorCancellation] CancellationToken token = default
+    )
+    {
+        if (source is null)
+        {
+            throw new ArgumentNullException(nameof(source));
+        }
+
+        if (middlewares is null)
+        {
+            throw new ArgumentNullException(nameof(middlewares));
+        }
+
+        // Snapshot the chain once so the same ordered set runs for every item.
+        var chain = new List<IItemMiddleware<T>>(middlewares);
+        foreach (var middleware in chain)
+        {
+            if (middleware is null)
+            {
+                // Stryker disable once String: the exception message is diagnostic-only, not a behavioural contract asserted by tests.
+                throw new ArgumentNullException(nameof(middlewares), "A middleware in the chain is null.");
+            }
+        }
+
+        await foreach (var item in source.WithCancellation(token))
+        {
+            var current = item;
+            var dropped = false;
+
+            foreach (var middleware in chain)
+            {
+                // Stryker disable once Boolean: equivalent — with no synchronization context in play, ConfigureAwait(false) and (true) are indistinguishable.
+                var result = await middleware.OnItemAsync(current, token).ConfigureAwait(false);
+                if (result.Skip)
+                {
+                    dropped = true;
+                    break;
+                }
+
+                current = result.Item;
+            }
+
+            if (!dropped)
+            {
+                yield return current;
+            }
+        }
+    }
+}

src/Wolfgang.Etl.Abstractions/MiddlewareResult.Factory.cs

@@ -0,0 +1,25 @@
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// Factory methods for creating <see cref="MiddlewareResult{T}"/> values from an
+/// <see cref="IItemMiddleware{T}"/> implementation.
+/// </summary>
+public static class MiddlewareResult
+{
+    /// <summary>
+    /// Keeps the item in the stream, optionally replacing it with a transformed value.
+    /// </summary>
+    /// <typeparam name="T">The item type.</typeparam>
+    /// <param name="item">The item to pass on.</param>
+    /// <returns>A result that keeps <paramref name="item"/> flowing.</returns>
+    public static MiddlewareResult<T> Continue<T>(T item) => new(item, skip: false);
+
+
+
+    /// <summary>
+    /// Drops the current item from the stream.
+    /// </summary>
+    /// <typeparam name="T">The item type.</typeparam>
+    /// <returns>A result that discards the current item.</returns>
+    public static MiddlewareResult<T> Drop<T>() => new(default!, skip: true);
+}

src/Wolfgang.Etl.Abstractions/MiddlewareResult.cs

@@ -0,0 +1,64 @@
+using System;
+using System.Collections.Generic;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// The outcome of running a single item through an <see cref="IItemMiddleware{T}"/>: the
+/// (possibly replaced) item to pass on, and whether the item should be dropped from the stream.
+/// Create one with <see cref="MiddlewareResult.Continue{T}(T)"/> to keep an item flowing or
+/// <see cref="MiddlewareResult.Drop{T}"/> to discard it.
+/// </summary>
+/// <typeparam name="T">The item type flowing through the pipeline.</typeparam>
+public readonly struct MiddlewareResult<T> : IEquatable<MiddlewareResult<T>>
+{
+    internal MiddlewareResult(T item, bool skip)
+    {
+        Item = item;
+        Skip = skip;
+    }
+
+
+
+    /// <summary>
+    /// The item to pass on to the next middleware (or to the stream). Meaningful only when
+    /// <see cref="Skip"/> is <see langword="false"/>.
+    /// </summary>
+    public T Item { get; }
+
+
+
+    /// <summary>
+    /// <see langword="true"/> to drop the item from the stream (later middleware is not run and the
+    /// item is not yielded); <see langword="false"/> to keep it.
+    /// </summary>
+    public bool Skip { get; }
+
+
+
+    /// <inheritdoc/>
+    public bool Equals(MiddlewareResult<T> other) =>
+        Skip == other.Skip && EqualityComparer<T>.Default.Equals(Item, other.Item);
+
+
+
+    /// <inheritdoc/>
+    public override bool Equals(object? obj) => obj is MiddlewareResult<T> other && Equals(other);
+
+
+
+    /// <inheritdoc/>
+    // Stryker disable once all: equivalent — any change to the hash formula still yields equal hash codes for equal values (the only GetHashCode contract), so no behavioural test can distinguish it.
+    public override int GetHashCode() =>
+        unchecked(((Skip ? 1 : 0) * 397) ^ (Item is null ? 0 : EqualityComparer<T>.Default.GetHashCode(Item)));
+
+
+
+    /// <summary>Indicates whether two results are equal.</summary>
+    public static bool operator ==(MiddlewareResult<T> left, MiddlewareResult<T> right) => left.Equals(right);
+
+
+
+    /// <summary>Indicates whether two results are not equal.</summary>
+    public static bool operator !=(MiddlewareResult<T> left, MiddlewareResult<T> right) => !left.Equals(right);
+}

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

@@ -3,3 +3,20 @@ virtual Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.WrapWorkerEx
 virtual Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.WrapWorkerExecution(System.Func<System.Threading.CancellationToken, System.Threading.Tasks.Task!>! workerFactory, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task!
 virtual Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.WrapWorkerExecution(System.Func<System.Threading.CancellationToken, System.Collections.Generic.IAsyncEnumerable<TDestination>!>! workerFactory, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable<TDestination>!
 Wolfgang.Etl.Abstractions.Report.Report(int currentItemCount, System.DateTimeOffset? startedAt, System.TimeSpan elapsed, int? totalItemCount = null) -> void
+Wolfgang.Etl.Abstractions.IItemMiddleware<T>
+Wolfgang.Etl.Abstractions.IItemMiddleware<T>.OnItemAsync(T item, System.Threading.CancellationToken token) -> System.Threading.Tasks.ValueTask<Wolfgang.Etl.Abstractions.MiddlewareResult<T>>
+Wolfgang.Etl.Abstractions.MiddlewareResult
+static Wolfgang.Etl.Abstractions.MiddlewareResult.Continue<T>(T item) -> Wolfgang.Etl.Abstractions.MiddlewareResult<T>
+static Wolfgang.Etl.Abstractions.MiddlewareResult.Drop<T>() -> Wolfgang.Etl.Abstractions.MiddlewareResult<T>
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>.MiddlewareResult() -> void
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>.Item.get -> T
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>.Skip.get -> bool
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>.Equals(Wolfgang.Etl.Abstractions.MiddlewareResult<T> other) -> bool
+override Wolfgang.Etl.Abstractions.MiddlewareResult<T>.Equals(object? obj) -> bool
+override Wolfgang.Etl.Abstractions.MiddlewareResult<T>.GetHashCode() -> int
+static Wolfgang.Etl.Abstractions.MiddlewareResult<T>.operator ==(Wolfgang.Etl.Abstractions.MiddlewareResult<T> left, Wolfgang.Etl.Abstractions.MiddlewareResult<T> right) -> bool
+static Wolfgang.Etl.Abstractions.MiddlewareResult<T>.operator !=(Wolfgang.Etl.Abstractions.MiddlewareResult<T> left, Wolfgang.Etl.Abstractions.MiddlewareResult<T> right) -> bool
+Wolfgang.Etl.Abstractions.MiddlewareExtensions
+static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware<T>(this System.Collections.Generic.IAsyncEnumerable<T>! source, Wolfgang.Etl.Abstractions.IItemMiddleware<T>! middleware, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<T>!
+static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware<T>(this System.Collections.Generic.IAsyncEnumerable<T>! source, System.Collections.Generic.IEnumerable<Wolfgang.Etl.Abstractions.IItemMiddleware<T>!>! middlewares, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<T>!