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