feat: add pub/sub support to dispatcher

Notifications are broadcast to zero or more handlers with configurable
publishing strategies (sequential or parallel) and pipeline behavior
support.

src/request.dispatcher/INotification.cs

@@ -0,0 +1,13 @@
+// Copyright (c) The Geekeey Authors
+// SPDX-License-Identifier: EUPL-1.2
+
+namespace Geekeey.Request.Dispatcher;
+
+/// <summary>
+/// Represents a notification message that can be published to zero or more
+/// <see cref="INotificationHandler{TNotification}"/> handlers.
+/// This interface serves as a marker for distinguishing notification types.
+/// </summary>
+public interface INotification
+{
+}

src/request.dispatcher/INotificationBehavior.cs

@@ -0,0 +1,34 @@
+// Copyright (c) The Geekeey Authors
+// SPDX-License-Identifier: EUPL-1.2
+
+#pragma warning disable CA1711
+#pragma warning disable CA1716
+
+namespace Geekeey.Request.Dispatcher;
+
+/// <summary>
+/// Represents a behavior in the notification pipeline, allowing interception, modification,
+/// or chaining of notification processing.
+/// </summary>
+/// <typeparam name="TNotification">The type of notification being processed. Must implement <see cref="INotification"/>.</typeparam>
+public interface INotificationBehavior<TNotification> where TNotification : INotification
+{
+	/// <summary>
+	/// Handles the asynchronous processing of a notification, allowing behavior customization
+	/// such as interception, modification, or chaining of the notification pipeline.
+	/// </summary>
+	/// <param name="notification">The notification instance being processed.</param>
+	/// <param name="next">The next delegate in the pipeline to execute after the custom behavior.</param>
+	/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+	/// <returns>A task representing the asynchronous operation.</returns>
+	Task HandleAsync(TNotification notification, NotificationHandlerDelegate<TNotification> next, CancellationToken cancellationToken);
+}
+
+/// <summary>
+/// Represents the delegate responsible for handling notifications in a pipeline.
+/// </summary>
+/// <typeparam name="TNotification">The type of notification being processed. Must implement <see cref="INotification"/>.</typeparam>
+/// <param name="notification">The notification instance being processed.</param>
+/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+/// <returns>A task representing the asynchronous operation.</returns>
+public delegate Task NotificationHandlerDelegate<TNotification>(TNotification notification, CancellationToken cancellationToken) where TNotification : INotification;

src/request.dispatcher/INotificationHandler.cs

@@ -0,0 +1,19 @@
+// Copyright (c) The Geekeey Authors
+// SPDX-License-Identifier: EUPL-1.2
+
+namespace Geekeey.Request.Dispatcher;
+
+/// <summary>
+/// Defines a handler for processing notifications of a specific type.
+/// </summary>
+/// <typeparam name="TNotification">The type of notification to handle. Must implement <see cref="INotification"/>.</typeparam>
+public interface INotificationHandler<in TNotification> where TNotification : INotification
+{
+	/// <summary>
+	/// Handles the specified notification asynchronously.
+	/// </summary>
+	/// <param name="notification">The notification object to process.</param>
+	/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+	/// <returns>A task representing the asynchronous operation.</returns>
+	Task HandleAsync(TNotification notification, CancellationToken cancellationToken);
+}

src/request.dispatcher/INotificationPublisher.cs

@@ -0,0 +1,20 @@
+// Copyright (c) The Geekeey Authors
+// SPDX-License-Identifier: EUPL-1.2
+
+namespace Geekeey.Request.Dispatcher;
+
+/// <summary>
+/// Defines a strategy for publishing notifications to registered handlers.
+/// </summary>
+public interface INotificationPublisher
+{
+	/// <summary>
+	/// Publishes the specified notification to the given collection of handlers.
+	/// </summary>
+	/// <typeparam name="TNotification">The type of notification being published. Must implement <see cref="INotification"/>.</typeparam>
+	/// <param name="handlers">The list of handlers to invoke.</param>
+	/// <param name="notification">The notification instance to publish.</param>
+	/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+	/// <returns>A task representing the asynchronous publish operation.</returns>
+	Task PublishAsync<TNotification>(IEnumerable<INotificationHandler<TNotification>> handlers, TNotification notification, CancellationToken cancellationToken) where TNotification : INotification;
+}

src/request.dispatcher/IRequestDispatcher.cs

