diff --git a/src/Core/src/Eventuous.Subscriptions/Filters/ConsumePipe.cs b/src/Core/src/Eventuous.Subscriptions/Filters/ConsumePipe.cs index 892548f6..fbe54a5a 100644 --- a/src/Core/src/Eventuous.Subscriptions/Filters/ConsumePipe.cs +++ b/src/Core/src/Eventuous.Subscriptions/Filters/ConsumePipe.cs @@ -7,9 +7,25 @@ namespace Eventuous.Subscriptions.Filters; public sealed class ConsumePipe : IAsyncDisposable { readonly LinkedList _filters = []; + readonly List _owned = []; + + /// + /// Completed when the disposal that won is done, so callers that lost the race + /// wait for the teardown instead of being told it finished. + /// + readonly TaskCompletionSource _disposed = new(TaskCreationOptions.RunContinuationsAsynchronously); + + int _disposing; public IEnumerable RegisteredFilters => _filters.AsEnumerable(); + /// + /// Gives the pipe ownership of a component, so it gets disposed with the pipe. Used for event handlers + /// created by the subscription builder, as nothing else would dispose them. + /// + /// Component to dispose with the pipe + internal void AddOwned(object component) => _owned.Add(component); + public ConsumePipe AddFilterFirst(IConsumeFilter filter) where TIn : class, IBaseConsumeContext where TOut : class, IBaseConsumeContext { @@ -50,11 +66,46 @@ public ConsumePipe AddFilterLast(IConsumeFilter filter) static ValueTask Move(LinkedListNode? node, IBaseConsumeContext context) => node == null ? default : node.Value.Send(context, node.Next); + /// + /// Disposes the filters and everything the pipe owns, exactly once. Concurrent callers, which the SignalR + /// gateway produces when a disconnect races the gateway shutdown, all wait for the single teardown. + /// public async ValueTask DisposeAsync() { - foreach (var filter in _filters) { - if (filter is IAsyncDisposable d) { - await d.DisposeAsync().NoContext(); + // Exchange, not check-then-set: two callers reading a plain flag both pass and double-dispose. + if (Interlocked.Exchange(ref _disposing, 1) != 0) { + await _disposed.Task.NoContext(); + + return; + } + + try { + foreach (var filter in _filters) { + if (filter is IAsyncDisposable d) { + await d.DisposeAsync().NoContext(); + } } + + // After the filters, as they drain in-flight messages that still need their handlers, and in + // reverse order of creation. + for (var i = _owned.Count - 1; i >= 0; i--) { + switch (_owned[i]) { + case IAsyncDisposable d: + await d.DisposeAsync().NoContext(); + + break; + case IDisposable d: + d.Dispose(); + + break; + } + } + + _disposed.TrySetResult(); + } catch (Exception e) { + // The losing callers get the same failure rather than a teardown that looks clean. + _disposed.TrySetException(e); + + throw; } } } diff --git a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionBuilder.cs b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionBuilder.cs index e3379b96..1e565138 100644 --- a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionBuilder.cs +++ b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionBuilder.cs @@ -19,7 +19,9 @@ public abstract class SubscriptionBuilder(IServiceCollection services, string su public string SubscriptionId { get; } = subscriptionId; public IServiceCollection Services { get; } = services; - readonly List _handlers = []; + readonly List _handlers = []; + readonly HashSet _handlerTypes = []; + readonly List _ownedHandlers = []; protected ConsumePipe Pipe { get; } = new(); protected ResolveConsumer ResolveConsumer { get; set; } = null!; @@ -27,11 +29,14 @@ public abstract class SubscriptionBuilder(IServiceCollection services, string su protected IEventHandler[] ResolveHandlers(IServiceProvider sp) => [.. _handlers.Select(x => x(sp))]; /// - /// Adds an event handler to the subscription + /// Adds an event handler to the subscription. The handler is registered in the container, keyed by + /// , so it can only be added once per subscription. /// /// Event handler type /// + /// The same handler type is already registered for this subscription public SubscriptionBuilder AddEventHandler<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] THandler>() where THandler : class, IEventHandler { + ReserveHandlerType(); Services.TryAddKeyedSingleton(SubscriptionId); AddHandlerResolve(sp => sp.GetRequiredKeyedService(SubscriptionId)); @@ -39,14 +44,34 @@ public abstract class SubscriptionBuilder(IServiceCollection services, string su } /// - /// Adds an event handler to the subscription + /// Adds an event handler to the subscription. The handler is created once by the given function and owned by + /// the subscription, it isn't registered in the container, so it gets disposed when the subscription is + /// disposed if it implements or . When the function + /// returns a handler owned elsewhere, use the overload with ownsHandler to decline ownership. /// /// A function to resolve event handler using the service provider /// Event handler type /// - public SubscriptionBuilder AddEventHandler(Func getHandler) where THandler : class, IEventHandler { - Services.TryAddKeyedSingleton(SubscriptionId, (sp, _) => getHandler(sp)); - AddHandlerResolve(sp => sp.GetRequiredKeyedService(SubscriptionId)); + public SubscriptionBuilder AddEventHandler(Func getHandler) where THandler : class, IEventHandler + => AddEventHandler(getHandler, true); + + /// + /// Adds an event handler to the subscription. The handler is created once by the given function and kept by + /// the subscription, it isn't registered in the container. + /// + /// A function to resolve event handler using the service provider + /// + /// When true, the default, the subscription owns the handler and disposes it when the subscription is + /// disposed, if it implements or . Set it to + /// false when the function returns a handler owned elsewhere, such as one it resolves from the + /// container, as disposing that would break the other components using it. To have the container create and + /// own the handler, use instead. + /// + /// Event handler type + /// + public SubscriptionBuilder AddEventHandler(Func getHandler, bool ownsHandler) where THandler : class, IEventHandler { + THandler? handler = null; + AddHandlerResolve(sp => handler ??= Own(getHandler(sp), ownsHandler)); return this; } @@ -73,10 +98,12 @@ public SubscriptionBuilder AddEventHandler(THandler handler) where THa /// Wrapping event handler type produced by the factory /// Factory that takes the resolved inner handler and returns the wrapping handler /// The current instance + /// The same inner handler type is already registered for this subscription public SubscriptionBuilder AddCompositionEventHandler<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] THandler, TWrappingHandler>( Func getWrappingHandler ) where THandler : class, IEventHandler where TWrappingHandler : class, IEventHandler { + ReserveHandlerType(); Services.TryAddKeyedSingleton(SubscriptionId); AddHandlerResolve(sp => getWrappingHandler(sp.GetRequiredKeyedService(SubscriptionId))); @@ -87,6 +114,11 @@ Func getWrappingHandler /// Adds a composition event handler to the subscription with a custom inner handler resolver. /// The inner handler is created via and then wrapped into /// using . + /// The inner handler is created once and owned by the subscription, it isn't registered in the container, so it + /// gets disposed when the subscription is disposed if it implements or + /// . When returns a handler owned elsewhere, + /// use the overload with ownsInnerHandler to decline ownership. The wrapping handler decorates the inner + /// one and is never disposed. /// /// Inner event handler type /// Wrapping event handler type @@ -96,9 +128,34 @@ Func getWrappingHandler public SubscriptionBuilder AddCompositionEventHandler( Func getInnerHandler, Func getWrappingHandler + ) where THandler : class, IEventHandler where TWrappingHandler : class, IEventHandler + => AddCompositionEventHandler(getInnerHandler, getWrappingHandler, true); + + /// + /// Adds a composition event handler to the subscription with a custom inner handler resolver. + /// The inner handler is created via and then wrapped into + /// using . + /// The inner handler is created once and kept by the subscription, it isn't registered in the container. + /// The wrapping handler decorates the inner one and is never disposed. + /// + /// Inner event handler type + /// Wrapping event handler type + /// Function that resolves or creates the inner handler using the service provider + /// Factory that produces the wrapping handler from the inner handler + /// + /// When true, the default, the subscription owns the inner handler and disposes it when the subscription + /// is disposed, if it implements or . Set it to + /// false when returns a handler owned elsewhere, such as one it + /// resolves from the container, as disposing that would break the other components using it. + /// + /// The current instance + public SubscriptionBuilder AddCompositionEventHandler( + Func getInnerHandler, + Func getWrappingHandler, + bool ownsInnerHandler ) where THandler : class, IEventHandler where TWrappingHandler : class, IEventHandler { - Services.TryAddKeyedSingleton(SubscriptionId, (sp, _) => getInnerHandler(sp)); - AddHandlerResolve(sp => getWrappingHandler(sp.GetRequiredKeyedService(SubscriptionId))); + THandler? innerHandler = null; + AddHandlerResolve(sp => getWrappingHandler(innerHandler ??= Own(getInnerHandler(sp), ownsInnerHandler))); return this; } @@ -107,6 +164,11 @@ Func getWrappingHandler /// Adds a composition event handler to the subscription with a custom inner handler resolver. /// The inner handler is created via and then wrapped into /// using . + /// The inner handler is created once and owned by the subscription, it isn't registered in the container, so it + /// gets disposed when the subscription is disposed if it implements or + /// . When returns a handler owned elsewhere, + /// use the overload with ownsInnerHandler to decline ownership. The wrapping handler decorates the inner + /// one and is never disposed. /// /// Inner event handler type /// Wrapping event handler type @@ -116,9 +178,34 @@ Func getWrappingHandler public SubscriptionBuilder AddCompositionEventHandler( Func getInnerHandler, Func getWrappingHandler + ) where THandler : class, IEventHandler where TWrappingHandler : class, IEventHandler + => AddCompositionEventHandler(getInnerHandler, getWrappingHandler, true); + + /// + /// Adds a composition event handler to the subscription with a custom inner handler resolver. + /// The inner handler is created via and then wrapped into + /// using . + /// The inner handler is created once and kept by the subscription, it isn't registered in the container. + /// The wrapping handler decorates the inner one and is never disposed. + /// + /// Inner event handler type + /// Wrapping event handler type + /// Function that resolves or creates the inner handler using the service provider + /// Factory that produces the wrapping handler from the inner handler + /// + /// When true, the default, the subscription owns the inner handler and disposes it when the subscription + /// is disposed, if it implements or . Set it to + /// false when returns a handler owned elsewhere, such as one it + /// resolves from the container, as disposing that would break the other components using it. + /// + /// The current instance + public SubscriptionBuilder AddCompositionEventHandler( + Func getInnerHandler, + Func getWrappingHandler, + bool ownsInnerHandler ) where THandler : class, IEventHandler where TWrappingHandler : class, IEventHandler { - Services.TryAddKeyedSingleton(SubscriptionId, (sp, _) => getInnerHandler(sp)); - AddHandlerResolve(sp => getWrappingHandler(sp.GetRequiredKeyedService(SubscriptionId), sp)); + THandler? innerHandler = null; + AddHandlerResolve(sp => getWrappingHandler(innerHandler ??= Own(getInnerHandler(sp), ownsInnerHandler), sp)); return this; } @@ -167,6 +254,42 @@ public SubscriptionBuilder AddConsumeFilterFirst(IConsumeFilter + /// Records a handler the subscription owns, so it gets disposed with the subscription. Ownership is stated by + /// the caller rather than inferred: a handler factory is free to return a handler the container owns, and + /// disposing that would break the other components using it. + /// + THandler Own(THandler handler, bool owns) where THandler : class, IEventHandler { + if (owns && handler is IDisposable or IAsyncDisposable) _ownedHandlers.Add(handler); + + return handler; + } + + /// + /// Hands the handlers created by the builder over to the pipe, which disposes them when the subscription + /// is disposed. + /// + protected void TransferHandlersOwnership() { + foreach (var handler in _ownedHandlers) { + Pipe.AddOwned(handler); + } + + _ownedHandlers.Clear(); + } + + /// + /// Claims the container slot keyed by for the given handler type. Two handlers of + /// the same type would share that slot, so the subscription would silently dispatch the same instance twice. + /// + void ReserveHandlerType() where THandler : class, IEventHandler { + if (!_handlerTypes.Add(typeof(THandler))) { + throw new ArgumentException( + $"Event handler {typeof(THandler).Name} is already registered for subscription {SubscriptionId}. " + + "Use the overload with a handler factory or instance to add several handlers of the same type." + ); + } + } + void AddHandlerResolve(ResolveHandler resolveHandler) => _handlers.Add(sp => { var handler = resolveHandler(sp); @@ -242,6 +365,7 @@ public T ResolveSubscription(IServiceProvider sp) { } var consumer = GetConsumer(sp); + TransferHandlersOwnership(); if (EventuousDiagnostics.Enabled) { Pipe.AddFilterLast(new TracingFilter(consumer.GetType().Name)); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/CompositionHandlerTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/CompositionHandlerTests.cs index 494197ef..53e8f857 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions/CompositionHandlerTests.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions/CompositionHandlerTests.cs @@ -38,11 +38,12 @@ public async Task Teardown() { [Test] public async Task ShouldResolveCompositionHandlerWithFactory() { - // This test validates that AddCompositionEventHandler correctly registers - // handlers when using a factory function - var handler = _server.Services.GetRequiredKeyedService("sub-with-factory"); - await Assert.That(handler).IsNotNull(); - await Assert.That(handler.Dependency.Value).IsEqualTo("test-value"); + // This test validates that AddCompositionEventHandler builds the inner handler from the factory, + // giving it access to the service provider. The handler is owned by the builder, not the container. + var capture = _server.Services.GetRequiredService(); + + await Assert.That(capture.Handler).IsNotNull(); + await Assert.That(capture.Handler!.Dependency.Value).IsEqualTo("test-value"); } [Test] @@ -82,12 +83,13 @@ class Startup { public static void ConfigureServices(IServiceCollection services) { services.AddSingleton(new TestHandlerLogger()); services.AddSingleton(); + services.AddSingleton(); // Test the AddCompositionEventHandler with a factory function services.AddSubscription( "sub-with-factory", builder => builder.AddCompositionEventHandler( - sp => new(sp.GetRequiredService(), sp.GetRequiredService()), + sp => sp.GetRequiredService().Handler = new(sp.GetRequiredService(), sp.GetRequiredService()), (handler, sp) => new(handler, sp.GetRequiredService()) ) ); @@ -103,6 +105,14 @@ class TestSub(TestOptions options, ConsumePipe consumePipe) protected override ValueTask Connect(SubscriptionRun run) => default; } + /// + /// Keeps the handler the factory created, as it's owned by the subscription builder and not resolvable + /// from the container. + /// + class HandlerCapture { + public TestHandler? Handler { get; set; } + } + public class TestDependency { #pragma warning disable CA1822 public string Value => "test-value"; diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/ConsumePipeTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/ConsumePipeTests.cs index 290b2c41..7f6d0e28 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions/ConsumePipeTests.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions/ConsumePipeTests.cs @@ -35,6 +35,45 @@ public async Task ShouldAddContextBaggage() { await Assert.That(handler.Received!.Items.GetItem(Key)).IsEqualTo(baggage); } + [Test] + public async Task ShouldMakeSecondDisposalWaitForTheFirst() { + var filter = new BlockingFilter(); + var pipe = new ConsumePipe().AddFilterFirst(filter); + + var first = pipe.DisposeAsync(); + var second = pipe.DisposeAsync(); + + // The pipe is still disposing, so a second caller must not be told the teardown is done + await Assert.That(second.IsCompleted).IsFalse(); + + filter.Release(); + await first; + await second; + + await Assert.That(filter.Disposals).IsEqualTo(1); + } + + /// + /// Blocks inside until released, so a second disposal can be observed while the + /// first one is still running. + /// + class BlockingFilter : ConsumeFilter, IAsyncDisposable { + readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public int Disposals { get; private set; } + + public void Release() => _release.TrySetResult(); + + protected override ValueTask Send(IMessageConsumeContext context, LinkedListNode? next) + => next == null ? default : next.Value.Send(context, next.Next); + + public async ValueTask DisposeAsync() { + Disposals++; + + await _release.Task; + } + } + class TestFilter(string key, string payload) : ConsumeFilter { protected override ValueTask Send(IMessageConsumeContext context, LinkedListNode? next) { context.Items.AddItem(key, payload); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/HandlerDisposalTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/HandlerDisposalTests.cs new file mode 100644 index 00000000..e5dd9db6 --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/HandlerDisposalTests.cs @@ -0,0 +1,239 @@ +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Context; +using Eventuous.Subscriptions.Filters; +using Eventuous.Subscriptions.Registrations; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Eventuous.Tests.Subscriptions; + +/// +/// The subscription disposes the handlers its builder created. Handlers owned by the container or supplied by +/// the caller are left alone. +/// +public class HandlerDisposalTests { + const string SubscriptionId = "handler-disposal"; + + [Test] + public async Task ShouldDisposeHandlerCreatedByFactory() { + DisposableHandler? handler = null; + + var resolved = Resolve(builder => builder.AddEventHandler(_ => handler = new(), ownsHandler: true)); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(handler!.Disposals).IsEqualTo(1); + } + + [Test] + public async Task ShouldDisposeAsyncHandlerCreatedByFactory() { + AsyncDisposableHandler? handler = null; + + var resolved = Resolve(builder => builder.AddEventHandler(_ => handler = new(), ownsHandler: true)); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(handler!.Disposals).IsEqualTo(1); + } + + [Test] + public async Task ShouldPreferAsyncDisposal() { + DoublyDisposableHandler? handler = null; + + var resolved = Resolve(builder => builder.AddEventHandler(_ => handler = new(), ownsHandler: true)); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(handler!.AsyncDisposals).IsEqualTo(1); + await Assert.That(handler.Disposals).IsEqualTo(0); + } + + [Test] + public async Task ShouldDisposeInnerCompositionHandlerCreatedByFactory() { + DisposableHandler? inner = null; + DisposableWrappingHandler? wrapper = null; + + var resolved = Resolve( + builder => builder.AddCompositionEventHandler( + _ => inner = new(), + handler => wrapper = new(handler) + ) + ); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(inner!.Disposals).IsEqualTo(1); + // The wrapping handler decorates the inner one, it holds nothing of its own + await Assert.That(wrapper!.Disposals).IsEqualTo(0); + } + + [Test] + public async Task ShouldNotDisposeCompositionOfContainerOwnedInnerHandler() { + DisposableWrappingHandler? wrapper = null; + + var resolved = Resolve( + builder => builder.AddCompositionEventHandler(handler => wrapper = new(handler)) + ); + var inner = resolved.Provider.GetRequiredKeyedService(SubscriptionId); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(wrapper!.Disposals).IsEqualTo(0); + await Assert.That(inner.Disposals).IsEqualTo(0); + } + + [Test] + public async Task ShouldNotDisposeHandlerOwnedByContainer() { + var resolved = Resolve(builder => builder.AddEventHandler()); + var handler = resolved.Provider.GetRequiredKeyedService(SubscriptionId); + + await resolved.Subscription.DisposeAsync(); + + await Assert.That(handler.Disposals).IsEqualTo(0); + } + + [Test] + public async Task ShouldDisposeFactoryHandlerByDefault() { + DisposableHandler? handler = null; + + var resolved = Resolve(builder => builder.AddEventHandler(_ => handler = new())); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(handler!.Disposals).IsEqualTo(1); + } + + [Test] + public async Task ShouldNotDisposeHandlerWhenOwnershipIsDeclined() { + // A factory is free to return a handler owned elsewhere, which is what declining ownership is for + var resolved = Resolve( + builder => builder.AddEventHandler(sp => sp.GetRequiredService(), ownsHandler: false), + services => services.AddSingleton() + ); + var handler = resolved.Provider.GetRequiredService(); + + await resolved.Subscription.DisposeAsync(); + + await Assert.That(handler.Disposals).IsEqualTo(0); + } + + [Test] + public async Task ShouldNotDisposeCompositionInnerHandlerWhenOwnershipIsDeclined() { + var resolved = Resolve( + builder => builder.AddCompositionEventHandler( + sp => sp.GetRequiredService(), + handler => new(handler), + ownsInnerHandler: false + ), + services => services.AddSingleton() + ); + var inner = resolved.Provider.GetRequiredService(); + + await resolved.Subscription.DisposeAsync(); + + await Assert.That(inner.Disposals).IsEqualTo(0); + } + + [Test] + public async Task ShouldNotDisposeHandlerSuppliedByCaller() { + var handler = new DisposableHandler(); + + var resolved = Resolve(builder => builder.AddEventHandler(handler)); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(handler.Disposals).IsEqualTo(0); + } + + [Test] + public async Task ShouldDisposeHandlerOnlyOnce() { + DisposableHandler? handler = null; + + var resolved = Resolve(builder => builder.AddEventHandler(_ => handler = new(), ownsHandler: true)); + await resolved.Subscription.DisposeAsync(); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(handler!.Disposals).IsEqualTo(1); + } + + [Test] + public async Task ShouldDisposeFiltersBeforeHandlers() { + List order = []; + + var resolved = Resolve( + builder => builder + .AddConsumeFilterFirst(new RecordingFilter(order)) + .AddEventHandler(_ => new RecordingHandler(order), ownsHandler: true) + ); + await resolved.Subscription.DisposeAsync(); + + await Assert.That(order).IsEquivalentTo(["filter", "handler"]); + } + + static Resolved Resolve(Action> configure, Action? configureServices = null) { + var services = new ServiceCollection(); + configureServices?.Invoke(services); + services.AddSubscription(SubscriptionId, configure); + + var provider = services.BuildServiceProvider(); + + return new(provider, provider.GetRequiredService()); + } + + record Resolved(ServiceProvider Provider, TestSub Subscription); + + record TestOptions : SubscriptionOptions; + + class TestSub(TestOptions options, ConsumePipe consumePipe) + : EventSubscription(options, consumePipe, NullLoggerFactory.Instance, null) { + protected override ValueTask Connect(SubscriptionRun run) => default; + } + + class SucceedingHandler : BaseEventHandler { + public override ValueTask HandleEvent(IMessageConsumeContext context) => ValueTask.FromResult(EventHandlingStatus.Success); + } + + sealed class DisposableHandler : SucceedingHandler, IDisposable { + public int Disposals { get; private set; } + + public void Dispose() => Disposals++; + } + + sealed class AsyncDisposableHandler : SucceedingHandler, IAsyncDisposable { + public int Disposals { get; private set; } + + public ValueTask DisposeAsync() { + Disposals++; + + return default; + } + } + + sealed class DoublyDisposableHandler : SucceedingHandler, IDisposable, IAsyncDisposable { + public int Disposals { get; private set; } + public int AsyncDisposals { get; private set; } + + public void Dispose() => Disposals++; + + public ValueTask DisposeAsync() { + AsyncDisposals++; + + return default; + } + } + + sealed class DisposableWrappingHandler(IEventHandler inner) : SucceedingHandler, IDisposable { + public IEventHandler Inner { get; } = inner; + public int Disposals { get; private set; } + + public void Dispose() => Disposals++; + } + + sealed class RecordingHandler(List order) : SucceedingHandler, IDisposable { + public void Dispose() => order.Add("handler"); + } + + sealed class RecordingFilter(List order) : ConsumeFilter, IAsyncDisposable { + protected override ValueTask Send(IMessageConsumeContext context, LinkedListNode? next) + => next == null ? default : next.Value.Send(context, next.Next); + + public ValueTask DisposeAsync() { + order.Add("filter"); + + return default; + } + } +} diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/HandlerRegistrationTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/HandlerRegistrationTests.cs new file mode 100644 index 00000000..9c21d0d0 --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/HandlerRegistrationTests.cs @@ -0,0 +1,165 @@ +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Context; +using Eventuous.Subscriptions.Filters; +using Eventuous.Subscriptions.Registrations; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Eventuous.Tests.Subscriptions; + +/// +/// Handler registrations that share an inferred handler type must not collapse into one. +/// +public class HandlerRegistrationTests { + const string SubscriptionId = "handler-registration"; + + [Test] + public async Task ShouldRunAllHandlersRegisteredByFactory() { + List handled = []; + + IReadOnlyList> factories = [ + _ => new FirstHandler(handled), + _ => new SecondHandler(handled), + _ => new ThirdHandler(handled) + ]; + + await Handle(builder => { + // THandler infers as IEventHandler for every call + foreach (var factory in factories) builder.AddEventHandler(sp => factory(sp)); + } + ); + + await Assert.That(handled).IsEquivalentTo([typeof(FirstHandler), typeof(SecondHandler), typeof(ThirdHandler)]); + } + + [Test] + public async Task ShouldRunAllCompositionHandlersRegisteredByFactory() { + List handled = []; + + await Handle(builder => builder + .AddCompositionEventHandler(_ => new FirstHandler(handled), inner => new(inner)) + .AddCompositionEventHandler(_ => new SecondHandler(handled), inner => new(inner)) + ); + + await Assert.That(handled).IsEquivalentTo([typeof(FirstHandler), typeof(SecondHandler)]); + } + + [Test] + public async Task ShouldRunAllCompositionHandlersRegisteredByFactoryWithProvider() { + List handled = []; + + await Handle(builder => builder + .AddCompositionEventHandler(_ => new FirstHandler(handled), (inner, _) => new(inner)) + .AddCompositionEventHandler(_ => new SecondHandler(handled), (inner, _) => new(inner)) + ); + + await Assert.That(handled).IsEquivalentTo([typeof(FirstHandler), typeof(SecondHandler)]); + } + + [Test] + public async Task ShouldCreateFactoryHandlerOnlyOnce() { + List handled = []; + var created = 0; + + var services = new ServiceCollection(); + var builder = new TestBuilder(services, SubscriptionId); + + builder.AddEventHandler( + _ => { + created++; + + return new FirstHandler(handled); + } + ); + + var provider = services.BuildServiceProvider(); + builder.Resolve(provider); + builder.Resolve(provider); + + await Assert.That(created).IsEqualTo(1); + } + + [Test] + public async Task ShouldThrowWhenSameHandlerTypeRegisteredTwice() { + var services = new ServiceCollection(); + + await Assert.That( + () => _ = services.AddSubscription( + SubscriptionId, + builder => builder.AddEventHandler().AddEventHandler() + ) + ) + .Throws(); + } + + [Test] + public async Task ShouldThrowWhenCompositionInnerHandlerTypeIsAlreadyRegistered() { + var services = new ServiceCollection(); + + await Assert.That( + () => _ = services.AddSubscription( + SubscriptionId, + builder => builder + .AddEventHandler() + .AddCompositionEventHandler(inner => new(inner)) + ) + ) + .Throws(); + } + + [Test] + public async Task ShouldAllowSameHandlerTypeInDifferentSubscriptions() { + List handled = []; + + var services = new ServiceCollection(); + services.AddSingleton(handled); + services.AddSubscription("sub1", builder => builder.AddEventHandler()); + services.AddSubscription("sub2", builder => builder.AddEventHandler()); + + await using var provider = services.BuildServiceProvider(); + + await Assert.That(provider.GetServices().ToArray()).HasCount(2); + } + + static async Task Handle(Action> configure) { + var services = new ServiceCollection(); + services.AddSubscription(SubscriptionId, configure); + + await using var provider = services.BuildServiceProvider(); + + var subscription = provider.GetRequiredService(); + + await subscription.Pipe.Send(TestContext.CreateContext()); + } + + record TestOptions : SubscriptionOptions; + + class TestSub(TestOptions options, ConsumePipe consumePipe) + : EventSubscription(options, consumePipe, NullLoggerFactory.Instance, null) { + protected override ValueTask Connect(SubscriptionRun run) => default; + } + + sealed class TestBuilder(IServiceCollection services, string subscriptionId) : SubscriptionBuilder(services, subscriptionId) { + public IEventHandler[] Resolve(IServiceProvider sp) => ResolveHandlers(sp); + } + + abstract class RecordingHandler(List handled) : BaseEventHandler { + public override ValueTask HandleEvent(IMessageConsumeContext context) { + handled.Add(GetType()); + + return ValueTask.FromResult(EventHandlingStatus.Success); + } + } + + sealed class FirstHandler(List handled) : RecordingHandler(handled); + + sealed class SecondHandler(List handled) : RecordingHandler(handled); + + sealed class ThirdHandler(List handled) : RecordingHandler(handled); + + sealed class WrappingHandler(IEventHandler inner) : BaseEventHandler { + public override ValueTask HandleEvent(IMessageConsumeContext context) => inner.HandleEvent(context); + } + + sealed class PlainHandler(List handled) : RecordingHandler(handled); +}