feat: streaming query pipeline (IAsyncEnumerable, ExecuteStream, resilience) — closes #299

.gitignore

@@ -264,3 +264,9 @@ paket-files/
 
 # Benchmark.Net
 BenchmarkDotNet.Artifacts/
+
+# macOS
+.DS_Store
+
+# JetBrains Junie
+.junie/

Directory.Packages.props

@@ -5,15 +5,16 @@
   </PropertyGroup>
   <ItemGroup>
     <PackageVersion Include="BenchmarkDotNet" Version="0.15.8" />
+    <PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="10.0.9" />
     <PackageVersion Include="Microsoft.CSharp" Version="4.7.0" />
     <PackageVersion Include="Microsoft.Extensions.Logging" Version="10.0.9" />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="10.0.9" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="10.0.9" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Debug" Version="10.0.9" />
     <PackageVersion Include="Microsoft.Maui.Controls" Version="$(MauiVersion)" />
-    <PackageVersion Include="OpenTelemetry" Version="1.11.0" />
-    <PackageVersion Include="OpenTelemetry.Exporter.InMemory" Version="1.11.0" />
-    <PackageVersion Include="OpenTelemetry.Extensions.Hosting" Version="1.11.0" />
+    <PackageVersion Include="OpenTelemetry" Version="1.16.0" />
+    <PackageVersion Include="OpenTelemetry.Exporter.InMemory" Version="1.16.0" />
+    <PackageVersion Include="OpenTelemetry.Extensions.Hosting" Version="1.16.0" />
     <PackageVersion Include="Polly" Version="8.7.0" />
     <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="18.7.0" />
     <PackageVersion Include="Shouldly" Version="4.3.0" />

README.md

@@ -115,3 +115,108 @@ IQueryProcessor queryProcessor = QueryProcessorBuilder.With()
 Instead of `Activator.CreateInstance`, you can pass any factory `Func<Type, object>` to construct handlers and decorators.
 
 > **Note:** The `Paramore.Darker.SimpleInjector` and `Paramore.Darker.LightInject` packages have been removed as of V5. If you use a third-party DI container, use its built-in adapter for `Microsoft.Extensions.DependencyInjection` and integrate with Darker via the `Paramore.Darker.Extensions.DependencyInjection` package instead.
+
+## Streaming Queries
+
+Darker supports streaming queries that yield results incrementally as `IAsyncEnumerable<TResult>`,
+so large result sets or real-time feeds are produced on demand rather than buffered into memory.
+
+### Define a stream query and handler
+
+```csharp
+using Paramore.Darker;
+using System.Collections.Generic;
+using System.Threading;
+
+// TResult is the item type, not the enumerable.
+public sealed class GetOrdersStream : IStreamQuery<Order>
+{
+    public string CustomerId { get; }
+    public GetOrdersStream(string customerId) => CustomerId = customerId;
+}
+
+public sealed class GetOrdersStreamHandler : StreamQueryHandler<GetOrdersStream, Order>
+{
+    public override async IAsyncEnumerable<Order> ExecuteAsync(
+        GetOrdersStream query,
+        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        await foreach (var order in _repository.StreamByCustomerAsync(query.CustomerId, cancellationToken))
+            yield return order;
+    }
+}
+```
+
+### Execute with `await foreach`
+
+```csharp
+await foreach (var order in queryProcessor.ExecuteStream(new GetOrdersStream("C42"), cancellationToken))
+{
+    // items arrive as the handler produces them — no buffering
+    Process(order);
+}
+```
+
+### Registration with DI (assembly scan)
+
+`AddHandlersFromAssemblies` picks up `IStreamQueryHandler<,>` implementations automatically alongside
+sync and async handlers:
+
+```csharp
+services.AddDarker()
+    .AddHandlersFromAssemblies(typeof(GetOrdersStreamHandler).Assembly);
+```
+
+### Registration with DI (explicit)
+
+```csharp
+services.AddDarker()
+    .AddStreamHandlers(r => r.Register<GetOrdersStream, Order, GetOrdersStreamHandler>());
+```
+
+### Registration without DI
+
+```csharp
+var streamRegistry = new StreamQueryHandlerRegistry();
+streamRegistry.Register<GetOrdersStream, Order, GetOrdersStreamHandler>();
+
+IQueryProcessor queryProcessor = QueryProcessorBuilder.With()
+    .Handlers(new HandlerConfiguration(
+        syncRegistry, handlerFactory, decoratorRegistry, decoratorFactory,
+        asyncRegistry, handlerFactory, decoratorRegistry, decoratorFactory,
+        streamRegistry))
+    .InMemoryQueryContextFactory()
+    .Build();
+```
+
+### Resilience (Polly v8)
+
+Use `[UseResiliencePipelineStream]` — **not** `[RetryableQuery]` or `[FallbackPolicy]`, which apply
+only to single-result handlers and throw a `ConfigurationException` on a stream handler:
+
+```csharp
+public sealed class GetOrdersStreamHandler : StreamQueryHandler<GetOrdersStream, Order>
+{
+    [UseResiliencePipelineStream(1, "MyRetryPipeline")]
+    public override async IAsyncEnumerable<Order> ExecuteAsync(GetOrdersStream query,
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    { ... }
+}
+```
+
+Resilience covers **stream establishment and the first item only**. Once the first item has been
+yielded to the caller, the pipeline has exited and subsequent faults propagate un-retried. A `Timeout`
+strategy therefore bounds *getting the stream started*, not total enumeration time. `Hedging` is not
+supported for streams. These are intentional semantics, not limitations to be worked around.
+
+### Documented semantics
+
+| Behaviour | Detail |
+|---|---|
+| **Laziness** | The framework never buffers the sequence; items are produced on demand. Custom decorators must also avoid buffering (e.g. `ToListAsync`). |
+| **Cancellation** | Pass a `CancellationToken` to `ExecuteStream`; cancelling mid-stream stops enumeration and propagates `OperationCanceledException`. |
+| **Exceptions mid-stream** | Faults during enumeration propagate with their original stack trace — no `TargetInvocationException` wrapper. |
+| **Configuration errors** | A missing or mismatched handler surfaces as `ConfigurationException` from the caller's **first `await foreach` iteration**, not from the `ExecuteStream` call itself (deliberate — resolving eagerly would leak the handler if the caller never enumerates). |
+| **Re-enumeration** | Each `await foreach` over the same `IAsyncEnumerable` re-executes the handler with a fresh pipeline. The stream is cold, not cached. To iterate twice over the same data, buffer it yourself (`await ToListAsync()`). |
+| **Caller-supplied context** | A `queryContext` passed to `ExecuteStream` is scoped to a **single enumeration**. Concurrent or repeated enumeration is only safe when the processor creates the context (pass `null`). |
+| **Legacy attributes** | `[RetryableQuery]` and `[FallbackPolicy]` do **not** apply to streams. Use `[UseResiliencePipelineStream]` for stream resilience. Applying a mismatched attribute throws `ConfigurationException`. |