@@ -25,4 +25,13 @@ public interface IRequestDispatcher
 	/// <typeparam name="TResponse">Response type</typeparam>
 	/// <returns>The created async enumerable, representing the stream of responses.</returns>
 	IAsyncEnumerable<TResponse> DispatchAsync<TResponse>(IStreamRequest<TResponse> request, CancellationToken cancellationToken = default);
+
+	/// <summary>
+	/// Asynchronously publish a notification to all registered handlers.
+	/// </summary>
+	/// <param name="notification">Notification object</param>
+	/// <param name="publisher">Optional custom notification publisher</param>
+	/// <param name="cancellationToken">Optional cancellation token</param>
+	/// <returns>A task that represents the publish operation.</returns>
+	Task PublishAsync(INotification notification, INotificationPublisher? publisher = null, CancellationToken cancellationToken = default);
 }

src/request.dispatcher/NotificationInvoker.cs

@@ -0,0 +1,50 @@
+// Copyright (c) The Geekeey Authors
+// SPDX-License-Identifier: EUPL-1.2
+
+using System.Diagnostics;
+
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+
+namespace Geekeey.Request.Dispatcher;
+
+internal abstract class NotificationInvoker
+{
+	public abstract Task HandleAsync(object notification, IServiceProvider serviceProvider, INotificationPublisher? publisher, CancellationToken cancellationToken);
+}
+
+internal sealed class NotificationInvoker<TNotification> : NotificationInvoker
+	where TNotification : INotification
+{
+	public override Task HandleAsync(object notification, IServiceProvider serviceProvider, INotificationPublisher? publisher, CancellationToken cancellationToken)
+	{
+		return HandleAsync((TNotification)notification, serviceProvider, publisher, cancellationToken);
+	}
+
+	public Task HandleAsync(TNotification notification, IServiceProvider serviceProvider, INotificationPublisher? publisher, CancellationToken cancellationToken)
+	{
+		var options = serviceProvider.GetRequiredService<IOptions<RequestDispatcherOptions>>().Value;
+
+		var handlers = options.GetRequestHandlers<INotificationHandler<TNotification>>(serviceProvider).ToList();
+
+		publisher ??= serviceProvider.GetRequiredService<INotificationPublisher>();
+
+		var behaviors = options.GetRequestBehaviors<INotificationBehavior<TNotification>>(serviceProvider);
+
+		var pipeline = behaviors
+			.Reverse()
+			.Aggregate((NotificationHandlerDelegate<TNotification>)Terminal, Chain);
+
+		return pipeline(notification, cancellationToken);
+
+		static NotificationHandlerDelegate<TNotification> Chain(NotificationHandlerDelegate<TNotification> next, INotificationBehavior<TNotification> behavior)
+		{
+			return [StackTraceHidden] (n, ct) => behavior.HandleAsync(n, next, ct);
+		}
+
+		Task Terminal(TNotification n, CancellationToken ct)
+		{
+			return publisher.PublishAsync(handlers, n, ct);
+		}
+	}
+}

src/request.dispatcher/ParallelNotificationPublisher.cs

@@ -0,0 +1,47 @@
+// Copyright (c) The Geekeey Authors
+// SPDX-License-Identifier: EUPL-1.2
+
+namespace Geekeey.Request.Dispatcher;
+
+internal sealed class ParallelNotificationPublisher : INotificationPublisher
+{
+	private readonly int _maxDegreeOfParallelism;
+
+	public ParallelNotificationPublisher(int? maxDegreeOfParallelism = null)
+	{
+		_maxDegreeOfParallelism = maxDegreeOfParallelism ?? Environment.ProcessorCount;
+	}
+
+	public async Task PublishAsync<TNotification>(IEnumerable<INotificationHandler<TNotification>> handlers, TNotification notification, CancellationToken cancellationToken)
+		where TNotification : INotification
+	{
+		var exceptions = new List<Exception>();
+		var @lock = new object();
+
+		await Parallel.ForEachAsync(handlers,
+			new ParallelOptions
+			{
+				CancellationToken = cancellationToken,
+				MaxDegreeOfParallelism = _maxDegreeOfParallelism,
+			},
+			async (handler, ct) =>
+			{
+				try
+				{
+					await handler.HandleAsync(notification, ct);
+				}
+				catch (Exception e) when (e is not OperationCanceledException)
+				{
+					lock (@lock)
+					{
+						exceptions.Add(e);
+					}
+				}
+			});
+
+		if (exceptions.Count > 0)
+		{
+			throw new AggregateException(exceptions);
+		}
+	}
+}

src/request.dispatcher/RequestDispatcher.cs

