Runtime: Command Router stream send primitives

DesignPatterns/Behavioral/CommandRouter.cs

@@ -116,6 +116,51 @@ public ValueTask<CommandSendAttempt<TResult>> TrySendAsync<TCommand, TResult>(
         return InvokeResultAsync(handler, command, cancellationToken);
     }
 
+    /// <inheritdoc />
+    public IAsyncEnumerable<TItem> SendStreamAsync<TCommand, TItem>(
+        TCommand command,
+        CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        if (!_handlers.TryGetValue(typeof(TCommand), out var registered))
+        {
+            throw CommandHandlerNotFoundException.ForCommand<TCommand>();
+        }
+
+        if (registered is not IStreamCommandHandler<TCommand, TItem> handler)
+        {
+            throw CreateHandlerContractMismatchException(
+                typeof(TCommand),
+                expected: $"IStreamCommandHandler<{typeof(TCommand).Name}, {typeof(TItem).Name}>");
+        }
+
+        return handler.HandleAsync(command, cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public CommandSendAttempt<IAsyncEnumerable<TItem>> TrySendStreamAsync<TCommand, TItem>(
+        TCommand command,
+        CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        if (!_handlers.TryGetValue(typeof(TCommand), out var registered))
+        {
+            return CommandSendAttempt<IAsyncEnumerable<TItem>>.Failed;
+        }
+
+        if (registered is not IStreamCommandHandler<TCommand, TItem> handler)
+        {
+            throw CreateHandlerContractMismatchException(
+                typeof(TCommand),
+                expected: $"IStreamCommandHandler<{typeof(TCommand).Name}, {typeof(TItem).Name}>");
+        }
+
+        return CommandSendAttempt<IAsyncEnumerable<TItem>>.FromResult(
+            handler.HandleAsync(command, cancellationToken));
+    }
+
     private static async ValueTask<bool> InvokeVoidAsync<TCommand>(
         ICommandHandler<TCommand> handler,
         TCommand command,
@@ -150,5 +195,5 @@ private static Dictionary<Type, object> Snapshot(IReadOnlyDictionary<Type, objec
     private static InvalidOperationException CreateHandlerContractMismatchException(Type commandType, string expected) =>
         new(
             $"Command type '{commandType}' is registered, but not as {expected}. " +
-            "Register a matching handler contract or call the matching SendAsync overload.");
+            "Register a matching handler contract or call the matching SendAsync / SendStreamAsync overload.");
 }

DesignPatterns/Behavioral/CommandRouterBuilder.cs

@@ -11,9 +11,11 @@ namespace DesignPatterns.Behavioral;
 /// </summary>
 /// <remarks>
 /// The builder is not thread-safe. The router returned by <see cref="Build"/> is safe for
-/// concurrent <see cref="ICommandRouter.SendAsync{TCommand}"/> / <c>TrySendAsync</c> calls.
-/// Pipeline behaviors are frozen into the handler map at <see cref="Build"/> time.
-/// Lower behavior <c>order</c> values run first (outermost inbound).
+/// concurrent <see cref="ICommandRouter.SendAsync{TCommand}"/> / <c>TrySendAsync</c> /
+/// <c>SendStreamAsync</c> / <c>TrySendStreamAsync</c> calls.
+/// A command CLR type may register either a void, result, or stream handler — not more than one.
+/// Pipeline behaviors apply to void/result handlers only and are frozen into the handler map at
+/// <see cref="Build"/> time. Lower behavior <c>order</c> values run first (outermost inbound).
 /// </remarks>
 public sealed class CommandRouterBuilder
 {
@@ -60,6 +62,26 @@ public CommandRouterBuilder Register<TCommand, TResult>(ICommandHandler<TCommand
         return this;
     }
 
+    /// <summary>
+    /// Registers a stream handler for <typeparamref name="TCommand"/>.
+    /// </summary>
+    /// <typeparam name="TCommand">The command CLR type used as the routing key.</typeparam>
+    /// <typeparam name="TItem">The item type yielded by the stream.</typeparam>
+    /// <param name="handler">The stream handler instance.</param>
+    /// <returns>This builder for chaining.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="handler"/> is <see langword="null"/>.</exception>
+    /// <exception cref="ArgumentException">A handler is already registered for <typeparamref name="TCommand"/>.</exception>
+    public CommandRouterBuilder Register<TCommand, TItem>(IStreamCommandHandler<TCommand, TItem> handler)
+    {
+        if (handler is null)
+        {
+            throw new ArgumentNullException(nameof(handler));
+        }
+
+        AddHandler(typeof(TCommand), handler);
+        return this;
+    }
+
     /// <summary>
     /// Registers a void-style pipeline behavior for <typeparamref name="TCommand"/>.
     /// Lower <paramref name="order"/> values run first (outermost inbound).

DesignPatterns/Behavioral/ICommandRouter.cs

@@ -1,3 +1,4 @@
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -7,8 +8,8 @@ namespace DesignPatterns.Behavioral;
 /// Dispatches commands to a single registered handler per command CLR type (1:1).
 /// </summary>
 /// <remarks>
-/// Missing handlers fail explicitly: throwing <c>Send*</c> APIs raise
-/// <see cref="CommandHandlerNotFoundException"/>; <c>TrySend*</c> APIs return
+/// Missing handlers fail explicitly: throwing <c>Send*</c> / <c>SendStream*</c> APIs raise
+/// <see cref="CommandHandlerNotFoundException"/>; <c>TrySend*</c> / <c>TrySendStream*</c> APIs return
 /// failure without throwing for the missing-handler case.
 /// </remarks>
 public interface ICommandRouter
@@ -69,4 +70,42 @@ ValueTask<TResult> SendAsync<TCommand, TResult>(
     ValueTask<CommandSendAttempt<TResult>> TrySendAsync<TCommand, TResult>(
         TCommand command,
         CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Sends a command to its registered stream handler and returns progressive results.
+    /// </summary>
+    /// <typeparam name="TCommand">The command CLR type used as the routing key.</typeparam>
+    /// <typeparam name="TItem">The item type yielded by the stream handler.</typeparam>
+    /// <param name="command">The command instance.</param>
+    /// <param name="cancellationToken">A cancellation token.</param>
+    /// <returns>An asynchronous stream of <typeparamref name="TItem"/> values.</returns>
+    /// <exception cref="CommandHandlerNotFoundException">
+    /// Thrown when no handler is registered for <typeparamref name="TCommand"/>.
+    /// </exception>
+    /// <exception cref="System.InvalidOperationException">
+    /// Thrown when a handler is registered for <typeparamref name="TCommand"/> but does not match the
+    /// <c>IStreamCommandHandler&lt;TCommand, TItem&gt;</c> contract.
+    /// </exception>
+    IAsyncEnumerable<TItem> SendStreamAsync<TCommand, TItem>(
+        TCommand command,
+        CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Tries to send a command to a stream handler. Returns
+    /// <see cref="CommandSendAttempt{TResult}.Failed"/> when no handler is registered.
+    /// </summary>
+    /// <typeparam name="TCommand">The command CLR type used as the routing key.</typeparam>
+    /// <typeparam name="TItem">The item type yielded by the stream handler.</typeparam>
+    /// <param name="command">The command instance.</param>
+    /// <param name="cancellationToken">A cancellation token.</param>
+    /// <returns>
+    /// A send attempt whose <see cref="CommandSendAttempt{TResult}.Result"/> is the stream when successful.
+    /// </returns>
+    /// <exception cref="System.InvalidOperationException">
+    /// Thrown when a handler is registered for <typeparamref name="TCommand"/> but does not match the
+    /// <c>IStreamCommandHandler&lt;TCommand, TItem&gt;</c> contract.
+    /// </exception>
+    CommandSendAttempt<IAsyncEnumerable<TItem>> TrySendStreamAsync<TCommand, TItem>(
+        TCommand command,
+        CancellationToken cancellationToken = default);
 }

DesignPatterns/Behavioral/IStreamCommandHandler.cs

@@ -0,0 +1,21 @@
+using System.Collections.Generic;
+using System.Threading;
+
+namespace DesignPatterns.Behavioral;
+
+/// <summary>
+/// Handles commands of type <typeparamref name="TCommand"/> by producing a progressive stream of
+/// <typeparamref name="TItem"/> values.
+/// </summary>
+/// <typeparam name="TCommand">The command type to handle.</typeparam>
+/// <typeparam name="TItem">The item type yielded by the stream.</typeparam>
+public interface IStreamCommandHandler<in TCommand, out TItem>
+{
+    /// <summary>
+    /// Handles the specified command and yields progressive results.
+    /// </summary>
+    /// <param name="command">The command instance.</param>
+    /// <param name="cancellationToken">A cancellation token.</param>
+    /// <returns>An asynchronous stream of <typeparamref name="TItem"/> values.</returns>
+    IAsyncEnumerable<TItem> HandleAsync(TCommand command, CancellationToken cancellationToken = default);
+}

DesignPatterns/DesignPatterns.csproj

@@ -11,6 +11,7 @@
   </PropertyGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" />
     <PackageReference Include="System.Threading.Tasks.Extensions" />
     <PackageReference Include="Polyfill" PrivateAssets="all" />
   </ItemGroup>

Directory.Packages.props

@@ -14,6 +14,7 @@
     <PackageVersion Include="Polyfill" Version="10.10.0" />
     <PackageVersion Include="Skymly.DesignPatterns" Version="$(PackageVersion)" />
     <PackageVersion Include="System.Security.Cryptography.Xml" Version="10.0.10" />
+    <PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="8.0.0" />
     <PackageVersion Include="System.Threading.Tasks.Extensions" Version="4.5.4" />
     <PackageVersion Include="Verify.Xunit" Version="28.3.2" />
     <PackageVersion Include="xunit" Version="2.9.2" />