Streaming, notification pipelines, and timeouts

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<Unit>`)
 * 🔧 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<PetDto>;
+
+public sealed class ExportPetsStreamHandler : IStreamRequestHandler<ExportPetsStream, PetDto>
+{
+    public async IAsyncEnumerable<PetDto> 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<TNotification> : INotificationPipelineBehavior<TNotification>
+    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<TResponse>` but no corresponding `IRequestHandler<TRequest, TResponse>` 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<in TRequest, in TResponse> where TRequest
 }
 ```
 
+### Streaming requests
+```csharp
+public interface IStreamRequest<out TResponse> { }
+
+public interface IStreamRequestHandler<in TRequest, TResponse>
+    where TRequest : IStreamRequest<TResponse>
+{
+    IAsyncEnumerable<TResponse> 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<Unit>`) |
-| **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 ❤️
+
+<a href="https://github.com/manusoft/Interlink/graphs/contributors">
+  <img src="https://contrib.rocks/image?repo=manusoft/Interlink" />
+</a>
+

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<IActionResult> GetAllPets(CancellationToken cancellationToken)
         return Ok(pets);
     }
 
+    [HttpGet("stream")]
+    public async IAsyncEnumerable<Pet> GetAllPetsStream([EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        await foreach (var pet in mediator.CreateStream(new ExportPetsStream.Stream(), cancellationToken))
+        {
+            yield return pet;
+        }
+    }
+
     [HttpPost]
     public async Task<IActionResult> CreatePet([FromBody] CreatePet.Command command, CancellationToken cancellationToken)
     {

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<Pet>;
+
+    public class Handler(AppDbContext context) : IStreamRequestHandler<Stream, Pet>
+    {
+        public async IAsyncEnumerable<Pet> Handle(Stream request, [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await foreach (var pet in context.Pets.AsAsyncEnumerable().WithCancellation(cancellationToken))
+                yield return pet;
+        }
+    }
+}

src/Interlink.AspNetCore/Interlink.AspNetCore.csproj

@@ -8,7 +8,7 @@
 		<RootNamespace>Interlink.AspNetCore</RootNamespace>
 		<AssemblyName>Interlink.AspNetCore</AssemblyName>
 		<PackageId>Interlink.AspNetCore</PackageId>
-		<Version>1.5.2</Version>
+		<Version>1.6.0</Version>
 		<Description>ASP.NET Core integration for the Interlink mediator library (filters, exception handling, helpers).</Description>
 		<Copyright>Copyright © 2025-2026 ManuHub</Copyright>
 		<Authors>Manojbabu</Authors>

src/Interlink.Extensions.Logging/Interlink.Extensions.Logging.csproj

@@ -8,7 +8,7 @@
 		<RootNamespace>Interlink.Extensions.Logging</RootNamespace>
 		<AssemblyName>Interlink.Extensions.Logging</AssemblyName>
 		<PackageId>Interlink.Extensions.Logging</PackageId>
-		<Version>1.5.2</Version>
+		<Version>1.6.0</Version>
 		<Description>Built-in logging pipeline behavior for the Interlink mediator library.</Description>
 		<Copyright>Copyright © 2025-2026 ManuHub</Copyright>
 		<Authors>Manojbabu</Authors>

src/Interlink.Extensions.Validation/Interlink.Extensions.Validation.csproj

@@ -8,7 +8,7 @@
 		<RootNamespace>Interlink.Extensions.Validation</RootNamespace>
 		<AssemblyName>Interlink.Extensions.Validation</AssemblyName>
 		<PackageId>Interlink.Extensions.Validation</PackageId>
-		<Version>1.5.2</Version>
+		<Version>1.6.0</Version>
 		<Description>FluentValidation integration for the Interlink mediator library.</Description>
 		<Copyright>Copyright © 2025-2026 ManuHub</Copyright>
 		<Authors>Manojbabu</Authors>

src/Interlink/Abstractions/INotificationPipelineBehavior.cs

@@ -0,0 +1,28 @@
+using Interlink.Contracts;
+
+namespace Interlink;
+
+/// <summary>
+/// Represents a delegate that continues the notification pipeline.
+/// </summary>
+/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+/// <returns>A task that represents the asynchronous operation.</returns>
+public delegate Task NotificationHandlerDelegate(CancellationToken cancellationToken = default);
+
+/// <summary>
+/// Defines a behavior in the notification pipeline that can inspect or wrap
+/// the publishing of a notification before and/or after the handlers run.
+/// </summary>
+/// <typeparam name="TNotification">The type of the notification.</typeparam>
+public interface INotificationPipelineBehavior<in TNotification>
+    where TNotification : INotification
+{
+    /// <summary>
+    /// Handles the notification and optionally calls the next delegate in the pipeline.
+    /// </summary>
+    /// <param name="notification">The notification being published.</param>
+    /// <param name="next">The next delegate in the pipeline (remaining behaviors + handlers).</param>
+    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    Task Handle(TNotification notification, NotificationHandlerDelegate next, CancellationToken cancellationToken);
+}

src/Interlink/Abstractions/ISender.cs

@@ -16,4 +16,14 @@ public interface ISender
     /// Sends a void/command request that does not produce a response value.
     /// </summary>
     Task Send(IRequest request, CancellationToken cancellationToken = default);
-}
+
+    /// <summary>
+    /// Sends a stream request and returns an asynchronous stream of responses.
+    /// </summary>
+    /// <typeparam name="TResponse">The type of each item in the response stream.</typeparam>
+    /// <param name="request">The stream request to send.</param>
+    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+    /// <returns>An asynchronous stream of response items.</returns>
+    /// <exception cref="HandlerNotFoundException">Thrown when no stream handler is registered for the request type.</exception>
+    IAsyncEnumerable<TResponse> CreateStream<TResponse>(IStreamRequest<TResponse> request, CancellationToken cancellationToken = default);
+}

src/Interlink/Abstractions/IStreamPipelineBehavior.cs

@@ -0,0 +1,32 @@
+using Interlink.Contracts;
+
+namespace Interlink;
+
+/// <summary>
+/// Represents a delegate that continues the stream request pipeline.
+/// </summary>
+/// <typeparam name="TResponse">The type of each streamed response item.</typeparam>
+/// <returns>An asynchronous stream of response items.</returns>
+public delegate IAsyncEnumerable<TResponse> StreamHandlerDelegate<TResponse>();
+
+/// <summary>
+/// Defines a behavior in the stream request pipeline that can inspect or wrap
+/// streaming before and/or after the next component.
+/// </summary>
+/// <typeparam name="TRequest">The type of the stream request.</typeparam>
+/// <typeparam name="TResponse">The type of each streamed response item.</typeparam>
+public interface IStreamPipelineBehavior<in TRequest, TResponse>
+    where TRequest : IStreamRequest<TResponse>
+{
+    /// <summary>
+    /// Handles the stream request and optionally calls the next delegate in the pipeline.
+    /// </summary>
+    /// <param name="request">The request to process.</param>
+    /// <param name="next">The next delegate in the pipeline.</param>
+    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+    /// <returns>An asynchronous stream of response items.</returns>
+    IAsyncEnumerable<TResponse> Handle(
+        TRequest request,
+        StreamHandlerDelegate<TResponse> next,
+        CancellationToken cancellationToken);
+}

src/Interlink/Abstractions/IStreamRequestHandler.cs

@@ -0,0 +1,21 @@
+using Interlink.Contracts;
+
+namespace Interlink;
+
+/// <summary>
+/// Defines a handler for a streaming request of type <typeparamref name="TRequest"/>
+/// that yields a sequence of <typeparamref name="TResponse"/> values.
+/// </summary>
+/// <typeparam name="TRequest">The type of the stream request.</typeparam>
+/// <typeparam name="TResponse">The type of each streamed response item.</typeparam>
+public interface IStreamRequestHandler<in TRequest, TResponse>
+    where TRequest : IStreamRequest<TResponse>
+{
+    /// <summary>
+    /// Handles the stream request and yields response items asynchronously.
+    /// </summary>
+    /// <param name="request">The request instance to handle.</param>
+    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+    /// <returns>An asynchronous stream of response items.</returns>
+    IAsyncEnumerable<TResponse> Handle(TRequest request, CancellationToken cancellationToken);
+}