@@ -12,6 +12,7 @@ internal sealed class RequestDispatcher : IRequestDispatcher
 {
 	private static readonly ConcurrentDictionary<(Type, Type), ScalarRequestInvoker> ScalarRequestHandlers = new();
 	private static readonly ConcurrentDictionary<(Type, Type), StreamRequestInvoker> StreamRequestHandlers = new();
+	private static readonly ConcurrentDictionary<Type, NotificationInvoker> NotificationHandlers = new();
 
 	private readonly IServiceProvider _serviceProvider;
 
@@ -24,6 +25,7 @@ internal sealed class RequestDispatcher : IRequestDispatcher
 	{
 		ScalarRequestHandlers.Clear();
 		StreamRequestHandlers.Clear();
+		NotificationHandlers.Clear();
 	}
 
 	public Task<TResponse> DispatchAsync<TResponse>(IScalarRequest<TResponse> request, CancellationToken cancellationToken = default)
@@ -51,4 +53,17 @@ internal sealed class RequestDispatcher : IRequestDispatcher
 
 		return ((StreamRequestInvoker<TResponse>)handler).HandleAsync(request, _serviceProvider, cancellationToken);
 	}
+
+	public Task PublishAsync(INotification notification, INotificationPublisher? publisher = null, CancellationToken cancellationToken = default)
+	{
+		ArgumentNullException.ThrowIfNull(notification);
+
+		var handler = NotificationHandlers.GetOrAdd(notification.GetType(), static key =>
+		{
+			var type = typeof(NotificationInvoker<>).MakeGenericType(key);
+			return (NotificationInvoker)Activator.CreateInstance(type)!;
+		});
+
+		return handler.HandleAsync(notification, _serviceProvider, publisher, cancellationToken);
+	}
 }

src/request.dispatcher/RequestDispatcherBuilderExtensions.cs

@@ -150,6 +150,70 @@ public static class RequestDispatcherBuilderExtensions
 		return builder;
 	}
 
+	/// <summary>
+	/// Registers <typeparamref name="T"/> as the default <see cref="INotificationPublisher"/>
+	/// in the dependency injection container.
+	/// </summary>
+	/// <typeparam name="T">The type implementing <see cref="INotificationPublisher"/> to register.</typeparam>
+	/// <param name="builder">The <see cref="IRequestDispatcherBuilder"/> to configure.</param>
+	/// <param name="lifetime">The lifetime with which the publisher is registered (default <see cref="ServiceLifetime.Transient"/>).</param>
+	/// <returns>The <see cref="IRequestDispatcherBuilder"/> instance for further configuration.</returns>
+	public static IRequestDispatcherBuilder UseNotificationPublisher<T>(this IRequestDispatcherBuilder builder, ServiceLifetime lifetime = ServiceLifetime.Transient)
+		where T : class, INotificationPublisher
+	{
+		ArgumentNullException.ThrowIfNull(builder);
+
+		builder.Services.Replace(new ServiceDescriptor(typeof(INotificationPublisher), typeof(T), lifetime));
+
+		return builder;
+	}
+
+	/// <summary>
+	/// Registers the <see cref="SequentialNotificationPublisher"/> as the default
+	/// <see cref="INotificationPublisher"/>.
+	/// </summary>
+	/// <param name="builder">The <see cref="IRequestDispatcherBuilder"/> to configure.</param>
+	/// <returns>The <see cref="IRequestDispatcherBuilder"/> instance for further configuration.</returns>
+	public static IRequestDispatcherBuilder UseSequentialNotificationPublisher(this IRequestDispatcherBuilder builder)
+	{
+		ArgumentNullException.ThrowIfNull(builder);
+
+		builder.Services.Replace(new ServiceDescriptor(typeof(INotificationPublisher), new SequentialNotificationPublisher()));
+
+		return builder;
+	}
+
+	/// <summary>
+	/// Registers the <see cref="ParallelNotificationPublisher"/> as the default
+	/// <see cref="INotificationPublisher"/> using the specified maximum degree of parallelism.
+	/// </summary>
+	/// <param name="builder">The <see cref="IRequestDispatcherBuilder"/> to configure.</param>
+	/// <returns>The <see cref="IRequestDispatcherBuilder"/> instance for further configuration.</returns>
+	public static IRequestDispatcherBuilder UseParallelNotificationPublisher(this IRequestDispatcherBuilder builder)
+	{
+		ArgumentNullException.ThrowIfNull(builder);
+
+		builder.Services.Replace(new ServiceDescriptor(typeof(INotificationPublisher), new ParallelNotificationPublisher()));
+
+		return builder;
+	}
+
+	/// <summary>
+	/// Registers the <see cref="ParallelNotificationPublisher"/> as the default
+	/// <see cref="INotificationPublisher"/> using the specified maximum degree of parallelism.
+	/// </summary>
+	/// <param name="builder">The <see cref="IRequestDispatcherBuilder"/> to configure.</param>
+	/// <param name="maxDegreeOfParallelism">The maximum number of handlers to execute concurrently.</param>
+	/// <returns>The <see cref="IRequestDispatcherBuilder"/> instance for further configuration.</returns>
+	public static IRequestDispatcherBuilder UseParallelNotificationPublisher(this IRequestDispatcherBuilder builder, int maxDegreeOfParallelism)
+	{
+		ArgumentNullException.ThrowIfNull(builder);
+
+		builder.Services.Replace(new ServiceDescriptor(typeof(INotificationPublisher), new ParallelNotificationPublisher(maxDegreeOfParallelism)));
+
+		return builder;
+	}
+
 	private static IReadOnlyCollection<Type> ValidateNoNestedRequestHandlers(IReadOnlyCollection<Type> types, [CallerMemberName] string? invoker = null)
 	{
 		var nestedHandlers = types

src/request.dispatcher/RequestDispatcherOptions.cs

@@ -94,7 +94,8 @@ internal sealed class RequestDispatcherOptions
 	{
 		return type.IsGenericType &&
 			   (type.GetGenericTypeDefinition() == typeof(IScalarRequestHandler<,>) ||
-				type.GetGenericTypeDefinition() == typeof(IStreamRequestHandler<,>));
+				type.GetGenericTypeDefinition() == typeof(IStreamRequestHandler<,>) ||
+				type.GetGenericTypeDefinition() == typeof(INotificationHandler<>));
 	}
 
 	private sealed class HandlerTypeIndex(IEnumerable<Type> collection)
