From af9f662c6e0265935ff2de6bec74a2dd2084e8f1 Mon Sep 17 00:00:00 2001 From: manusoft Date: Fri, 25 Sep 2026 15:44:02 +0400 Subject: [PATCH] Streaming, notification pipelines, and timeouts Introduce streaming requests and pipeline behaviors, notification pipeline behaviors with configurable publish strategies, and request timeout support. Add IStreamRequest, IStreamRequestHandler (IAsyncEnumerable), ISender.CreateStream and IMediator.CreateStream; implement stream pipelines with IStreamPipelineBehavior and StreamHandlerDelegate. Add INotificationPipelineBehavior, NotificationHandlerDelegate, and PublishStrategy (Sequential/Parallel); publisher now resolves/ orders behaviors and builds a behavior->handlers pipeline honoring PipelineOrderAttribute and InterlinkOptions.PublishStrategy. Add TimeoutBehavior and InterlinkOptions.DefaultRequestTimeout to cancel requests on timeout; register TimeoutBehavior automatically when configured. Extend InterlinkOptions with open behavior collections and Add...Behavior APIs and expose ServiceFactory; ServiceCollectionExtensions now registers options, open-generic behaviors, timeout behavior, and scans for stream/notification handlers and behaviors. Refactor Publisher and Sender to use InterlinkOptions and the service factory, build dynamic pipelines for notifications and streams, and use dynamic invocation for handlers. Tweak TypeScanner to skip open generic type definitions. Bump package to 1.6.0, add Microsoft.Bcl.AsyncInterfaces for netstandard2.0, and update README with examples and changelog. --- README.md | 139 +++++++++++++- .../Controllers/PetController.cs | 13 +- .../Features/ExportPetsStream.cs | 20 ++ .../Interlink.AspNetCore.csproj | 2 +- .../Interlink.Extensions.Logging.csproj | 2 +- .../Interlink.Extensions.Validation.csproj | 2 +- .../INotificationPipelineBehavior.cs | 28 +++ src/Interlink/Abstractions/ISender.cs | 12 +- .../Abstractions/IStreamPipelineBehavior.cs | 32 ++++ .../Abstractions/IStreamRequestHandler.cs | 21 +++ src/Interlink/Behaviors/TimeoutBehavior.cs | 48 +++++ .../Configurations/InterlinkOptions.cs | 102 +++++++--- .../Configurations/PublishStrategy.cs | 19 ++ src/Interlink/Contracts/IStreamRequest.cs | 9 + src/Interlink/Core/Mediator.cs | 7 +- src/Interlink/Core/Publisher.cs | 83 ++++++-- src/Interlink/Core/Sender.cs | 37 ++++ src/Interlink/Interlink.csproj | 9 +- src/Interlink/Internals/TypeScanner.cs | 4 +- src/Interlink/README.md | 120 +++++++++++- src/Interlink/ServiceCollectionExtensions.cs | 49 +++-- test/Interlink.Test/BasicTests.cs | 4 +- test/Interlink.Test/NotificationTests.cs | 3 +- test/Interlink.Test/PipelineBehaviorTests.cs | 6 +- test/Interlink.Test/StreamingTests.cs | 177 ++++++++++++++++++ 25 files changed, 875 insertions(+), 73 deletions(-) create mode 100644 sample/Interlink.Sample/Features/ExportPetsStream.cs create mode 100644 src/Interlink/Abstractions/INotificationPipelineBehavior.cs create mode 100644 src/Interlink/Abstractions/IStreamPipelineBehavior.cs create mode 100644 src/Interlink/Abstractions/IStreamRequestHandler.cs create mode 100644 src/Interlink/Behaviors/TimeoutBehavior.cs create mode 100644 src/Interlink/Configurations/PublishStrategy.cs create mode 100644 src/Interlink/Contracts/IStreamRequest.cs create mode 100644 test/Interlink.Test/StreamingTests.cs diff --git a/README.md b/README.md index 257ee1d..8c36edd 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,10 @@ * 🧩 Simple mediator pattern for request/response * 🔁 Publish/Subscribe notification system +* ⚡ Notification publish strategies (sequential / parallel) +* 🔔 Notification pipeline behaviors +* ⏱️ Optional request timeout behavior +* 📡 Streaming requests (`IStreamRequest` / `CreateStream`) * 🎯 Unified `IMediator` (combines `ISender` + `IPublisher`) * ⚪ `Unit` support for fire-and-forget / void commands (`IRequest` / `IRequest`) * 🔧 Pipeline behaviors (logging, validation, etc.) @@ -45,6 +49,7 @@ * ✅ Compatible with .NET Standard 2.0+ to .NET 10 * 📦 Optional packages: Logging, FluentValidation, ASP.NET Core, Analyzer + --- ## 📦 Installation @@ -346,6 +351,94 @@ They are discovered automatically by `AddInterlink()`. --- +--- + +## 📡 Streaming requests (v1.6) + +```csharp +public sealed record ExportPetsStream : IStreamRequest; + +public sealed class ExportPetsStreamHandler : IStreamRequestHandler +{ + public async IAsyncEnumerable Handle( + ExportPetsStream request, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var pet in _repository.StreamAllAsync(cancellationToken)) + yield return pet; + } +} + +// Consumer +await foreach (var pet in sender.CreateStream(new ExportPetsStream(), ct)) +{ + // process each item +} + +// or via IMediator +await foreach (var pet in mediator.CreateStream(new ExportPetsStream(), ct)) +{ + // ... +} +``` + +Register stream pipeline behaviors: + +```csharp +builder.Services.AddInterlink(options => +{ + options.AddStreamBehavior(typeof(MyStreamBehavior<,>)); +}, typeof(MyHandler).Assembly); +``` + +## ⚡ Notification publish strategies (v1.6) + +```csharp +builder.Services.AddInterlink(options => +{ + options.PublishStrategy = PublishStrategy.Parallel; // or Sequential (default) +}, typeof(MyHandler).Assembly); +``` + +- **Sequential** — handlers run one after another (ordering preserved). +- **Parallel** — handlers run concurrently via `Task.WhenAll`. + +## 🔔 Notification pipeline behaviors (v1.6) + +```csharp +public sealed class NotificationLoggingBehavior : INotificationPipelineBehavior + where TNotification : INotification +{ + public async Task Handle( + TNotification notification, + NotificationHandlerDelegate next, + CancellationToken cancellationToken) + { + // before + await next(cancellationToken); + // after + } +} + +builder.Services.AddInterlink(options => +{ + options.AddNotificationBehavior(typeof(NotificationLoggingBehavior<>)); +}); +``` + +## ⏱️ Request timeout (v1.6) + +```csharp +builder.Services.AddInterlink(options => +{ + options.DefaultRequestTimeout = TimeSpan.FromSeconds(30); +}, typeof(MyHandler).Assembly); +``` + +Registers `TimeoutBehavior<,>` automatically. Exceeding the timeout throws `TimeoutException`. + +--- + ## 📋 Built-in Logging Behavior ```bash @@ -420,7 +513,10 @@ The filter maps: dotnet add package Interlink.Analyzers ``` -Produces diagnostic **ILINK001** (warning) when a type implements `IRequest` but no corresponding `IRequestHandler` is found in the compilation. +| Id | Meaning | +|----|---------| +| **ILINK001** | Request has no handler | +| **ILINK002** | Multiple handlers for the same request (v1.6) | --- @@ -515,6 +611,25 @@ public interface IRequestPostProcessor where TRequest } ``` +### Streaming requests +```csharp +public interface IStreamRequest { } + +public interface IStreamRequestHandler + where TRequest : IStreamRequest +{ + IAsyncEnumerable Handle(TRequest request, CancellationToken cancellationToken); +} +``` + +Register stream pipeline behaviors: +```csharp +builder.Services.AddInterlink(opt => +{ + opt.AddStreamBehavior(typeof(StreamLoggingBehavior<,>)); +}, typeof(MyHandler).Assembly); +``` + ### Exception ```csharp @@ -535,11 +650,13 @@ public class HandlerNotFoundException : InvalidOperationException | 1.4 | ✅ Released | .NET Standard 2.0+ | | 1.5 | ✅ Released | Logging, Validation, ASP.NET Core, Analyzer, exceptions, ordering fixes | | 1.5.1 | ✅ Released | `Unit` support for fire-and-forget / void commands (`IRequest` / `IRequest`) | -| **1.5.2** | ✅ Current | Added unified `IMediator` interface (composes `ISender` + `IPublisher`) | +| 1.5.2 | ✅ Released | Added unified `IMediator` interface (composes `ISender` + `IPublisher`) | +| **1.6** | 🔄 Current | Streaming requests, notification publish strategies, notification pipeline behaviors, request timeout behavior | + ### Future ideas -* Request cancellation / timeout behaviors +* Request cancellation * Metrics & tracing support * Dynamic / externalized pipeline configuration @@ -548,3 +665,19 @@ public class HandlerNotFoundException : InvalidOperationException ## 📜 License MIT License © ManuHub + +--- + +## 👨‍💻 Author + +**Manoj Babu**, +ManuHub + +## 👥 Contributors + +Thanks to all contributors ❤️ + + + + + diff --git a/sample/Interlink.Sample/Controllers/PetController.cs b/sample/Interlink.Sample/Controllers/PetController.cs index f6d9231..09cdfe3 100644 --- a/sample/Interlink.Sample/Controllers/PetController.cs +++ b/sample/Interlink.Sample/Controllers/PetController.cs @@ -1,5 +1,7 @@ -using Interlink.Sample.Features; +using Interlink.Sample.Entities; +using Interlink.Sample.Features; using Microsoft.AspNetCore.Mvc; +using System.Runtime.CompilerServices; namespace Interlink.Sample.Controllers; @@ -14,6 +16,15 @@ public async Task GetAllPets(CancellationToken cancellationToken) return Ok(pets); } + [HttpGet("stream")] + public async IAsyncEnumerable GetAllPetsStream([EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var pet in mediator.CreateStream(new ExportPetsStream.Stream(), cancellationToken)) + { + yield return pet; + } + } + [HttpPost] public async Task CreatePet([FromBody] CreatePet.Command command, CancellationToken cancellationToken) { diff --git a/sample/Interlink.Sample/Features/ExportPetsStream.cs b/sample/Interlink.Sample/Features/ExportPetsStream.cs new file mode 100644 index 0000000..e8a6776 --- /dev/null +++ b/sample/Interlink.Sample/Features/ExportPetsStream.cs @@ -0,0 +1,20 @@ +using Interlink.Contracts; +using Interlink.Sample.Data; +using Interlink.Sample.Entities; +using System.Runtime.CompilerServices; + +namespace Interlink.Sample.Features; + +public class ExportPetsStream +{ + public record Stream : IStreamRequest; + + public class Handler(AppDbContext context) : IStreamRequestHandler + { + public async IAsyncEnumerable Handle(Stream request, [EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var pet in context.Pets.AsAsyncEnumerable().WithCancellation(cancellationToken)) + yield return pet; + } + } +} diff --git a/src/Interlink.AspNetCore/Interlink.AspNetCore.csproj b/src/Interlink.AspNetCore/Interlink.AspNetCore.csproj index 204eec1..efdd603 100644 --- a/src/Interlink.AspNetCore/Interlink.AspNetCore.csproj +++ b/src/Interlink.AspNetCore/Interlink.AspNetCore.csproj @@ -8,7 +8,7 @@ Interlink.AspNetCore Interlink.AspNetCore Interlink.AspNetCore - 1.5.2 + 1.6.0 ASP.NET Core integration for the Interlink mediator library (filters, exception handling, helpers). Copyright © 2025-2026 ManuHub Manojbabu diff --git a/src/Interlink.Extensions.Logging/Interlink.Extensions.Logging.csproj b/src/Interlink.Extensions.Logging/Interlink.Extensions.Logging.csproj index 8894ee3..8a573bb 100644 --- a/src/Interlink.Extensions.Logging/Interlink.Extensions.Logging.csproj +++ b/src/Interlink.Extensions.Logging/Interlink.Extensions.Logging.csproj @@ -8,7 +8,7 @@ Interlink.Extensions.Logging Interlink.Extensions.Logging Interlink.Extensions.Logging - 1.5.2 + 1.6.0 Built-in logging pipeline behavior for the Interlink mediator library. Copyright © 2025-2026 ManuHub Manojbabu diff --git a/src/Interlink.Extensions.Validation/Interlink.Extensions.Validation.csproj b/src/Interlink.Extensions.Validation/Interlink.Extensions.Validation.csproj index 5c97983..da63fc4 100644 --- a/src/Interlink.Extensions.Validation/Interlink.Extensions.Validation.csproj +++ b/src/Interlink.Extensions.Validation/Interlink.Extensions.Validation.csproj @@ -8,7 +8,7 @@ Interlink.Extensions.Validation Interlink.Extensions.Validation Interlink.Extensions.Validation - 1.5.2 + 1.6.0 FluentValidation integration for the Interlink mediator library. Copyright © 2025-2026 ManuHub Manojbabu diff --git a/src/Interlink/Abstractions/INotificationPipelineBehavior.cs b/src/Interlink/Abstractions/INotificationPipelineBehavior.cs new file mode 100644 index 0000000..0b5e24f --- /dev/null +++ b/src/Interlink/Abstractions/INotificationPipelineBehavior.cs @@ -0,0 +1,28 @@ +using Interlink.Contracts; + +namespace Interlink; + +/// +/// Represents a delegate that continues the notification pipeline. +/// +/// A token to monitor for cancellation requests. +/// A task that represents the asynchronous operation. +public delegate Task NotificationHandlerDelegate(CancellationToken cancellationToken = default); + +/// +/// Defines a behavior in the notification pipeline that can inspect or wrap +/// the publishing of a notification before and/or after the handlers run. +/// +/// The type of the notification. +public interface INotificationPipelineBehavior + where TNotification : INotification +{ + /// + /// Handles the notification and optionally calls the next delegate in the pipeline. + /// + /// The notification being published. + /// The next delegate in the pipeline (remaining behaviors + handlers). + /// A token to monitor for cancellation requests. + /// A task that represents the asynchronous operation. + Task Handle(TNotification notification, NotificationHandlerDelegate next, CancellationToken cancellationToken); +} diff --git a/src/Interlink/Abstractions/ISender.cs b/src/Interlink/Abstractions/ISender.cs index bd346eb..9db3d02 100644 --- a/src/Interlink/Abstractions/ISender.cs +++ b/src/Interlink/Abstractions/ISender.cs @@ -16,4 +16,14 @@ public interface ISender /// Sends a void/command request that does not produce a response value. /// Task Send(IRequest request, CancellationToken cancellationToken = default); -} \ No newline at end of file + + /// + /// Sends a stream request and returns an asynchronous stream of responses. + /// + /// The type of each item in the response stream. + /// The stream request to send. + /// A token to monitor for cancellation requests. + /// An asynchronous stream of response items. + /// Thrown when no stream handler is registered for the request type. + IAsyncEnumerable CreateStream(IStreamRequest request, CancellationToken cancellationToken = default); +} diff --git a/src/Interlink/Abstractions/IStreamPipelineBehavior.cs b/src/Interlink/Abstractions/IStreamPipelineBehavior.cs new file mode 100644 index 0000000..5c7a9c8 --- /dev/null +++ b/src/Interlink/Abstractions/IStreamPipelineBehavior.cs @@ -0,0 +1,32 @@ +using Interlink.Contracts; + +namespace Interlink; + +/// +/// Represents a delegate that continues the stream request pipeline. +/// +/// The type of each streamed response item. +/// An asynchronous stream of response items. +public delegate IAsyncEnumerable StreamHandlerDelegate(); + +/// +/// Defines a behavior in the stream request pipeline that can inspect or wrap +/// streaming before and/or after the next component. +/// +/// The type of the stream request. +/// The type of each streamed response item. +public interface IStreamPipelineBehavior + where TRequest : IStreamRequest +{ + /// + /// Handles the stream request and optionally calls the next delegate in the pipeline. + /// + /// The request to process. + /// The next delegate in the pipeline. + /// A token to monitor for cancellation requests. + /// An asynchronous stream of response items. + IAsyncEnumerable Handle( + TRequest request, + StreamHandlerDelegate next, + CancellationToken cancellationToken); +} diff --git a/src/Interlink/Abstractions/IStreamRequestHandler.cs b/src/Interlink/Abstractions/IStreamRequestHandler.cs new file mode 100644 index 0000000..e06fb43 --- /dev/null +++ b/src/Interlink/Abstractions/IStreamRequestHandler.cs @@ -0,0 +1,21 @@ +using Interlink.Contracts; + +namespace Interlink; + +/// +/// Defines a handler for a streaming request of type +/// that yields a sequence of values. +/// +/// The type of the stream request. +/// The type of each streamed response item. +public interface IStreamRequestHandler + where TRequest : IStreamRequest +{ + /// + /// Handles the stream request and yields response items asynchronously. + /// + /// The request instance to handle. + /// A token to monitor for cancellation requests. + /// An asynchronous stream of response items. + IAsyncEnumerable Handle(TRequest request, CancellationToken cancellationToken); +} diff --git a/src/Interlink/Behaviors/TimeoutBehavior.cs b/src/Interlink/Behaviors/TimeoutBehavior.cs new file mode 100644 index 0000000..4a36f38 --- /dev/null +++ b/src/Interlink/Behaviors/TimeoutBehavior.cs @@ -0,0 +1,48 @@ +namespace Interlink; + +/// +/// Pipeline behavior that cancels a request if it exceeds the configured timeout. +/// Registered automatically when is set. +/// +/// The type of the request. +/// The type of the response. +public sealed class TimeoutBehavior : IPipelineBehavior + where TRequest : notnull +{ + private readonly TimeSpan _timeout; + + /// + /// Initializes a new instance of the class. + /// + /// Interlink options that supply the default timeout. + public TimeoutBehavior(InterlinkOptions options) + { + if (options is null) + throw new ArgumentNullException(nameof(options)); + + _timeout = options.DefaultRequestTimeout ?? Timeout.InfiniteTimeSpan; + } + + /// + public async Task Handle( + TRequest request, + RequestHandlerDelegate next, + CancellationToken cancellationToken) + { + if (_timeout <= TimeSpan.Zero || _timeout == Timeout.InfiniteTimeSpan) + return await next(cancellationToken).ConfigureAwait(false); + + using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + timeoutCts.CancelAfter(_timeout); + + try + { + return await next(timeoutCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + throw new TimeoutException( + $"Handling of '{typeof(TRequest).Name}' timed out after {_timeout.TotalMilliseconds:0} ms."); + } + } +} diff --git a/src/Interlink/Configurations/InterlinkOptions.cs b/src/Interlink/Configurations/InterlinkOptions.cs index f6d74f1..7ba391a 100644 --- a/src/Interlink/Configurations/InterlinkOptions.cs +++ b/src/Interlink/Configurations/InterlinkOptions.cs @@ -7,6 +7,10 @@ public class InterlinkOptions { internal List<(Type Type, int? Order)> OpenBehaviors { get; } = new(); + internal List<(Type Type, int? Order)> OpenNotificationBehaviors { get; } = new(); + + internal List<(Type Type, int? Order)> OpenStreamBehaviors { get; } = new(); + /// /// Gets or sets an optional custom factory used to resolve handlers and pipeline components. /// When set, this factory is preferred over the default . @@ -14,20 +18,21 @@ public class InterlinkOptions public Func? ServiceFactory { get; set; } /// - /// Adds an open-generic pipeline behavior type to the configuration. - /// - /// - /// An open generic type that implements - /// (for example typeof(LoggingBehavior<,>)). - /// - /// - /// Optional explicit order. When provided, takes precedence over any - /// on the type. Lower values run first. - /// - /// is null. - /// - /// is not an open generic type with exactly two type parameters. - /// + /// Gets or sets how notification handlers are invoked when publishing. + /// Default is . + /// + public PublishStrategy PublishStrategy { get; set; } = PublishStrategy.Sequential; + + /// + /// Gets or sets an optional default timeout applied to all requests via + /// . + /// When null, no timeout behavior is registered. + /// + public TimeSpan? DefaultRequestTimeout { get; set; } + + /// + /// Adds an open-generic request pipeline behavior type to the configuration. + /// public void AddBehavior(Type openGenericBehaviorType, int? order = null) { if (openGenericBehaviorType is null) @@ -46,18 +51,69 @@ public void AddBehavior(Type openGenericBehaviorType, int? order = null) } /// - /// Adds an open-generic pipeline behavior type to the configuration using a generic type parameter. + /// Adds an open-generic request pipeline behavior using a generic type parameter. /// - /// - /// The open generic behavior type that implements . - /// - /// - /// Optional explicit order. When provided, takes precedence over any - /// on the type. Lower values run first. - /// public void AddBehavior(int? order = null) where TBehavior : class { AddBehavior(typeof(TBehavior), order); } -} \ No newline at end of file + + /// + /// Adds an open-generic notification pipeline behavior type to the configuration. + /// + public void AddNotificationBehavior(Type openGenericBehaviorType, int? order = null) + { + if (openGenericBehaviorType is null) + throw new ArgumentNullException(nameof(openGenericBehaviorType)); + + if (!openGenericBehaviorType.IsGenericTypeDefinition || + openGenericBehaviorType.GetGenericArguments().Length != 1) + { + throw new ArgumentException( + "Notification behavior must be an open generic type definition with exactly one generic parameter " + + "(for example typeof(MyNotificationBehavior<>)).", + nameof(openGenericBehaviorType)); + } + + OpenNotificationBehaviors.Add((openGenericBehaviorType, order)); + } + + /// + /// Adds an open-generic notification pipeline behavior using a generic type parameter. + /// + public void AddNotificationBehavior(int? order = null) + where TBehavior : class + { + AddNotificationBehavior(typeof(TBehavior), order); + } + + /// + /// Adds an open-generic stream pipeline behavior type to the configuration. + /// + public void AddStreamBehavior(Type openGenericBehaviorType, int? order = null) + { + if (openGenericBehaviorType is null) + throw new ArgumentNullException(nameof(openGenericBehaviorType)); + + if (!openGenericBehaviorType.IsGenericTypeDefinition || + openGenericBehaviorType.GetGenericArguments().Length != 2) + { + throw new ArgumentException( + "Stream behavior must be an open generic type definition with exactly two generic parameters " + + "(for example typeof(MyStreamBehavior<,>)).", + nameof(openGenericBehaviorType)); + } + + OpenStreamBehaviors.Add((openGenericBehaviorType, order)); + } + + /// + /// Adds an open-generic stream pipeline behavior using a generic type parameter. + /// + public void AddStreamBehavior(int? order = null) + where TBehavior : class + { + AddStreamBehavior(typeof(TBehavior), order); + } +} diff --git a/src/Interlink/Configurations/PublishStrategy.cs b/src/Interlink/Configurations/PublishStrategy.cs new file mode 100644 index 0000000..7d6d8ee --- /dev/null +++ b/src/Interlink/Configurations/PublishStrategy.cs @@ -0,0 +1,19 @@ +namespace Interlink; + +/// +/// Specifies how notification handlers are invoked when a notification is published. +/// +public enum PublishStrategy +{ + /// + /// Handlers are invoked one after another in registration order. + /// This is the default and preserves ordering guarantees. + /// + Sequential = 0, + + /// + /// Handlers are invoked concurrently. Completes when all handlers finish. + /// Exceptions from any handler are aggregated into an . + /// + Parallel = 1 +} diff --git a/src/Interlink/Contracts/IStreamRequest.cs b/src/Interlink/Contracts/IStreamRequest.cs new file mode 100644 index 0000000..c1a28f8 --- /dev/null +++ b/src/Interlink/Contracts/IStreamRequest.cs @@ -0,0 +1,9 @@ +namespace Interlink.Contracts; + +/// +/// Marker interface for a request that produces a stream of values. +/// +/// The type of each item in the response stream. +public interface IStreamRequest +{ +} diff --git a/src/Interlink/Core/Mediator.cs b/src/Interlink/Core/Mediator.cs index 72992dd..86ebb97 100644 --- a/src/Interlink/Core/Mediator.cs +++ b/src/Interlink/Core/Mediator.cs @@ -19,7 +19,12 @@ public Task Send(IRequest request, Cancellation public Task Send(IRequest request, CancellationToken cancellationToken = default) => _sender.Send(request, cancellationToken); + public IAsyncEnumerable CreateStream( + IStreamRequest request, + CancellationToken cancellationToken = default) + => _sender.CreateStream(request, cancellationToken); + public Task Publish(TNotification notification, CancellationToken cancellationToken = default) where TNotification : INotification => _publisher.Publish(notification, cancellationToken); -} \ No newline at end of file +} diff --git a/src/Interlink/Core/Publisher.cs b/src/Interlink/Core/Publisher.cs index 5c0a2d3..aa3f779 100644 --- a/src/Interlink/Core/Publisher.cs +++ b/src/Interlink/Core/Publisher.cs @@ -1,28 +1,33 @@ using Interlink.Contracts; +using System.Reflection; namespace Interlink; /// -/// Default implementation of that resolves all -/// notification handlers and invokes them sequentially. +/// Default implementation of that resolves notification handlers, +/// runs notification pipeline behaviors, and invokes handlers using the configured +/// . /// internal sealed class Publisher : IPublisher { private readonly IServiceProvider _provider; private readonly Func _serviceFactory; + private readonly PublishStrategy _publishStrategy; /// /// Initializes a new instance of the class. /// - /// The service provider used for resolution when no custom factory is supplied. - /// - /// Optional custom factory. When provided it is used instead of - /// for resolving notification handlers. - /// - public Publisher(IServiceProvider provider, Func? customFactory = null) + public Publisher( + IServiceProvider provider, + InterlinkOptions options, + Func? customFactory = null) { _provider = provider ?? throw new ArgumentNullException(nameof(provider)); + if (options is null) + throw new ArgumentNullException(nameof(options)); + _serviceFactory = customFactory ?? (type => _provider.GetService(type)); + _publishStrategy = options.PublishStrategy; } /// @@ -36,12 +41,68 @@ public async Task Publish(TNotification notification, Cancellatio var handlerType = typeof(INotificationHandler<>).MakeGenericType(notificationType); var enumerableType = typeof(IEnumerable<>).MakeGenericType(handlerType); - var handlers = _serviceFactory(enumerableType) as IEnumerable ?? Array.Empty(); + var handlers = (_serviceFactory(enumerableType) as IEnumerable)?.ToList() + ?? new List(); + + // Innermost: invoke all handlers according to strategy + NotificationHandlerDelegate pipeline = token => InvokeHandlersAsync(handlers, notification, token); + + // Notification pipeline behaviors (ordered, outermost first) + var behaviorType = typeof(INotificationPipelineBehavior<>).MakeGenericType(notificationType); + var behaviors = ResolveEnumerable(behaviorType) + .Select(b => (Instance: b, Order: GetOrder(b.GetType()))) + .OrderBy(x => x.Order) + .Select(x => x.Instance) + .ToList(); + + for (var i = behaviors.Count - 1; i >= 0; i--) + { + var behavior = behaviors[i]; + var next = pipeline; + pipeline = token => ((dynamic)behavior).Handle((dynamic)notification, next, token); + } + + await pipeline(cancellationToken).ConfigureAwait(false); + } + + private async Task InvokeHandlersAsync( + List handlers, + object notification, + CancellationToken cancellationToken) + { + if (handlers.Count == 0) + return; + + if (_publishStrategy == PublishStrategy.Parallel) + { + var tasks = new Task[handlers.Count]; + for (var i = 0; i < handlers.Count; i++) + { + dynamic handler = handlers[i]; + tasks[i] = handler.Handle((dynamic)notification, cancellationToken); + } + await Task.WhenAll(tasks).ConfigureAwait(false); + return; + } + + // Sequential (default) foreach (var handler in handlers) { - // Use dynamic for consistency with Sender and to avoid reflection Invoke overhead await ((dynamic)handler).Handle((dynamic)notification, cancellationToken).ConfigureAwait(false); } } -} \ No newline at end of file + + private IEnumerable ResolveEnumerable(Type elementType) + { + var enumerableType = typeof(IEnumerable<>).MakeGenericType(elementType); + var resolved = _serviceFactory(enumerableType); + return resolved as IEnumerable ?? Array.Empty(); + } + + private static int GetOrder(Type type) + { + var attr = type.GetCustomAttribute(inherit: false); + return attr?.Order ?? int.MaxValue; + } +} diff --git a/src/Interlink/Core/Sender.cs b/src/Interlink/Core/Sender.cs index 9c21163..3faf7eb 100644 --- a/src/Interlink/Core/Sender.cs +++ b/src/Interlink/Core/Sender.cs @@ -93,6 +93,43 @@ public Task Send(IRequest request, CancellationToken cancellationToken = default return Send(request, cancellationToken); } + + /// + public IAsyncEnumerable CreateStream( + IStreamRequest request, + CancellationToken cancellationToken = default) + { + if (request is null) + throw new ArgumentNullException(nameof(request)); + + var requestType = request.GetType(); + var responseType = typeof(TResponse); + + var handlerType = typeof(IStreamRequestHandler<,>).MakeGenericType(requestType, responseType); + dynamic handler = ResolveRequired(handlerType, requestType); + + // Handler + StreamHandlerDelegate pipeline = () => + ((dynamic)handler).Handle((dynamic)request, cancellationToken); + + var behaviorType = typeof(IStreamPipelineBehavior<,>).MakeGenericType(requestType, responseType); + var behaviors = ResolveEnumerable(behaviorType) + .Select(b => (Instance: b, Order: GetOrder(b.GetType()))) + .OrderBy(x => x.Order) + .Select(x => x.Instance) + .ToList(); + + // Behaviors + for (var i = behaviors.Count - 1; i >= 0; i--) + { + var behavior = behaviors[i]; + var next = pipeline; + pipeline = () => ((dynamic)behavior).Handle((dynamic)request, next, cancellationToken); + } + + return pipeline(); + } + private object ResolveRequired(Type serviceType, Type requestType) { var instance = _serviceFactory(serviceType); diff --git a/src/Interlink/Interlink.csproj b/src/Interlink/Interlink.csproj index b5ab7f4..f294659 100644 --- a/src/Interlink/Interlink.csproj +++ b/src/Interlink/Interlink.csproj @@ -10,7 +10,7 @@ Interlink Interlink Interlink - 1.5.2 + 1.6.0 A lightweight and minimal mediator library for .NET. Interlink helps you decouple your application using request/response and notification patterns with simple, clean code. Copyright © 2025-2026 ManuHub Manojbabu @@ -33,12 +33,13 @@ + - - - + + + diff --git a/src/Interlink/Internals/TypeScanner.cs b/src/Interlink/Internals/TypeScanner.cs index 0f010ec..9123049 100644 --- a/src/Interlink/Internals/TypeScanner.cs +++ b/src/Interlink/Internals/TypeScanner.cs @@ -14,6 +14,8 @@ internal static class TypeScanner /// /// Scans the given assembly for non-abstract, non-interface types that implement /// the specified open generic interface. + /// Open generic type definitions are skipped (register those via AddBehavior / + /// AddStreamBehavior / AddNotificationBehavior). /// public static IEnumerable<(Type ServiceType, Type ImplementationType)> Scan(Assembly assembly, Type openGenericType) { @@ -40,7 +42,7 @@ internal static class TypeScanner foreach (var type in types) { - if (type is null || type.IsAbstract || type.IsInterface) + if (type is null || type.IsAbstract || type.IsInterface || type.IsGenericTypeDefinition) continue; foreach (var iface in type.GetInterfaces()) diff --git a/src/Interlink/README.md b/src/Interlink/README.md index d57dc3f..bbb73f1 100644 --- a/src/Interlink/README.md +++ b/src/Interlink/README.md @@ -12,6 +12,10 @@ * 🧩 Simple mediator pattern for request/response * 🔁 Publish/Subscribe notification system +* ⚡ Notification publish strategies (sequential / parallel) +* 🔔 Notification pipeline behaviors +* ⏱️ Optional request timeout behavior +* 📡 Streaming requests (`IStreamRequest` / `CreateStream`) * 🎯 Unified `IMediator` (combines `ISender` + `IPublisher`) * ⚪ `Unit` support for fire-and-forget / void commands (`IRequest` / `IRequest`) * 🔧 Pipeline behaviors (logging, validation, etc.) @@ -326,6 +330,92 @@ They are discovered automatically by `AddInterlink()`. --- +## 📡 Streaming requests (v1.6) + +```csharp +public sealed record ExportPetsStream : IStreamRequest; + +public sealed class ExportPetsStreamHandler : IStreamRequestHandler +{ + public async IAsyncEnumerable Handle( + ExportPetsStream request, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var pet in _repository.StreamAllAsync(cancellationToken)) + yield return pet; + } +} + +// Consumer +await foreach (var pet in sender.CreateStream(new ExportPetsStream(), ct)) +{ + // process each item +} + +// or via IMediator +await foreach (var pet in mediator.CreateStream(new ExportPetsStream(), ct)) +{ + // ... +} +``` + +Register stream pipeline behaviors: + +```csharp +builder.Services.AddInterlink(options => +{ + options.AddStreamBehavior(typeof(MyStreamBehavior<,>)); +}, typeof(MyHandler).Assembly); +``` + +## ⚡ Notification publish strategies (v1.6) + +```csharp +builder.Services.AddInterlink(options => +{ + options.PublishStrategy = PublishStrategy.Parallel; // or Sequential (default) +}, typeof(MyHandler).Assembly); +``` + +- **Sequential** — handlers run one after another (ordering preserved). +- **Parallel** — handlers run concurrently via `Task.WhenAll`. + +## 🔔 Notification pipeline behaviors (v1.6) + +```csharp +public sealed class NotificationLoggingBehavior : INotificationPipelineBehavior + where TNotification : INotification +{ + public async Task Handle( + TNotification notification, + NotificationHandlerDelegate next, + CancellationToken cancellationToken) + { + // before + await next(cancellationToken); + // after + } +} + +builder.Services.AddInterlink(options => +{ + options.AddNotificationBehavior(typeof(NotificationLoggingBehavior<>)); +}); +``` + +## ⏱️ Request timeout (v1.6) + +```csharp +builder.Services.AddInterlink(options => +{ + options.DefaultRequestTimeout = TimeSpan.FromSeconds(30); +}, typeof(MyHandler).Assembly); +``` + +Registers `TimeoutBehavior<,>` automatically. Exceeding the timeout throws `TimeoutException`. + +--- + ## 📋 Built-in Logging Behavior ```bash @@ -400,7 +490,10 @@ The filter maps: dotnet add package Interlink.Analyzers ``` -Produces diagnostic **ILINK001** (warning) when a type implements `IRequest` but no corresponding `IRequestHandler` is found in the compilation. +| Id | Meaning | +|----|---------| +| **ILINK001** | Request has no handler | +| **ILINK002** | Multiple handlers for the same request (v1.6) | --- @@ -495,6 +588,25 @@ public interface IRequestPostProcessor where TRequest } ``` +### Streaming requests +```csharp +public interface IStreamRequest { } + +public interface IStreamRequestHandler + where TRequest : IStreamRequest +{ + IAsyncEnumerable Handle(TRequest request, CancellationToken cancellationToken); +} +``` + +Register stream pipeline behaviors: +```csharp +builder.Services.AddInterlink(opt => +{ + opt.AddStreamBehavior(typeof(StreamLoggingBehavior<,>)); +}, typeof(MyHandler).Assembly); +``` + ### Exception ```csharp @@ -515,11 +627,13 @@ public class HandlerNotFoundException : InvalidOperationException | 1.4 | ✅ Released | .NET Standard 2.0+ | | 1.5 | ✅ Released | Logging, Validation, ASP.NET Core, Analyzer, exceptions, ordering fixes | | 1.5.1 | ✅ Released | `Unit` support for fire-and-forget / void commands (`IRequest` / `IRequest`) | -| **1.5.2** | ✅ Current | Added unified `IMediator` interface (composes `ISender` + `IPublisher`) | +| 1.5.2 | ✅ Released | Added unified `IMediator` interface (composes `ISender` + `IPublisher`) | +| **1.6** | 🔄 Current | Streaming requests, notification publish strategies, notification pipeline behaviors, request timeout behavior | + ### Future ideas -* Request cancellation / timeout behaviors +* Request cancellation * Metrics & tracing support * Dynamic / externalized pipeline configuration diff --git a/src/Interlink/ServiceCollectionExtensions.cs b/src/Interlink/ServiceCollectionExtensions.cs index 5021f38..e38112a 100644 --- a/src/Interlink/ServiceCollectionExtensions.cs +++ b/src/Interlink/ServiceCollectionExtensions.cs @@ -14,18 +14,14 @@ public static class ServiceCollectionExtensions /// /// The service collection to add services to. /// - /// Optional configuration callback used to register open-generic pipeline behaviors - /// and to supply a custom service factory. + /// Optional configuration callback used to register open-generic pipeline behaviors, + /// publish strategy, timeouts, and to supply a custom service factory. /// /// - /// Assemblies to scan for , - /// , - /// and - /// implementations. + /// Assemblies to scan for handlers and processors. /// When omitted, the calling assembly is scanned. /// /// The same instance so that calls can be chained. - /// is null. public static IServiceCollection AddInterlink( this IServiceCollection services, Action? configure = null, @@ -37,15 +33,36 @@ public static IServiceCollection AddInterlink( var options = new InterlinkOptions(); configure?.Invoke(options); + // Make options available to TimeoutBehavior and Publisher + services.AddSingleton(options); + if (assemblies is null || assemblies.Length == 0) assemblies = new[] { Assembly.GetCallingAssembly() }; - // Register explicitly configured open-generic pipeline behaviors + // Request pipeline behaviors foreach (var (behaviorType, _) in options.OpenBehaviors) { services.AddScoped(typeof(IPipelineBehavior<,>), behaviorType); } + // Notification pipeline behaviors + foreach (var (behaviorType, _) in options.OpenNotificationBehaviors) + { + services.AddScoped(typeof(INotificationPipelineBehavior<>), behaviorType); + } + + // Stream pipeline behaviors + foreach (var (behaviorType, _) in options.OpenStreamBehaviors) + { + services.AddScoped(typeof(IStreamPipelineBehavior<,>), behaviorType); + } + + // Built-in timeout behavior when configured + if (options.DefaultRequestTimeout is { } timeout && timeout > TimeSpan.Zero) + { + services.AddScoped(typeof(IPipelineBehavior<,>), typeof(TimeoutBehavior<,>)); + } + // Scan assemblies for handlers and processors foreach (var assembly in assemblies) { @@ -53,22 +70,24 @@ public static IServiceCollection AddInterlink( RegisterClosedGenericImplementations(services, assembly, typeof(INotificationHandler<>)); RegisterClosedGenericImplementations(services, assembly, typeof(IRequestPreProcessor<>)); RegisterClosedGenericImplementations(services, assembly, typeof(IRequestPostProcessor<,>)); + RegisterClosedGenericImplementations(services, assembly, typeof(INotificationPipelineBehavior<>)); + RegisterClosedGenericImplementations(services, assembly, typeof(IStreamRequestHandler<,>)); + RegisterClosedGenericImplementations(services, assembly, typeof(IStreamPipelineBehavior<,>)); } - // Register core sender / publisher, honouring any custom factory - if (options.ServiceFactory is not null) + var factory = options.ServiceFactory; + + if (factory is not null) { - var factory = options.ServiceFactory; services.AddScoped(sp => new Sender(sp, factory)); - services.AddScoped(sp => new Publisher(sp, factory)); + services.AddScoped(sp => new Publisher(sp, options, factory)); } else { services.AddScoped(); - services.AddScoped(); + services.AddScoped(sp => new Publisher(sp, options)); } - // Mediator (composes ISender + IPublisher) services.AddScoped(); return services; @@ -84,4 +103,4 @@ private static void RegisterClosedGenericImplementations( services.AddScoped(serviceType, implementationType); } } -} \ No newline at end of file +} diff --git a/test/Interlink.Test/BasicTests.cs b/test/Interlink.Test/BasicTests.cs index 84c09a0..8dc2a59 100644 --- a/test/Interlink.Test/BasicTests.cs +++ b/test/Interlink.Test/BasicTests.cs @@ -9,7 +9,7 @@ public async Task CanSend_EchoRequest_ReturnsResponse() services.AddInterlink(null, typeof(BasicTests).Assembly); // Resolve ISender from the service provider - var serviceProvider = services.BuildServiceProvider(); + var serviceProvider = services.BuildServiceProvider(); var sender = serviceProvider.GetRequiredService(); var response = await sender.Send(new EchoRequest("Hello")); @@ -25,4 +25,4 @@ public Task Handle(EchoRequest request, CancellationToken cancellationTo => Task.FromResult($"Echo: {request.Message}"); } -} +} \ No newline at end of file diff --git a/test/Interlink.Test/NotificationTests.cs b/test/Interlink.Test/NotificationTests.cs index 7aca126..d32f011 100644 --- a/test/Interlink.Test/NotificationTests.cs +++ b/test/Interlink.Test/NotificationTests.cs @@ -29,7 +29,7 @@ public async Task Publishes_Notification_ToAllHandlers() } } -public class MyTestNotification : INotification +public class MyTestNotification : INotification { public string Message { get; set; } = "Test notification"; } @@ -45,7 +45,6 @@ public class NotificationCounter public void Reset() => Interlocked.Exchange(ref _count, 0); } - public class Handler1 : INotificationHandler { private readonly NotificationCounter _counter; diff --git a/test/Interlink.Test/PipelineBehaviorTests.cs b/test/Interlink.Test/PipelineBehaviorTests.cs index 380c4da..19b6716 100644 --- a/test/Interlink.Test/PipelineBehaviorTests.cs +++ b/test/Interlink.Test/PipelineBehaviorTests.cs @@ -21,14 +21,14 @@ public async Task Executes_PipelineBehaviors_InOrder() } public class LoggingBehavior : IPipelineBehavior - where TRequest : notnull + where TRequest : notnull { - public static List Logs = new(); + public static List Logs = new(); public async Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken) { Logs.Add($"[Before] {typeof(TRequest).Name}"); - var response = await next(cancellationToken); + var response = await next(cancellationToken); Logs.Add($"[After] {typeof(TRequest).Name}"); return response; } diff --git a/test/Interlink.Test/StreamingTests.cs b/test/Interlink.Test/StreamingTests.cs new file mode 100644 index 0000000..3185a15 --- /dev/null +++ b/test/Interlink.Test/StreamingTests.cs @@ -0,0 +1,177 @@ +using System.Runtime.CompilerServices; + +namespace Interlink.Test; + +public class StreamingTests +{ + [Fact] + public async Task CreateStream_YieldsAllItems_InOrder() + { + var services = new ServiceCollection(); + services.AddInterlink(null, typeof(StreamingTests).Assembly); + + await using var provider = services.BuildServiceProvider(); + var sender = provider.GetRequiredService(); + + var items = new List(); + await foreach (var n in sender.CreateStream(new NumberStreamRequest(1, 5))) + items.Add(n); + + Assert.Equal(new[] { 1, 2, 3, 4, 5 }, items); + } + + [Fact] + public async Task CreateStream_ViaMediator_Works() + { + var services = new ServiceCollection(); + services.AddInterlink(null, typeof(StreamingTests).Assembly); + + await using var provider = services.BuildServiceProvider(); + var mediator = provider.GetRequiredService(); + + var items = new List(); + await foreach (var s in mediator.CreateStream(new StringStreamRequest())) + items.Add(s); + + Assert.Equal(new[] { "a", "b", "c" }, items); + } + + [Fact] + public async Task CreateStream_NoHandler_ThrowsHandlerNotFoundException() + { + var services = new ServiceCollection(); + services.AddInterlink(null, typeof(StreamingTests).Assembly); + + await using var provider = services.BuildServiceProvider(); + var sender = provider.GetRequiredService(); + + await Assert.ThrowsAsync(async () => + { + await foreach (var _ in sender.CreateStream(new OrphanStreamRequest())) + { + } + }); + } + + [Fact] + public async Task CreateStream_HonorsCancellation() + { + var services = new ServiceCollection(); + services.AddInterlink(null, typeof(StreamingTests).Assembly); + + await using var provider = services.BuildServiceProvider(); + var sender = provider.GetRequiredService(); + + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(async () => + { + await foreach (var _ in sender.CreateStream(new SlowStreamRequest(), cts.Token)) + { + } + }); + } + + [Fact] + public async Task CreateStream_WithStreamBehavior_InvokesBehavior() + { + var services = new ServiceCollection(); + services.AddSingleton(); + services.AddInterlink(options => + { + // Open generic behaviors must be registered explicitly (not via type scan) + options.AddStreamBehavior(typeof(CountingStreamBehavior<,>)); + }, typeof(StreamingTests).Assembly); + + await using var provider = services.BuildServiceProvider(); + var sender = provider.GetRequiredService(); + var probe = provider.GetRequiredService(); + + var count = 0; + await foreach (var _ in sender.CreateStream(new NumberStreamRequest(1, 3))) + count++; + + Assert.Equal(3, count); + Assert.True(probe.Entered); + Assert.True(probe.Exited); + } +} + +// --- top-level test types (open generics must not be nested for DI) --- + +public sealed record NumberStreamRequest(int From, int To) : IStreamRequest; + +public sealed class NumberStreamHandler : IStreamRequestHandler +{ + public async IAsyncEnumerable Handle( + NumberStreamRequest request, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + for (var i = request.From; i <= request.To; i++) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return i; + await Task.Yield(); + } + } +} + +public sealed record StringStreamRequest : IStreamRequest; + +public sealed class StringStreamHandler : IStreamRequestHandler +{ + public async IAsyncEnumerable Handle( + StringStreamRequest request, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + foreach (var s in new[] { "a", "b", "c" }) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return s; + await Task.Yield(); + } + } +} + +public sealed record OrphanStreamRequest : IStreamRequest; + +public sealed record SlowStreamRequest : IStreamRequest; + +public sealed class SlowStreamHandler : IStreamRequestHandler +{ + public async IAsyncEnumerable Handle( + SlowStreamRequest request, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + await Task.Delay(Timeout.Infinite, cancellationToken); + yield break; + } +} + +public sealed class StreamBehaviorProbe +{ + public bool Entered { get; set; } + public bool Exited { get; set; } +} + +public sealed class CountingStreamBehavior + : IStreamPipelineBehavior + where TRequest : IStreamRequest +{ + private readonly StreamBehaviorProbe _probe; + + public CountingStreamBehavior(StreamBehaviorProbe probe) => _probe = probe; + + public async IAsyncEnumerable Handle( + TRequest request, + StreamHandlerDelegate next, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + _probe.Entered = true; + await foreach (var item in next().WithCancellation(cancellationToken)) + yield return item; + _probe.Exited = true; + } +} \ No newline at end of file