Streaming handlers are used for scenarios where responses are produced incrementally using IAsyncEnumerable<TResponse>.
1. Create a Streaming Handler
Return IAsyncEnumerable<TResponse> from the HandleAsync method and use the [EnumeratorCancellation] attribute on the token.
[Handler]
public static partial class StreamItems
{
public record Query(int Count);
private static async IAsyncEnumerable<int> HandleAsync(
Query query,
[EnumeratorCancellation] CancellationToken token)
{
for (var i = 0; i < query.Count; i++)
{
await Task.Yield();
yield return i;
}
}
}
2. Create a Streaming Behavior
Extend StreamingBehavior<TRequest, TResponse> instead of the standard Behavior class.
public class LoggingBehavior<TRequest, TResponse>(ILogger<LoggingBehavior<TRequest, TResponse>> logger)
: StreamingBehavior<TRequest, TResponse>
{
public override async IAsyncEnumerable<TResponse> HandleAsync(
TRequest request,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
logger.LogInformation("LoggingBehavior.Enter");
await foreach (var item in Next(request, cancellationToken))
yield return item;
logger.LogInformation("LoggingBehavior.Exit");
}
}
3. Consuming Streaming Handlers
The generated handler implements IStreamingHandler<TRequest, TResponse>.
public class Consumer(IStreamingHandler<StreamItems.Query, int> handler)
{
public async Task ConsumeAsync(CancellationToken token)
{
await foreach (var item in handler.HandleAsync(new(5), token))
Console.WriteLine(item);
}
}
Note: Streaming behaviors are only applied to streaming handlers, and non-streaming behaviors are only applied to non-streaming handlers. They can coexist in the same configuration without interference.
