From 9232716901bc0f8cef613db242b3d54b5c06b78a Mon Sep 17 00:00:00 2001 From: Louis Seubert Date: Wed, 1 Jul 2026 22:00:05 +0200 Subject: [PATCH] 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. --- CHANGELOG.md | 1 + .../NotificationBehaviourTests.cs | 74 +++++++++ .../NotificationDispatcherTests.cs | 143 +++++++++++++++++ .../NotificationPublisherStrategyTests.cs | 146 ++++++++++++++++++ .../_fixtures/BlockingNotificationBehavior.cs | 13 ++ .../_fixtures/CustomNotificationPublisher.cs | 17 ++ .../_fixtures/FailingNotificationHandler.cs | 12 ++ .../NotificationChainedBehaviour1.cs | 13 ++ .../NotificationChainedBehaviour2.cs | 13 ++ .../_fixtures/NotificationTestTracker.cs | 9 ++ .../_fixtures/OpenNotificationHandler.cs | 14 ++ .../_fixtures/OtherNotificationPublisher.cs | 17 ++ .../SecondTestNotificationHandler.cs | 13 ++ .../_fixtures/TestNotification.cs | 9 ++ .../_fixtures/TestNotificationBehavior.cs | 14 ++ .../_fixtures/TestNotificationHandler.cs | 13 ++ src/request.dispatcher/INotification.cs | 13 ++ .../INotificationBehavior.cs | 34 ++++ .../INotificationHandler.cs | 19 +++ .../INotificationPublisher.cs | 20 +++ src/request.dispatcher/IRequestDispatcher.cs | 9 ++ src/request.dispatcher/NotificationInvoker.cs | 50 ++++++ .../ParallelNotificationPublisher.cs | 47 ++++++ src/request.dispatcher/RequestDispatcher.cs | 15 ++ .../RequestDispatcherBuilderExtensions.cs | 64 ++++++++ .../RequestDispatcherOptions.cs | 13 +- .../SequentialNotificationPublisher.cs | 16 ++ 27 files changed, 814 insertions(+), 7 deletions(-) create mode 100644 src/request.dispatcher.tests/NotificationBehaviourTests.cs create mode 100644 src/request.dispatcher.tests/NotificationDispatcherTests.cs create mode 100644 src/request.dispatcher.tests/NotificationPublisherStrategyTests.cs create mode 100644 src/request.dispatcher.tests/_fixtures/BlockingNotificationBehavior.cs create mode 100644 src/request.dispatcher.tests/_fixtures/CustomNotificationPublisher.cs create mode 100644 src/request.dispatcher.tests/_fixtures/FailingNotificationHandler.cs create mode 100644 src/request.dispatcher.tests/_fixtures/NotificationChainedBehaviour1.cs create mode 100644 src/request.dispatcher.tests/_fixtures/NotificationChainedBehaviour2.cs create mode 100644 src/request.dispatcher.tests/_fixtures/NotificationTestTracker.cs create mode 100644 src/request.dispatcher.tests/_fixtures/OpenNotificationHandler.cs create mode 100644 src/request.dispatcher.tests/_fixtures/OtherNotificationPublisher.cs create mode 100644 src/request.dispatcher.tests/_fixtures/SecondTestNotificationHandler.cs create mode 100644 src/request.dispatcher.tests/_fixtures/TestNotification.cs create mode 100644 src/request.dispatcher.tests/_fixtures/TestNotificationBehavior.cs create mode 100644 src/request.dispatcher.tests/_fixtures/TestNotificationHandler.cs create mode 100644 src/request.dispatcher/INotification.cs create mode 100644 src/request.dispatcher/INotificationBehavior.cs create mode 100644 src/request.dispatcher/INotificationHandler.cs create mode 100644 src/request.dispatcher/INotificationPublisher.cs create mode 100644 src/request.dispatcher/NotificationInvoker.cs create mode 100644 src/request.dispatcher/ParallelNotificationPublisher.cs create mode 100644 src/request.dispatcher/SequentialNotificationPublisher.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 240c251..a221e06 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ To have a consistent experience across all packages, some public interfaces have ### Added - **request.validation:** Support `PropertyPath` JSON converter for string values and dictionary property names +- **request.dispatcher:** Notification (pub/sub) support with `INotification` ### Changed diff --git a/src/request.dispatcher.tests/NotificationBehaviourTests.cs b/src/request.dispatcher.tests/NotificationBehaviourTests.cs new file mode 100644 index 0000000..9794729 --- /dev/null +++ b/src/request.dispatcher.tests/NotificationBehaviourTests.cs @@ -0,0 +1,74 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +using Microsoft.Extensions.DependencyInjection; + +namespace Geekeey.Request.Dispatcher.Tests; + +internal sealed class NotificationBehaviourTests +{ + [Test] + public async Task I_can_execute_a_single_behaviour_wrapping_the_publisher() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .Add(typeof(TestNotificationBehavior)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "Wrapped" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(3); + await Assert.That(tracker.Log[0]).IsEquivalentTo("Before"); + await Assert.That(tracker.Log[1]).IsEquivalentTo("Handler:Wrapped"); + await Assert.That(tracker.Log[2]).IsEquivalentTo("After"); + } + + [Test] + public async Task I_can_chain_behaviours_in_registration_order() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .Add(typeof(NotificationChainedBehaviour1)) + .Add(typeof(NotificationChainedBehaviour2)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "Chain" }); + + // behaviours.Reverse() wrapping means Behaviour1 wraps Behaviour2 wraps publisher + // Execution: Behaviour1 before -> Behaviour2 before -> handler -> Behaviour2 after -> Behaviour1 after + await Assert.That(tracker.Log).Count().IsEqualTo(3); + await Assert.That(tracker.Log[0]).IsEquivalentTo("Behaviour1"); + await Assert.That(tracker.Log[1]).IsEquivalentTo("Behaviour2"); + await Assert.That(tracker.Log[2]).IsEquivalentTo("Handler:Chain"); + } + + [Test] + public async Task I_can_see_a_behaviour_that_exits_early_blocks_subsequent_execution() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .Add(typeof(BlockingNotificationBehavior)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "Blocked" }); + + // BlockingNotificationBehavior does not call next, so the handler should not execute. + await Assert.That(tracker.Log).Count().IsEqualTo(1); + await Assert.That(tracker.Log[0]).IsEquivalentTo("Blocked"); + } +} diff --git a/src/request.dispatcher.tests/NotificationDispatcherTests.cs b/src/request.dispatcher.tests/NotificationDispatcherTests.cs new file mode 100644 index 0000000..e820052 --- /dev/null +++ b/src/request.dispatcher.tests/NotificationDispatcherTests.cs @@ -0,0 +1,143 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +using Microsoft.Extensions.DependencyInjection; + +namespace Geekeey.Request.Dispatcher.Tests; + +internal sealed class NotificationDispatcherTests +{ + [Test] + public async Task I_can_publish_a_notification_to_a_single_handler() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "Hello" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(1); + await Assert.That(tracker.Log[0]).IsEquivalentTo("Handler:Hello"); + } + + [Test] + public async Task I_can_publish_a_notification_to_multiple_handlers() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .Add(typeof(SecondTestNotificationHandler)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "Multi" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(2); + await Assert.That(tracker.Log[0]).IsEquivalentTo("Handler:Multi"); + await Assert.That(tracker.Log[1]).IsEquivalentTo("SecondHandler:Multi"); + } + + [Test] + public async Task I_can_publish_a_notification_with_zero_handlers() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "NoOp" }); + } + + [Test] + public async Task I_can_publish_a_notification_to_an_open_generic_handler() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(OpenNotificationHandler<>)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "Open" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(1); + await Assert.That(tracker.Log[0]).IsEquivalentTo("OpenHandler:TestNotification"); + } + + [Test] + public async Task I_can_see_the_exception_propagate_from_a_failing_handler() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(FailingNotificationHandler)) + .Add(typeof(SecondTestNotificationHandler)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + + var ex = await Assert.ThrowsAsync(() => + dispatcher.PublishAsync(new TestNotification { Value = "Fail" })); + + await Assert.That(ex?.Message).IsEquivalentTo("Handler failed"); + } + + [Test] + public async Task I_can_see_subsequent_handlers_are_skipped_after_a_handler_failure() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(FailingNotificationHandler)) + .Add(typeof(SecondTestNotificationHandler)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + try + { + await dispatcher.PublishAsync(new TestNotification { Value = "Skip" }); + } + catch (InvalidOperationException) + { + } + + // Failing handler throws, so the second handler should not be executed + await Assert.That(tracker.Log).IsEmpty(); + } + + [Test] + public async Task I_can_see_inherited_notification_types_are_handled() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(OpenNotificationHandler<>)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new DerivedNotification { Value = "Derived" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(1); + await Assert.That(tracker.Log[0]).IsEquivalentTo("OpenHandler:DerivedNotification"); + } + + private sealed class DerivedNotification : TestNotification + { + } +} diff --git a/src/request.dispatcher.tests/NotificationPublisherStrategyTests.cs b/src/request.dispatcher.tests/NotificationPublisherStrategyTests.cs new file mode 100644 index 0000000..0c9d0d1 --- /dev/null +++ b/src/request.dispatcher.tests/NotificationPublisherStrategyTests.cs @@ -0,0 +1,146 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +using Microsoft.Extensions.DependencyInjection; + +namespace Geekeey.Request.Dispatcher.Tests; + +internal sealed class NotificationPublisherStrategyTests +{ + [Test] + public async Task I_can_use_a_custom_publisher_via_per_call_parameter() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler))); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + var customPub = new CustomNotificationPublisher(tracker); + + await dispatcher.PublishAsync(new TestNotification { Value = "PerCall" }, publisher: customPub); + + await Assert.That(tracker.Log).Count().IsEqualTo(2); + await Assert.That(tracker.Log[0]).IsEquivalentTo("CustomPublisher"); + await Assert.That(tracker.Log[1]).IsEquivalentTo("Handler:PerCall"); + } + + [Test] + public async Task I_can_use_a_custom_publisher_via_di_registration() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .UseNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "DI" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(2); + await Assert.That(tracker.Log[0]).IsEquivalentTo("CustomPublisher"); + await Assert.That(tracker.Log[1]).IsEquivalentTo("Handler:DI"); + } + + [Test] + public async Task I_can_use_the_default_sequential_publisher_when_no_custom_publisher_is_registered() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .UseSequentialNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "Default" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(1); + await Assert.That(tracker.Log[0]).IsEquivalentTo("Handler:Default"); + } + + [Test] + public async Task I_can_see_per_call_publisher_wins_over_di_registered_publisher() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .UseNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + // Use a different publisher per-call that doesn't log "CustomPublisher" + var otherPub = new OtherNotificationPublisher(tracker); + + await dispatcher.PublishAsync(new TestNotification { Value = "Override" }, publisher: otherPub); + + await Assert.That(tracker.Log).Count().IsEqualTo(2); + await Assert.That(tracker.Log[0]).IsEquivalentTo("OtherPublisher"); + await Assert.That(tracker.Log[1]).IsEquivalentTo("Handler:Override"); + } + + [Test] + public async Task I_can_use_the_parallel_publisher_via_di_registration() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .Add(typeof(SecondTestNotificationHandler)) + .UseParallelNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "Parallel" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(2); + await Assert.That(tracker.Log).Contains("Handler:Parallel"); + await Assert.That(tracker.Log).Contains("SecondHandler:Parallel"); + } + + [Test] + public async Task I_can_use_the_parallel_publisher_with_max_degree_of_parallelism() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(TestNotificationHandler)) + .Add(typeof(SecondTestNotificationHandler)) + .UseParallelNotificationPublisher(maxDegreeOfParallelism: 1)); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + await dispatcher.PublishAsync(new TestNotification { Value = "SequentialParallel" }); + + await Assert.That(tracker.Log).Count().IsEqualTo(2); + } + + [Test] + public async Task I_can_see_the_parallel_publisher_collects_all_exceptions() + { + var sc = new ServiceCollection(); + sc.AddSingleton(); + sc.AddRequestDispatcher(builder => builder + .Add(typeof(FailingNotificationHandler)) + .Add(typeof(SecondTestNotificationHandler)) + .UseParallelNotificationPublisher()); + var provider = sc.BuildServiceProvider(); + var dispatcher = provider.GetRequiredService(); + var tracker = provider.GetRequiredService(); + + var ex = await Assert.ThrowsAsync(async () => + await dispatcher.PublishAsync(new TestNotification { Value = "Fail" })); + + await Assert.That(ex?.InnerExceptions).Count().IsEqualTo(1); + await Assert.That(ex?.InnerExceptions[0].Message).IsEquivalentTo("Handler failed"); + } +} diff --git a/src/request.dispatcher.tests/_fixtures/BlockingNotificationBehavior.cs b/src/request.dispatcher.tests/_fixtures/BlockingNotificationBehavior.cs new file mode 100644 index 0000000..f80b660 --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/BlockingNotificationBehavior.cs @@ -0,0 +1,13 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class BlockingNotificationBehavior(NotificationTestTracker tracker) : INotificationBehavior +{ + public Task HandleAsync(TestNotification notification, NotificationHandlerDelegate next, CancellationToken cancellationToken) + { + tracker.Log.Add("Blocked"); + return Task.CompletedTask; + } +} diff --git a/src/request.dispatcher.tests/_fixtures/CustomNotificationPublisher.cs b/src/request.dispatcher.tests/_fixtures/CustomNotificationPublisher.cs new file mode 100644 index 0000000..697970f --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/CustomNotificationPublisher.cs @@ -0,0 +1,17 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class CustomNotificationPublisher(NotificationTestTracker tracker) : INotificationPublisher +{ + public async Task PublishAsync(IEnumerable> handlers, TNotification notification, CancellationToken cancellationToken) where TNotification : INotification + { + tracker.Log.Add("CustomPublisher"); + + foreach (var handler in handlers) + { + await handler.HandleAsync(notification, cancellationToken); + } + } +} diff --git a/src/request.dispatcher.tests/_fixtures/FailingNotificationHandler.cs b/src/request.dispatcher.tests/_fixtures/FailingNotificationHandler.cs new file mode 100644 index 0000000..59b1c04 --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/FailingNotificationHandler.cs @@ -0,0 +1,12 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class FailingNotificationHandler : INotificationHandler +{ + public Task HandleAsync(TestNotification notification, CancellationToken cancellationToken) + { + throw new InvalidOperationException("Handler failed"); + } +} diff --git a/src/request.dispatcher.tests/_fixtures/NotificationChainedBehaviour1.cs b/src/request.dispatcher.tests/_fixtures/NotificationChainedBehaviour1.cs new file mode 100644 index 0000000..c97a79b --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/NotificationChainedBehaviour1.cs @@ -0,0 +1,13 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class NotificationChainedBehaviour1(NotificationTestTracker tracker) : INotificationBehavior +{ + public async Task HandleAsync(TestNotification notification, NotificationHandlerDelegate next, CancellationToken cancellationToken) + { + tracker.Log.Add("Behaviour1"); + await next(notification, cancellationToken); + } +} diff --git a/src/request.dispatcher.tests/_fixtures/NotificationChainedBehaviour2.cs b/src/request.dispatcher.tests/_fixtures/NotificationChainedBehaviour2.cs new file mode 100644 index 0000000..0ffeb82 --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/NotificationChainedBehaviour2.cs @@ -0,0 +1,13 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class NotificationChainedBehaviour2(NotificationTestTracker tracker) : INotificationBehavior +{ + public async Task HandleAsync(TestNotification notification, NotificationHandlerDelegate next, CancellationToken cancellationToken) + { + tracker.Log.Add("Behaviour2"); + await next(notification, cancellationToken); + } +} diff --git a/src/request.dispatcher.tests/_fixtures/NotificationTestTracker.cs b/src/request.dispatcher.tests/_fixtures/NotificationTestTracker.cs new file mode 100644 index 0000000..4f21e88 --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/NotificationTestTracker.cs @@ -0,0 +1,9 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class NotificationTestTracker +{ + public List Log { get; } = []; +} diff --git a/src/request.dispatcher.tests/_fixtures/OpenNotificationHandler.cs b/src/request.dispatcher.tests/_fixtures/OpenNotificationHandler.cs new file mode 100644 index 0000000..e31ebd3 --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/OpenNotificationHandler.cs @@ -0,0 +1,14 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class OpenNotificationHandler(NotificationTestTracker tracker) : INotificationHandler + where T : INotification +{ + public Task HandleAsync(T notification, CancellationToken cancellationToken) + { + tracker.Log.Add($"OpenHandler:{typeof(T).Name}"); + return Task.CompletedTask; + } +} diff --git a/src/request.dispatcher.tests/_fixtures/OtherNotificationPublisher.cs b/src/request.dispatcher.tests/_fixtures/OtherNotificationPublisher.cs new file mode 100644 index 0000000..d9eb024 --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/OtherNotificationPublisher.cs @@ -0,0 +1,17 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class OtherNotificationPublisher(NotificationTestTracker tracker) : INotificationPublisher +{ + public async Task PublishAsync(IEnumerable> handlers, TNotification notification, CancellationToken cancellationToken) where TNotification : INotification + { + tracker.Log.Add("OtherPublisher"); + + foreach (var handler in handlers) + { + await handler.HandleAsync(notification, cancellationToken); + } + } +} diff --git a/src/request.dispatcher.tests/_fixtures/SecondTestNotificationHandler.cs b/src/request.dispatcher.tests/_fixtures/SecondTestNotificationHandler.cs new file mode 100644 index 0000000..d23f86f --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/SecondTestNotificationHandler.cs @@ -0,0 +1,13 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class SecondTestNotificationHandler(NotificationTestTracker tracker) : INotificationHandler +{ + public Task HandleAsync(TestNotification notification, CancellationToken cancellationToken) + { + tracker.Log.Add($"SecondHandler:{notification.Value}"); + return Task.CompletedTask; + } +} diff --git a/src/request.dispatcher.tests/_fixtures/TestNotification.cs b/src/request.dispatcher.tests/_fixtures/TestNotification.cs new file mode 100644 index 0000000..0579949 --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/TestNotification.cs @@ -0,0 +1,9 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class TestNotification : INotification +{ + public string? Value { get; set; } +} diff --git a/src/request.dispatcher.tests/_fixtures/TestNotificationBehavior.cs b/src/request.dispatcher.tests/_fixtures/TestNotificationBehavior.cs new file mode 100644 index 0000000..0048910 --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/TestNotificationBehavior.cs @@ -0,0 +1,14 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class TestNotificationBehavior(NotificationTestTracker tracker) : INotificationBehavior +{ + public async Task HandleAsync(TestNotification notification, NotificationHandlerDelegate next, CancellationToken cancellationToken) + { + tracker.Log.Add("Before"); + await next(notification, cancellationToken); + tracker.Log.Add("After"); + } +} diff --git a/src/request.dispatcher.tests/_fixtures/TestNotificationHandler.cs b/src/request.dispatcher.tests/_fixtures/TestNotificationHandler.cs new file mode 100644 index 0000000..a4f416b --- /dev/null +++ b/src/request.dispatcher.tests/_fixtures/TestNotificationHandler.cs @@ -0,0 +1,13 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher.Tests; + +public class TestNotificationHandler(NotificationTestTracker tracker) : INotificationHandler +{ + public Task HandleAsync(TestNotification notification, CancellationToken cancellationToken) + { + tracker.Log.Add($"Handler:{notification.Value}"); + return Task.CompletedTask; + } +} diff --git a/src/request.dispatcher/INotification.cs b/src/request.dispatcher/INotification.cs new file mode 100644 index 0000000..d006880 --- /dev/null +++ b/src/request.dispatcher/INotification.cs @@ -0,0 +1,13 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher; + +/// +/// Represents a notification message that can be published to zero or more +/// handlers. +/// This interface serves as a marker for distinguishing notification types. +/// +public interface INotification +{ +} diff --git a/src/request.dispatcher/INotificationBehavior.cs b/src/request.dispatcher/INotificationBehavior.cs new file mode 100644 index 0000000..bf6afcd --- /dev/null +++ b/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; + +/// +/// Represents a behavior in the notification pipeline, allowing interception, modification, +/// or chaining of notification processing. +/// +/// The type of notification being processed. Must implement . +public interface INotificationBehavior where TNotification : INotification +{ + /// + /// Handles the asynchronous processing of a notification, allowing behavior customization + /// such as interception, modification, or chaining of the notification pipeline. + /// + /// The notification instance being processed. + /// The next delegate in the pipeline to execute after the custom behavior. + /// A token to monitor for cancellation requests. + /// A task representing the asynchronous operation. + Task HandleAsync(TNotification notification, NotificationHandlerDelegate next, CancellationToken cancellationToken); +} + +/// +/// Represents the delegate responsible for handling notifications in a pipeline. +/// +/// The type of notification being processed. Must implement . +/// The notification instance being processed. +/// A token to monitor for cancellation requests. +/// A task representing the asynchronous operation. +public delegate Task NotificationHandlerDelegate(TNotification notification, CancellationToken cancellationToken) where TNotification : INotification; diff --git a/src/request.dispatcher/INotificationHandler.cs b/src/request.dispatcher/INotificationHandler.cs new file mode 100644 index 0000000..7cac341 --- /dev/null +++ b/src/request.dispatcher/INotificationHandler.cs @@ -0,0 +1,19 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher; + +/// +/// Defines a handler for processing notifications of a specific type. +/// +/// The type of notification to handle. Must implement . +public interface INotificationHandler where TNotification : INotification +{ + /// + /// Handles the specified notification asynchronously. + /// + /// The notification object to process. + /// A token to monitor for cancellation requests. + /// A task representing the asynchronous operation. + Task HandleAsync(TNotification notification, CancellationToken cancellationToken); +} diff --git a/src/request.dispatcher/INotificationPublisher.cs b/src/request.dispatcher/INotificationPublisher.cs new file mode 100644 index 0000000..4cc9811 --- /dev/null +++ b/src/request.dispatcher/INotificationPublisher.cs @@ -0,0 +1,20 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Request.Dispatcher; + +/// +/// Defines a strategy for publishing notifications to registered handlers. +/// +public interface INotificationPublisher +{ + /// + /// Publishes the specified notification to the given collection of handlers. + /// + /// The type of notification being published. Must implement . + /// The list of handlers to invoke. + /// The notification instance to publish. + /// A token to monitor for cancellation requests. + /// A task representing the asynchronous publish operation. + Task PublishAsync(IEnumerable> handlers, TNotification notification, CancellationToken cancellationToken) where TNotification : INotification; +} diff --git a/src/request.dispatcher/IRequestDispatcher.cs b/src/request.dispatcher/IRequestDispatcher.cs index bd39e6f..af68ab7 100644 --- a/src/request.dispatcher/IRequestDispatcher.cs +++ b/src/request.dispatcher/IRequestDispatcher.cs @@ -25,4 +25,13 @@ public interface IRequestDispatcher /// Response type /// The created async enumerable, representing the stream of responses. IAsyncEnumerable DispatchAsync(IStreamRequest request, CancellationToken cancellationToken = default); + + /// + /// Asynchronously publish a notification to all registered handlers. + /// + /// Notification object + /// Optional custom notification publisher + /// Optional cancellation token + /// A task that represents the publish operation. + Task PublishAsync(INotification notification, INotificationPublisher? publisher = null, CancellationToken cancellationToken = default); } diff --git a/src/request.dispatcher/NotificationInvoker.cs b/src/request.dispatcher/NotificationInvoker.cs new file mode 100644 index 0000000..2104667 --- /dev/null +++ b/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 : 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>().Value; + + var handlers = options.GetRequestHandlers>(serviceProvider).ToList(); + + publisher ??= serviceProvider.GetRequiredService(); + + var behaviors = options.GetRequestBehaviors>(serviceProvider); + + var pipeline = behaviors + .Reverse() + .Aggregate((NotificationHandlerDelegate)Terminal, Chain); + + return pipeline(notification, cancellationToken); + + static NotificationHandlerDelegate Chain(NotificationHandlerDelegate next, INotificationBehavior behavior) + { + return [StackTraceHidden] (n, ct) => behavior.HandleAsync(n, next, ct); + } + + Task Terminal(TNotification n, CancellationToken ct) + { + return publisher.PublishAsync(handlers, n, ct); + } + } +} diff --git a/src/request.dispatcher/ParallelNotificationPublisher.cs b/src/request.dispatcher/ParallelNotificationPublisher.cs new file mode 100644 index 0000000..d94976c --- /dev/null +++ b/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(IEnumerable> handlers, TNotification notification, CancellationToken cancellationToken) + where TNotification : INotification + { + var exceptions = new List(); + 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); + } + } +} diff --git a/src/request.dispatcher/RequestDispatcher.cs b/src/request.dispatcher/RequestDispatcher.cs index 2acb426..0b77bda 100644 --- a/src/request.dispatcher/RequestDispatcher.cs +++ b/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 NotificationHandlers = new(); private readonly IServiceProvider _serviceProvider; @@ -24,6 +25,7 @@ internal sealed class RequestDispatcher : IRequestDispatcher { ScalarRequestHandlers.Clear(); StreamRequestHandlers.Clear(); + NotificationHandlers.Clear(); } public Task DispatchAsync(IScalarRequest request, CancellationToken cancellationToken = default) @@ -51,4 +53,17 @@ internal sealed class RequestDispatcher : IRequestDispatcher return ((StreamRequestInvoker)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); + } } diff --git a/src/request.dispatcher/RequestDispatcherBuilderExtensions.cs b/src/request.dispatcher/RequestDispatcherBuilderExtensions.cs index 8d53e3b..ee93acf 100644 --- a/src/request.dispatcher/RequestDispatcherBuilderExtensions.cs +++ b/src/request.dispatcher/RequestDispatcherBuilderExtensions.cs @@ -150,6 +150,70 @@ public static class RequestDispatcherBuilderExtensions return builder; } + /// + /// Registers as the default + /// in the dependency injection container. + /// + /// The type implementing to register. + /// The to configure. + /// The lifetime with which the publisher is registered (default ). + /// The instance for further configuration. + public static IRequestDispatcherBuilder UseNotificationPublisher(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; + } + + /// + /// Registers the as the default + /// . + /// + /// The to configure. + /// The instance for further configuration. + public static IRequestDispatcherBuilder UseSequentialNotificationPublisher(this IRequestDispatcherBuilder builder) + { + ArgumentNullException.ThrowIfNull(builder); + + builder.Services.Replace(new ServiceDescriptor(typeof(INotificationPublisher), new SequentialNotificationPublisher())); + + return builder; + } + + /// + /// Registers the as the default + /// using the specified maximum degree of parallelism. + /// + /// The to configure. + /// The instance for further configuration. + public static IRequestDispatcherBuilder UseParallelNotificationPublisher(this IRequestDispatcherBuilder builder) + { + ArgumentNullException.ThrowIfNull(builder); + + builder.Services.Replace(new ServiceDescriptor(typeof(INotificationPublisher), new ParallelNotificationPublisher())); + + return builder; + } + + /// + /// Registers the as the default + /// using the specified maximum degree of parallelism. + /// + /// The to configure. + /// The maximum number of handlers to execute concurrently. + /// The instance for further configuration. + 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 ValidateNoNestedRequestHandlers(IReadOnlyCollection types, [CallerMemberName] string? invoker = null) { var nestedHandlers = types diff --git a/src/request.dispatcher/RequestDispatcherOptions.cs b/src/request.dispatcher/RequestDispatcherOptions.cs index d18def9..d4efd69 100644 --- a/src/request.dispatcher/RequestDispatcherOptions.cs +++ b/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 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 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 : IRequestBehaviour - 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 : IRequestBehaviour, TResponse> + var requestType = @interface.GetGenericArguments()[0]; var impl = behaviour.MakeGenericType(requestType.GetGenericArguments()); if (impl.IsAssignableTo(@interface)) { diff --git a/src/request.dispatcher/SequentialNotificationPublisher.cs b/src/request.dispatcher/SequentialNotificationPublisher.cs new file mode 100644 index 0000000..d32054a --- /dev/null +++ b/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(IEnumerable> handlers, TNotification notification, CancellationToken cancellationToken) + where TNotification : INotification + { + foreach (var handler in handlers) + { + await handler.HandleAsync(notification, cancellationToken); + } + } +}