IStreamRequest / CreateStreamIMediator combines ISender + IPublisherUnit support for fire-and-forget / void commands IRequest or IRequestpublic sealed class TimingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
- where TRequest : notnull
+where TRequest : notnull
{
public async Task<TResponse> Handle(
TRequest request,
@@ -435,6 +442,91 @@ Pre & Post Processors
+
+
+ Streaming Requests
+
+ Stream large result sets with IStreamRequest<T>
+ and IAsyncEnumerable<T>.
+ Handlers yield items incrementally; consumers use
+ CreateStream on
+ ISender / IMediator.
+
+
+ Request & handler
+
+
+public sealed record ExportPetsStream : IStreamRequest<Pet>;
+
+public sealed class ExportPetsStreamHandler : IStreamRequestHandler<ExportPetsStream, Pet>
+{
+ public async IAsyncEnumerable<Pet> Handle(
+ ExportPetsStream request,
+ [EnumeratorCancellation] CancellationToken cancellationToken)
+ {
+ await foreach (var pet in _db.Pets.AsAsyncEnumerable().WithCancellation(cancellationToken))
+ yield return pet;
+ }
+}
+
+
+
+ Consume the stream
+
+
+// Via ISender or IMediator
+await foreach (var pet in sender.CreateStream(new ExportPetsStream(), cancellationToken))
+{
+ // process each item
+}
+
+
+
+ ASP.NET Core action
+
+ Return IAsyncEnumerable<T> from the action
+ (not Task<IActionResult> with yield return).
+
+
+
+[HttpGet("stream")]
+public IAsyncEnumerable<Pet> GetAllPetsStream(CancellationToken cancellationToken)
+ => mediator.CreateStream(new ExportPetsStream(), cancellationToken);
+
+
+
+ Stream pipeline behaviors
+
+ Register open-generic stream behaviors with
+ options.AddStreamBehavior(...).
+
+
+
+public sealed class StreamLoggingBehavior<TRequest, TResponse>
+ : IStreamPipelineBehavior<TRequest, TResponse>
+ where TRequest : IStreamRequest<TResponse>
+{
+ public async IAsyncEnumerable<TResponse> Handle(
+ TRequest request,
+ StreamHandlerDelegate<TResponse> next,
+ [EnumeratorCancellation] CancellationToken cancellationToken)
+ {
+ await foreach (var item in next().WithCancellation(cancellationToken))
+ yield return item;
+ }
+}
+
+// Registration
+builder.Services.AddInterlink(options =>
+{
+ options.AddStreamBehavior(typeof(StreamLoggingBehavior<,>));
+}, typeof(ExportPetsStreamHandler).Assembly);
+
+
+
+
+
+
Built-in Logging Behavior
@@ -461,7 +553,7 @@ FluentValidation Integration
public sealed class CreateUserValidator : AbstractValidator<CreateUser.Command>
{
- public CreateUserValidator()
+ public CreateUserValidator()
{
RuleFor(x => x.Email).NotEmpty().EmailAddress();
RuleFor(x => x.Name).NotEmpty().MaximumLength(100);
@@ -535,7 +627,7 @@ Core Contracts
public interface IRequest<out TResponse> { }
public interface IRequestHandler<in TRequest, TResponse>
- where TRequest : IRequest<TResponse>
+ where TRequest : IRequest<TResponse>
{
Task<TResponse> Handle(TRequest request, CancellationToken cancellationToken);
}
@@ -543,7 +635,7 @@ Core Contracts
public interface INotification { }
public interface INotificationHandler<in TNotification>
- where TNotification : INotification
+ where TNotification : INotification
{
Task Handle(TNotification notification, CancellationToken cancellationToken);
}