@@ -110,7 +111,6 @@ internal sealed class RequestDispatcherOptions
 			}
 
 			var requestType = @interface.GetGenericArguments()[0];
-			_ = @interface.GetGenericArguments()[1];
 
 			foreach (var type in _openTypeInfo)
 			{
@@ -154,7 +154,8 @@ internal sealed class RequestDispatcherOptions
 	{
 		return type.IsGenericType &&
 			   (type.GetGenericTypeDefinition() == typeof(IScalarRequestBehavior<,>) ||
-				type.GetGenericTypeDefinition() == typeof(IStreamRequestBehavior<,>));
+				type.GetGenericTypeDefinition() == typeof(IStreamRequestBehavior<,>) ||
+				type.GetGenericTypeDefinition() == typeof(INotificationBehavior<>));
 	}
 
 	private sealed class BehaviorTypeIndex(IEnumerable<Type> collection)
@@ -169,15 +170,12 @@ internal sealed class RequestDispatcherOptions
 				result.AddRange(list);
 			}
 
-			var requestType = @interface.GetGenericArguments()[0];
-			var responseType = @interface.GetGenericArguments()[1];
-
 			foreach (var behaviour in _openTypeInfo)
 			{
 				try
 				{
 					// open type case one: Behaviour<TRequest, TResponse> : IRequestBehaviour<TRequest, TResponse>
-					var impl = behaviour.MakeGenericType(requestType, responseType);
+					var impl = behaviour.MakeGenericType(@interface.GetGenericArguments());
 					if (impl.IsAssignableTo(@interface))
 					{
 						result.Add(impl);
@@ -190,6 +188,7 @@ internal sealed class RequestDispatcherOptions
 				try
 				{
 					// open type case two: Behaviour<T> : IRequestBehaviour<WrapperRequest<T>, TResponse>
+					var requestType = @interface.GetGenericArguments()[0];
 					var impl = behaviour.MakeGenericType(requestType.GetGenericArguments());
 					if (impl.IsAssignableTo(@interface))
 					{

src/request.dispatcher/SequentialNotificationPublisher.cs

@@ -0,0 +1,16 @@
+// Copyright (c) The Geekeey Authors
+// SPDX-License-Identifier: EUPL-1.2
+
+namespace Geekeey.Request.Dispatcher;
+
+internal sealed class SequentialNotificationPublisher : INotificationPublisher
+{
+	public async Task PublishAsync<TNotification>(IEnumerable<INotificationHandler<TNotification>> handlers, TNotification notification, CancellationToken cancellationToken)
+		where TNotification : INotification
+	{
+		foreach (var handler in handlers)
+		{
+			await handler.HandleAsync(notification, cancellationToken);
+		}
+	}
+}