diff --git a/src/Compat/IsExternalInit.cs b/src/Compat/IsExternalInit.cs deleted file mode 100644 index bf44d73..0000000 --- a/src/Compat/IsExternalInit.cs +++ /dev/null @@ -1,5 +0,0 @@ -// Init accessors compile against this marker type; netstandard2.0 and net462 do not ship it. -namespace System.Runtime.CompilerServices; - -internal static class IsExternalInit -{ } diff --git a/src/Directory.Build.props b/src/Directory.Build.props index 58b969c..730951f 100644 --- a/src/Directory.Build.props +++ b/src/Directory.Build.props @@ -36,8 +36,8 @@ - - + + diff --git a/src/RequestFlow.Cqrs/CqrsDispatcher.cs b/src/RequestFlow.Cqrs/CqrsDispatcher.cs index 7ae9b30..83c24b9 100644 --- a/src/RequestFlow.Cqrs/CqrsDispatcher.cs +++ b/src/RequestFlow.Cqrs/CqrsDispatcher.cs @@ -14,8 +14,7 @@ internal sealed class CqrsDispatcher(IRequestDispatcher dispatcher) : ICommandDi /// public Task SendAsync(ICommand command, CancellationToken cancellationToken = default) { - if (command is null) - throw new ArgumentNullException(nameof(command)); + ThrowHelper.ThrowIfNull(command); return _dispatcher.SendAsync(command, cancellationToken); } @@ -23,8 +22,7 @@ public Task SendAsync(ICommand command, Cancell /// public Task SendAsync(ICommand command, CancellationToken cancellationToken = default) { - if (command is null) - throw new ArgumentNullException(nameof(command)); + ThrowHelper.ThrowIfNull(command); return _dispatcher.SendAsync(command, cancellationToken); } @@ -32,8 +30,7 @@ public Task SendAsync(ICommand command, CancellationToken cancellationToken = de /// public Task SendAsync(IQuery query, CancellationToken cancellationToken = default) { - if (query is null) - throw new ArgumentNullException(nameof(query)); + ThrowHelper.ThrowIfNull(query); return _dispatcher.SendAsync(query, cancellationToken); } diff --git a/src/RequestFlow.Cqrs/CqrsStreamDispatcher.cs b/src/RequestFlow.Cqrs/CqrsStreamDispatcher.cs index 3734a0e..295aa6f 100644 --- a/src/RequestFlow.Cqrs/CqrsStreamDispatcher.cs +++ b/src/RequestFlow.Cqrs/CqrsStreamDispatcher.cs @@ -14,8 +14,7 @@ internal sealed class CqrsStreamDispatcher(IStreamDispatcher dispatcher) : IStre /// public IAsyncEnumerable Stream(IStreamQuery query, CancellationToken cancellationToken = default) { - if (query is null) - throw new ArgumentNullException(nameof(query)); + ThrowHelper.ThrowIfNull(query); return _dispatcher.Stream(query, cancellationToken); } diff --git a/src/RequestFlow.Cqrs/CqrsValueDispatcher.cs b/src/RequestFlow.Cqrs/CqrsValueDispatcher.cs index 1979590..5fa3c34 100644 --- a/src/RequestFlow.Cqrs/CqrsValueDispatcher.cs +++ b/src/RequestFlow.Cqrs/CqrsValueDispatcher.cs @@ -15,8 +15,7 @@ public ValueTask SendAsync( IValueCommand command, CancellationToken cancellationToken = default) { - if (command is null) - throw new ArgumentNullException(nameof(command)); + ThrowHelper.ThrowIfNull(command); return _dispatcher.SendAsync(command, cancellationToken); } @@ -26,8 +25,7 @@ public ValueTask SendAsync( IValueCommand command, CancellationToken cancellationToken = default) { - if (command is null) - throw new ArgumentNullException(nameof(command)); + ThrowHelper.ThrowIfNull(command); return _dispatcher.SendAsync(command, cancellationToken); } @@ -37,8 +35,7 @@ public ValueTask SendAsync( IValueQuery query, CancellationToken cancellationToken = default) { - if (query is null) - throw new ArgumentNullException(nameof(query)); + ThrowHelper.ThrowIfNull(query); return _dispatcher.SendAsync(query, cancellationToken); } diff --git a/src/RequestFlow.Cqrs/ThrowHelper.cs b/src/RequestFlow.Cqrs/ThrowHelper.cs new file mode 100644 index 0000000..8fbf3d1 --- /dev/null +++ b/src/RequestFlow.Cqrs/ThrowHelper.cs @@ -0,0 +1,21 @@ +using System; +using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; + +namespace RequestFlow.Cqrs; + +internal static class ThrowHelper +{ + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static void ThrowIfNull( + [NotNull] object? argument, + [CallerArgumentExpression(nameof(argument))] string? paramName = null) + { + if (argument is null) + ArgumentNull(paramName); + } + + [DoesNotReturn] + [MethodImpl(MethodImplOptions.NoInlining)] + private static void ArgumentNull(string? paramName) => throw new ArgumentNullException(paramName); +} diff --git a/src/RequestFlow/Dispatch/DispatchMap.cs b/src/RequestFlow/Dispatch/DispatchMap.cs index 5a0bb93..dff2bd5 100644 --- a/src/RequestFlow/Dispatch/DispatchMap.cs +++ b/src/RequestFlow/Dispatch/DispatchMap.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; using System.Runtime.CompilerServices; #if NET8_0_OR_GREATER using System.Collections.Frozen; @@ -16,6 +17,6 @@ internal sealed class DispatchMap(Dictionary plans) #endif [MethodImpl(MethodImplOptions.AggressiveInlining)] - public bool TryGetPlanFor(Type requestType, out RequestPlanBase? plan) + public bool TryGetPlanFor(Type requestType, [NotNullWhen(true)] out RequestPlanBase? plan) => _plans.TryGetValue(requestType, out plan); } diff --git a/src/RequestFlow/Dispatch/NullTaskGuard.cs b/src/RequestFlow/Dispatch/NullTaskGuard.cs index 685c8c6..0a2a649 100644 --- a/src/RequestFlow/Dispatch/NullTaskGuard.cs +++ b/src/RequestFlow/Dispatch/NullTaskGuard.cs @@ -8,17 +8,17 @@ internal static class NullTaskGuard { [MethodImpl(MethodImplOptions.AggressiveInlining)] public static Task FromHandler(Task task, Type requestType) - => task ?? throw new HandlerNullTaskException(requestType); + => task ?? ThrowHelper.HandlerNullTask>(requestType); [MethodImpl(MethodImplOptions.AggressiveInlining)] public static Task FromStage(Task task, Type stageType) - => task ?? throw new StageNullTaskException(stageType); + => task ?? ThrowHelper.StageNullTask>(stageType); [MethodImpl(MethodImplOptions.AggressiveInlining)] public static Task FromEventHandler(Task task, Type eventType, Type handlerType) - => task ?? throw new EventHandlerNullTaskException(eventType, handlerType); + => task ?? ThrowHelper.EventHandlerNullTask(eventType, handlerType); [MethodImpl(MethodImplOptions.AggressiveInlining)] public static Task FromStrategy(Task task, Type strategyType) - => task ?? throw new EventStrategyNullTaskException(strategyType); + => task ?? ThrowHelper.EventStrategyNullTask(strategyType); } diff --git a/src/RequestFlow/Dispatch/RequestDispatcher.cs b/src/RequestFlow/Dispatch/RequestDispatcher.cs index d7eb99d..97f1211 100644 --- a/src/RequestFlow/Dispatch/RequestDispatcher.cs +++ b/src/RequestFlow/Dispatch/RequestDispatcher.cs @@ -12,16 +12,16 @@ internal sealed class RequestDispatcher(DispatchMap map, IServiceProvider servic /// public Task SendAsync(IRequest request, CancellationToken cancellationToken = default) { - if (request is null) - throw new ArgumentNullException(nameof(request)); + ThrowHelper.ThrowIfNull(request); Type requestType = request.GetType(); if (!_map.TryGetPlanFor(requestType, out RequestPlanBase? plan)) - throw new HandlerNotFoundException(requestType); + return ThrowHelper.HandlerNotFound>(requestType); if (plan is not RequestPlan typedPlan) - throw new ResponseTypeMismatchException(requestType, expected: plan!.ResponseType, actual: typeof(TResponse)); + return ThrowHelper.ResponseTypeMismatch>( + requestType, expected: plan.ResponseType, actual: typeof(TResponse)); return typedPlan.ExecuteAsync(request, _services, cancellationToken); } @@ -29,8 +29,7 @@ public Task SendAsync(IRequest request, Cancell /// public Task SendAsync(IRequest request, CancellationToken cancellationToken = default) { - if (request is null) - throw new ArgumentNullException(nameof(request)); + ThrowHelper.ThrowIfNull(request); return SendAsync(request, cancellationToken); } diff --git a/src/RequestFlow/Dispatch/Streaming/NullStreamGuard.cs b/src/RequestFlow/Dispatch/Streaming/NullStreamGuard.cs index c6bb134..5b4c8a8 100644 --- a/src/RequestFlow/Dispatch/Streaming/NullStreamGuard.cs +++ b/src/RequestFlow/Dispatch/Streaming/NullStreamGuard.cs @@ -8,9 +8,9 @@ internal static class NullStreamGuard { [MethodImpl(MethodImplOptions.AggressiveInlining)] public static IAsyncEnumerable FromHandler(IAsyncEnumerable stream, Type requestType) - => stream ?? throw new HandlerNullStreamException(requestType); + => stream ?? ThrowHelper.HandlerNullStream>(requestType); [MethodImpl(MethodImplOptions.AggressiveInlining)] public static IAsyncEnumerable FromStage(IAsyncEnumerable stream, Type stageType) - => stream ?? throw new StageNullStreamException(stageType); + => stream ?? ThrowHelper.StageNullStream>(stageType); } diff --git a/src/RequestFlow/Dispatch/Streaming/StreamDispatcher.cs b/src/RequestFlow/Dispatch/Streaming/StreamDispatcher.cs index 3a68643..4726ed1 100644 --- a/src/RequestFlow/Dispatch/Streaming/StreamDispatcher.cs +++ b/src/RequestFlow/Dispatch/Streaming/StreamDispatcher.cs @@ -19,18 +19,32 @@ internal sealed class StreamDispatcher(DispatchMap map, IServiceProvider service public IAsyncEnumerable Stream( IStreamRequest request, CancellationToken cancellationToken = default) { - if (request is null) - throw new ArgumentNullException(nameof(request)); + ThrowHelper.ThrowIfNull(request); Type requestType = request.GetType(); if (!_map.TryGetPlanFor(requestType, out RequestPlanBase? plan)) - throw new HandlerNotFoundException(requestType); + return ThrowHelper.HandlerNotFound>(requestType); if (plan is not StreamPlan streamPlan) - throw new ResponseTypeMismatchException(requestType, expected: plan!.ResponseType, actual: typeof(TItem)); + return ThrowHelper.ResponseTypeMismatch>( + requestType, expected: plan.ResponseType, actual: typeof(TItem)); - return Walk(streamPlan, request, _services, cancellationToken); + return cancellationToken.CanBeCanceled + ? Walk(streamPlan, request, _services, cancellationToken) + : WalkWithoutDispatchToken(streamPlan, request, _services); + } + + private static async IAsyncEnumerable WalkWithoutDispatchToken( + StreamPlan plan, + object request, + IServiceProvider services, + [EnumeratorCancellation] CancellationToken iterationToken = default) + { + await foreach (TItem item in plan.Execute(request, services, iterationToken).ConfigureAwait(false)) + { + yield return item; + } } // The dispatch token arrives as a parameter and the iteration token through the attribute, so diff --git a/src/RequestFlow/Dispatch/ThrowHelper.cs b/src/RequestFlow/Dispatch/ThrowHelper.cs new file mode 100644 index 0000000..cf02d9e --- /dev/null +++ b/src/RequestFlow/Dispatch/ThrowHelper.cs @@ -0,0 +1,57 @@ +using System; +using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; + +namespace RequestFlow; + +internal static class ThrowHelper +{ + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static void ThrowIfNull( + [NotNull] object? argument, + [CallerArgumentExpression(nameof(argument))] string? paramName = null) + { + if (argument is null) + ArgumentNull(paramName); + } + + [DoesNotReturn] + [MethodImpl(MethodImplOptions.NoInlining)] + private static void ArgumentNull(string? paramName) => throw new ArgumentNullException(paramName); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult HandlerNotFound(Type requestType) + => throw new HandlerNotFoundException(requestType); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult ResponseTypeMismatch(Type requestType, Type expected, Type actual) + => throw new ResponseTypeMismatchException(requestType, expected, actual); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult EventNotRegistered(Type eventType) + => throw new EventNotRegisteredException(eventType); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult HandlerNullTask(Type requestType) + => throw new HandlerNullTaskException(requestType); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult StageNullTask(Type stageType) + => throw new StageNullTaskException(stageType); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult EventHandlerNullTask(Type eventType, Type handlerType) + => throw new EventHandlerNullTaskException(eventType, handlerType); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult EventStrategyNullTask(Type strategyType) + => throw new EventStrategyNullTaskException(strategyType); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult HandlerNullStream(Type requestType) + => throw new HandlerNullStreamException(requestType); + + [MethodImpl(MethodImplOptions.NoInlining)] + public static TResult StageNullStream(Type stageType) + => throw new StageNullStreamException(stageType); +} diff --git a/src/RequestFlow/Dispatch/ValueTask/ValueRequestDispatcher.cs b/src/RequestFlow/Dispatch/ValueTask/ValueRequestDispatcher.cs index 9f4fe54..fa8cdb9 100644 --- a/src/RequestFlow/Dispatch/ValueTask/ValueRequestDispatcher.cs +++ b/src/RequestFlow/Dispatch/ValueTask/ValueRequestDispatcher.cs @@ -15,18 +15,17 @@ public ValueTask SendAsync( IValueRequest request, CancellationToken cancellationToken = default) { - if (request is null) - throw new ArgumentNullException(nameof(request)); + ThrowHelper.ThrowIfNull(request); Type requestType = request.GetType(); if (!_map.TryGetPlanFor(requestType, out RequestPlanBase? plan)) - throw new HandlerNotFoundException(requestType); + return ThrowHelper.HandlerNotFound>(requestType); if (plan is not ValueRequestPlan typedPlan) - throw new ResponseTypeMismatchException( + return ThrowHelper.ResponseTypeMismatch>( requestType, - expected: plan!.ResponseType, + expected: plan.ResponseType, actual: typeof(TResponse)); return typedPlan.ExecuteAsync(request, _services, cancellationToken); @@ -37,13 +36,12 @@ public ValueTask SendAsync( IValueRequest request, CancellationToken cancellationToken = default) { - if (request is null) - throw new ArgumentNullException(nameof(request)); + ThrowHelper.ThrowIfNull(request); Type requestType = request.GetType(); if (!_map.TryGetPlanFor(requestType, out RequestPlanBase? plan)) - throw new HandlerNotFoundException(requestType); + return ThrowHelper.HandlerNotFound(requestType); if (plan is ValueVoidRequestPlan voidPlan) return voidPlan.ExecuteVoidAsync(request, _services, cancellationToken); @@ -52,9 +50,9 @@ public ValueTask SendAsync( return ValueNoResultBridge.Discard( typedPlan.ExecuteAsync(request, _services, cancellationToken)); - throw new ResponseTypeMismatchException( + return ThrowHelper.ResponseTypeMismatch( requestType, - expected: plan!.ResponseType, + expected: plan.ResponseType, actual: typeof(NoResult)); } } diff --git a/src/RequestFlow/Events/EventMap.cs b/src/RequestFlow/Events/EventMap.cs index bf4c6f4..21a5a74 100644 --- a/src/RequestFlow/Events/EventMap.cs +++ b/src/RequestFlow/Events/EventMap.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; using System.Runtime.CompilerServices; #if NET8_0_OR_GREATER using System.Collections.Frozen; @@ -16,6 +17,6 @@ internal sealed class EventMap(Dictionary plans) #endif [MethodImpl(MethodImplOptions.AggressiveInlining)] - public bool TryGetPlanFor(Type eventType, out EventPlan? plan) + public bool TryGetPlanFor(Type eventType, [NotNullWhen(true)] out EventPlan? plan) => _plans.TryGetValue(eventType, out plan); } diff --git a/src/RequestFlow/Events/EventPlanFactory.cs b/src/RequestFlow/Events/EventPlanFactory.cs index ebd5e34..b4be81f 100644 --- a/src/RequestFlow/Events/EventPlanFactory.cs +++ b/src/RequestFlow/Events/EventPlanFactory.cs @@ -43,7 +43,9 @@ private static EventPlan CreatePlan( if (strategyType == typeof(SequentialPublishStrategy)) return new SequentialEventPlan(@event, entries); if (strategyType == typeof(ParallelPublishStrategy)) - return new ParallelEventPlan(@event, entries); + return entries.Length == 1 + ? new SequentialEventPlan(@event, entries) + : new ParallelEventPlan(@event, entries); if (strategyType == typeof(FailFastPublishStrategy)) return new SequentialEventPlan(@event, entries, stopOnFirstFailure: true); diff --git a/src/RequestFlow/Events/EventPublisher.cs b/src/RequestFlow/Events/EventPublisher.cs index d345122..718e72a 100644 --- a/src/RequestFlow/Events/EventPublisher.cs +++ b/src/RequestFlow/Events/EventPublisher.cs @@ -11,13 +11,12 @@ internal sealed class EventPublisher(EventMap map, IServiceProvider services) : public Task PublishAsync(IEvent @event, CancellationToken cancellationToken = default) { - if (@event is null) - throw new ArgumentNullException(nameof(@event)); + ThrowHelper.ThrowIfNull(@event, nameof(@event)); Type eventType = @event.GetType(); if (!_map.TryGetPlanFor(eventType, out EventPlan? plan)) - throw new EventNotRegisteredException(eventType); + return ThrowHelper.EventNotRegistered(eventType); - return plan!.ExecuteAsync(@event, _services, cancellationToken); + return plan.ExecuteAsync(@event, _services, cancellationToken); } } diff --git a/tests/Compat/IsExternalInit.cs b/tests/Compat/IsExternalInit.cs deleted file mode 100644 index df3e4e8..0000000 --- a/tests/Compat/IsExternalInit.cs +++ /dev/null @@ -1,5 +0,0 @@ -// Records compile to init accessors, and the compiler wants this marker type; .NET Framework does not ship it. -namespace System.Runtime.CompilerServices; - -internal static class IsExternalInit -{ } diff --git a/tests/Directory.Build.props b/tests/Directory.Build.props index bba9cd3..bbbcf1b 100644 --- a/tests/Directory.Build.props +++ b/tests/Directory.Build.props @@ -32,8 +32,8 @@ - - + + diff --git a/tests/RequestFlow.Cqrs.Tests.Unit/CqrsDispatcherTests.cs b/tests/RequestFlow.Cqrs.Tests.Unit/CqrsDispatcherTests.cs index df59447..aa23cf3 100644 --- a/tests/RequestFlow.Cqrs.Tests.Unit/CqrsDispatcherTests.cs +++ b/tests/RequestFlow.Cqrs.Tests.Unit/CqrsDispatcherTests.cs @@ -52,24 +52,30 @@ public async Task Given_Query_When_Sending_Query_Then_Forwards_Query_And_Token_A } [Fact] - public async Task Given_Null_Typed_Command_When_Sending_Command_Then_Throws_Argument_Null_Exception() + public void Given_Null_Typed_Command_When_Sending_Command_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - () => _sut.SendAsync((ICommand)null!)); + var exception = Should.Throw( + () => { _sut.SendAsync((ICommand)null!); }); + + exception.ParamName.ShouldBe("command"); } [Fact] - public async Task Given_Null_Void_Command_When_Sending_Command_Then_Throws_Argument_Null_Exception() + public void Given_Null_Void_Command_When_Sending_Command_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - () => _sut.SendAsync((ICommand)null!)); + var exception = Should.Throw( + () => { _sut.SendAsync((ICommand)null!); }); + + exception.ParamName.ShouldBe("command"); } [Fact] - public async Task Given_Null_Query_When_Sending_Query_Then_Throws_Argument_Null_Exception() + public void Given_Null_Query_When_Sending_Query_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - () => _sut.SendAsync((IQuery)null!)); + var exception = Should.Throw( + () => { _sut.SendAsync((IQuery)null!); }); + + exception.ParamName.ShouldBe("query"); } [Fact] diff --git a/tests/RequestFlow.Cqrs.Tests.Unit/CqrsValueDispatcherTests.cs b/tests/RequestFlow.Cqrs.Tests.Unit/CqrsValueDispatcherTests.cs index fdda492..0f6cebc 100644 --- a/tests/RequestFlow.Cqrs.Tests.Unit/CqrsValueDispatcherTests.cs +++ b/tests/RequestFlow.Cqrs.Tests.Unit/CqrsValueDispatcherTests.cs @@ -57,24 +57,30 @@ public async Task Given_A_Value_Query_When_Sending_Then_Forwards_It_And_The_Toke } [Fact] - public async Task Given_A_Null_Value_Command_When_Sending_Then_Throws_Argument_Null_Exception() + public void Given_A_Null_Value_Command_When_Sending_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - async () => await _sut.SendAsync((IValueCommand)null!)); + var exception = Should.Throw( + () => { _sut.SendAsync((IValueCommand)null!); }); + + exception.ParamName.ShouldBe("command"); } [Fact] - public async Task Given_A_Null_Void_Value_Command_When_Sending_Then_Throws_Argument_Null_Exception() + public void Given_A_Null_Void_Value_Command_When_Sending_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - async () => await _sut.SendAsync((IValueCommand)null!)); + var exception = Should.Throw( + () => { _sut.SendAsync((IValueCommand)null!); }); + + exception.ParamName.ShouldBe("command"); } [Fact] - public async Task Given_A_Null_Value_Query_When_Sending_Then_Throws_Argument_Null_Exception() + public void Given_A_Null_Value_Query_When_Sending_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - async () => await _sut.SendAsync((IValueQuery)null!)); + var exception = Should.Throw( + () => { _sut.SendAsync((IValueQuery)null!); }); + + exception.ParamName.ShouldBe("query"); } [Fact] diff --git a/tests/RequestFlow.Tests.Unit/Dispatch/RequestDispatcherTests.cs b/tests/RequestFlow.Tests.Unit/Dispatch/RequestDispatcherTests.cs index 641fb0f..709f5d8 100644 --- a/tests/RequestFlow.Tests.Unit/Dispatch/RequestDispatcherTests.cs +++ b/tests/RequestFlow.Tests.Unit/Dispatch/RequestDispatcherTests.cs @@ -25,10 +25,10 @@ public async Task Given_Registered_Handler_When_Sending_Request_Then_Returns_Han } [Fact] - public async Task Given_Unknown_Request_Type_When_Sending_Request_Then_Throws_Handler_Not_Found_Exception() + public void Given_Unknown_Request_Type_When_Sending_Request_Then_Throws_Handler_Not_Found_Exception() { - var exception = await Should.ThrowAsync( - () => _sut.SendAsync(new Unknown())); + var exception = Should.Throw( + () => { _sut.SendAsync(new Unknown()); }); exception.RequestType.ShouldBe(typeof(Unknown)); } @@ -55,10 +55,10 @@ public async Task Given_Synchronously_Completing_Void_Handler_When_Sending_Reque } [Fact] - public async Task Given_Covariant_Response_Type_When_Sending_Request_Then_Throws_Response_Type_Mismatch_Exception() + public void Given_Covariant_Response_Type_When_Sending_Request_Then_Throws_Response_Type_Mismatch_Exception() { - var exception = await Should.ThrowAsync( - () => _sut.SendAsync(new Ping("bob"))); + var exception = Should.Throw( + () => { _sut.SendAsync(new Ping("bob")); }); exception.RequestType.ShouldBe(typeof(Ping)); exception.ExpectedResponseType.ShouldBe(typeof(string)); @@ -90,17 +90,21 @@ await Should.ThrowAsync( } [Fact] - public async Task Given_Null_Request_When_Sending_Request_Then_Throws_Argument_Null_Exception() + public void Given_Null_Request_When_Sending_Request_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - () => _sut.SendAsync(null!)); + var exception = Should.Throw( + () => { _sut.SendAsync(null!); }); + + exception.ParamName.ShouldBe("request"); } [Fact] - public async Task Given_Null_Void_Request_When_Sending_Request_Then_Throws_Argument_Null_Exception() + public void Given_Null_Void_Request_When_Sending_Request_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - () => _sut.SendAsync((IRequest)null!)); + var exception = Should.Throw( + () => { _sut.SendAsync((IRequest)null!); }); + + exception.ParamName.ShouldBe("request"); } [Fact] @@ -190,6 +194,40 @@ public async Task Given_Cancellation_Token_When_Sending_Void_Request_Then_Handle await _logHandler.Received(1).HandleAsync(Arg.Any(), cts.Token); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Given_A_Void_Handler_Task_Faulted_With_Cancellation_When_Sending_Then_The_Returned_Task_Is_Canceled( + bool completesLater) + { + using var source = new CancellationTokenSource(); + source.Cancel(); + var failure = new OperationCanceledException(source.Token); + var completion = new TaskCompletionSource(); + _logHandler.HandleAsync(Arg.Any(), Arg.Any()).Returns(completion.Task); + if (!completesLater) + completion.SetException(failure); + + Task operation = _sut.SendAsync(new Log("hi")); + if (completesLater) + completion.SetException(failure); + + OperationCanceledException? exception = null; + try + { + await operation; + } + catch (OperationCanceledException caught) + { + exception = caught; + } + + exception.ShouldNotBeNull(); + exception.ShouldBeSameAs(failure); + exception.CancellationToken.ShouldBe(source.Token); + operation.IsCanceled.ShouldBeTrue(); + } + #region Initialization private readonly IRequestHandler _pingHandler; diff --git a/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamCancellationTests.cs b/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamCancellationTests.cs index 6cbb123..b5765f8 100644 --- a/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamCancellationTests.cs +++ b/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamCancellationTests.cs @@ -130,6 +130,44 @@ public async Task Given_A_Pass_Through_Stage_Substituting_A_Token_When_The_Dispa items.ShouldBe([0, 1, 2, 3, 4]); } + [Fact] + public async Task Given_A_Pass_Through_Stage_Substituting_A_Token_When_Only_The_Iteration_Token_Is_Cancelled_Then_The_Walk_Completes() + { + using var iteration = new CancellationTokenSource(); + using var shield = new CancellationTokenSource(); + ShieldingStage.Shield = shield.Token; + IStreamDispatcher sut = Build(options => options.AddStreamStage()); + + List items = []; + await foreach (int item in sut.Stream(new Shielded()).WithCancellation(iteration.Token)) + { + items.Add(item); + iteration.Cancel(); + } + + items.ShouldBe([0, 1, 2, 3, 4]); + } + + [Fact] + public async Task Given_One_Stream_With_Two_Enumerators_When_One_Iteration_Token_Is_Cancelled_Then_The_Other_Continues() + { + using var firstToken = new CancellationTokenSource(); + using var secondToken = new CancellationTokenSource(); + IAsyncEnumerable stream = Build().Stream(new Endless()); + await using IAsyncEnumerator first = stream.GetAsyncEnumerator(firstToken.Token); + await using IAsyncEnumerator second = stream.GetAsyncEnumerator(secondToken.Token); + + (await first.MoveNextAsync()).ShouldBeTrue(); + (await second.MoveNextAsync()).ShouldBeTrue(); + firstToken.Cancel(); + Task canceled = first.MoveNextAsync().AsTask(); + + await Should.ThrowAsync(() => canceled); + canceled.IsCanceled.ShouldBeTrue(); + (await second.MoveNextAsync()).ShouldBeTrue(); + second.Current.ShouldBe(1); + } + #region Helpers private static IStreamDispatcher Build(Action? configure = null) diff --git a/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamDispatcherTests.cs b/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamDispatcherTests.cs index 51bb9bd..77727ad 100644 --- a/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamDispatcherTests.cs +++ b/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamDispatcherTests.cs @@ -115,6 +115,22 @@ public async Task Given_A_Scanned_Stream_Handler_When_Streaming_Through_The_Cont items.ShouldBe([0, 1, 2]); } + [Fact] + public async Task Given_An_Enumerator_Disposed_Before_Moving_When_Streaming_Without_A_Token_Then_The_Handler_Is_Not_Resolved() + { + var services = Substitute.For(); + var map = new DispatchMap(new Dictionary + { + [typeof(Tail)] = new StreamPlan(), + }); + var dispatcher = new StreamDispatcher(map, services); + IAsyncEnumerator enumerator = dispatcher.Stream(new Tail(3)).GetAsyncEnumerator(); + + await enumerator.DisposeAsync(); + + services.DidNotReceive().GetService(Arg.Any()); + } + #region Initialization private readonly StreamDispatcher _sut; diff --git a/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamFailureCompatibilityTests.cs b/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamFailureCompatibilityTests.cs new file mode 100644 index 0000000..b6afba2 --- /dev/null +++ b/tests/RequestFlow.Tests.Unit/Dispatch/Streaming/StreamFailureCompatibilityTests.cs @@ -0,0 +1,127 @@ +using Microsoft.Extensions.DependencyInjection; + +namespace RequestFlow.Tests.Unit; + +public sealed class StreamFailureCompatibilityTests +{ + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task Given_Move_Next_Faulted_With_Cancellation_When_Enumerating_Then_The_Outer_Move_Is_Canceled_And_Disposes_The_Inner( + bool dispatchToken, bool failsLater) + { + using ServiceProvider provider = Build(); + using IServiceScope scope = provider.CreateScope(); + var dispatcher = scope.ServiceProvider.GetRequiredService(); + using var source = new CancellationTokenSource(); + source.Cancel(); + var failure = new OperationCanceledException(source.Token); + var request = new Pending(); + if (!failsLater) + request.Completion.SetException(failure); + await using IAsyncEnumerator enumerator = dispatcher + .Stream(request, dispatchToken ? source.Token : CancellationToken.None).GetAsyncEnumerator(); + + Task move = enumerator.MoveNextAsync().AsTask(); + if (failsLater) + { + move.IsCompleted.ShouldBeFalse(); + request.Completion.SetException(failure); + } + + await Should.ThrowAsync(() => move); + move.IsCanceled.ShouldBeTrue(); + request.Disposals.ShouldBe(1); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Given_Move_Next_Faulted_With_Cancellation_When_Awaiting_The_Value_Task_Then_The_Original_Exception_And_Token_Are_Preserved(bool dispatchToken) + { + using ServiceProvider provider = Build(); + using IServiceScope scope = provider.CreateScope(); + var dispatcher = scope.ServiceProvider.GetRequiredService(); + using var source = new CancellationTokenSource(); + source.Cancel(); + var failure = new OperationCanceledException(source.Token); + var request = new Pending(); + await using IAsyncEnumerator enumerator = dispatcher + .Stream(request, dispatchToken ? source.Token : CancellationToken.None).GetAsyncEnumerator(); + + ValueTask move = enumerator.MoveNextAsync(); + request.Completion.SetException(failure); + + OperationCanceledException? exception = null; + try { await move; } + catch (OperationCanceledException caught) { exception = caught; } + exception.ShouldBeSameAs(failure); + exception!.CancellationToken.ShouldBe(source.Token); + request.Disposals.ShouldBe(1); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Given_A_Stream_Disposed_After_One_Item_When_Stopping_Early_Then_The_Inner_Enumerator_Is_Disposed(bool dispatchToken) + { + using ServiceProvider provider = Build(); + using IServiceScope scope = provider.CreateScope(); + var dispatcher = scope.ServiceProvider.GetRequiredService(); + using var source = new CancellationTokenSource(); + var request = new Pending(); + request.Completion.SetResult(true); + IAsyncEnumerator enumerator = dispatcher + .Stream(request, dispatchToken ? source.Token : CancellationToken.None).GetAsyncEnumerator(); + + (await enumerator.MoveNextAsync()).ShouldBeTrue(); + enumerator.Current.ShouldBe(7); + await enumerator.DisposeAsync(); + + request.Disposals.ShouldBe(1); + } + + #region Helpers + + private static ServiceProvider Build() + { + var services = new ServiceCollection(); + services.AddRequestFlow(options => options.WithScopedHandlers().AddHandler()); + return services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true }); + } + + public sealed class Pending : IStreamRequest + { + public TaskCompletionSource Completion { get; } = new(); + public int Disposals { get; set; } + } + + public sealed class PendingHandler : IStreamRequestHandler + { + public IAsyncEnumerable Handle(Pending request, CancellationToken cancellationToken) + => new PendingSequence(request); + } + + private sealed class PendingSequence(Pending request) : IAsyncEnumerable + { + public IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken = default) + => new PendingEnumerator(request); + } + + private sealed class PendingEnumerator(Pending request) : IAsyncEnumerator + { + public int Current => 7; + + public ValueTask MoveNextAsync() => new(request.Completion.Task); + + public ValueTask DisposeAsync() + { + request.Disposals++; + return default; + } + } + + #endregion +} diff --git a/tests/RequestFlow.Tests.Unit/Dispatch/ValueTask/ValueRequestDispatcherTests.cs b/tests/RequestFlow.Tests.Unit/Dispatch/ValueTask/ValueRequestDispatcherTests.cs index d3fd480..47dffa5 100644 --- a/tests/RequestFlow.Tests.Unit/Dispatch/ValueTask/ValueRequestDispatcherTests.cs +++ b/tests/RequestFlow.Tests.Unit/Dispatch/ValueTask/ValueRequestDispatcherTests.cs @@ -17,19 +17,19 @@ public async Task Given_Registered_Value_Handler_When_Sending_Request_Then_Retur } [Fact] - public async Task Given_Unknown_Value_Request_When_Sending_Request_Then_Throws_Handler_Not_Found_Exception() + public void Given_Unknown_Value_Request_When_Sending_Request_Then_Throws_Handler_Not_Found_Exception() { - var exception = await Should.ThrowAsync( - async () => await _sut.SendAsync(new Unknown())); + var exception = Should.Throw( + () => { _sut.SendAsync(new Unknown()); }); exception.RequestType.ShouldBe(typeof(Unknown)); } [Fact] - public async Task Given_Covariant_Value_Response_When_Sending_Request_Then_Throws_Response_Type_Mismatch_Exception() + public void Given_Covariant_Value_Response_When_Sending_Request_Then_Throws_Response_Type_Mismatch_Exception() { - var exception = await Should.ThrowAsync( - async () => await _sut.SendAsync(new Fetch())); + var exception = Should.Throw( + () => { _sut.SendAsync(new Fetch()); }); exception.RequestType.ShouldBe(typeof(Fetch)); exception.ExpectedResponseType.ShouldBe(typeof(DerivedResult)); @@ -37,17 +37,21 @@ public async Task Given_Covariant_Value_Response_When_Sending_Request_Then_Throw } [Fact] - public async Task Given_Null_Value_Request_When_Sending_Request_Then_Throws_Argument_Null_Exception() + public void Given_Null_Value_Request_When_Sending_Request_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - async () => await _sut.SendAsync(null!)); + var exception = Should.Throw( + () => { _sut.SendAsync(null!); }); + + exception.ParamName.ShouldBe("request"); } [Fact] - public async Task Given_Null_Void_Value_Request_When_Sending_Request_Then_Throws_Argument_Null_Exception() + public void Given_Null_Void_Value_Request_When_Sending_Request_Then_Throws_Argument_Null_Exception() { - await Should.ThrowAsync( - async () => await _sut.SendAsync((IValueRequest)null!)); + var exception = Should.Throw( + () => { _sut.SendAsync((IValueRequest)null!); }); + + exception.ParamName.ShouldBe("request"); } [Fact] diff --git a/tests/RequestFlow.Tests.Unit/Events/ParallelEventCompletionTests.cs b/tests/RequestFlow.Tests.Unit/Events/ParallelEventCompletionTests.cs new file mode 100644 index 0000000..07d1e8f --- /dev/null +++ b/tests/RequestFlow.Tests.Unit/Events/ParallelEventCompletionTests.cs @@ -0,0 +1,115 @@ +namespace RequestFlow.Tests.Unit.Events; + +public sealed class ParallelEventCompletionTests +{ + [Fact] + public async Task Given_One_Pending_Handler_Among_Synchronous_Handlers_When_Publishing_Then_All_Start_Before_Waiting() + { + var gate = new TaskCompletionSource(); + var calls = new List(); + ParallelEventPlan plan = CreatePlan((index, _) => + { + calls.Add(index); + return index == 1 ? gate.Task : Task.CompletedTask; + }); + + Task publish = plan.ExecuteAsync(new Notice(), _services, default); + + calls.ShouldBe([0, 1, 2], ignoreOrder: false); + publish.IsCompleted.ShouldBeFalse(); + gate.SetResult(null); + await publish; + } + + [Fact] + public async Task Given_Concurrent_Publications_When_One_Completes_Synchronously_Then_The_Other_Keeps_Its_Pending_Handler() + { + var gate = new TaskCompletionSource(); + var pending = new Notice(gate.Task); + var entries = new HandlerEntry[] + { + (_, _, _) => Task.CompletedTask, + (_, value, _) => ((Notice)value).Operation, + (_, _, _) => Task.CompletedTask, + }; + ParallelEventPlan plan = CreatePlan(entries); + + Task first = plan.ExecuteAsync(pending, _services, default); + Task second = plan.ExecuteAsync(new Notice(), _services, default); + + await second; + first.IsCompleted.ShouldBeFalse(); + gate.SetResult(null); + await first; + } + + [Fact] + public async Task Given_A_Synchronous_Handler_Cancels_The_Publishing_Token_When_Publishing_Then_All_Entries_Still_Start() + { + using var source = new CancellationTokenSource(); + var calls = new List(); + ParallelEventPlan plan = CreatePlan((index, token) => + { + token.ShouldBe(source.Token); + calls.Add(index); + if (index == 0) + source.Cancel(); + return Task.CompletedTask; + }); + + await plan.ExecuteAsync(new Notice(), _services, source.Token); + + calls.ShouldBe([0, 1, 2], ignoreOrder: false); + } + + #region Initialization + + private readonly IServiceProvider _services = Substitute.For(); + + #endregion + + #region Helpers + + private static readonly Type[] HandlerTypes = [typeof(AlphaHandler), typeof(BravoHandler), typeof(CharlieHandler)]; + + private static ParallelEventPlan CreatePlan(Func start) + { + var entries = new HandlerEntry[HandlerTypes.Length]; + for (int i = 0; i < entries.Length; i++) + { + int index = i; + entries[i] = (_, _, token) => start(index, token); + } + return CreatePlan(entries); + } + + private static ParallelEventPlan CreatePlan(HandlerEntry[] entries) + { + var handlers = new EventHandlerModel[entries.Length]; + for (int i = 0; i < handlers.Length; i++) + handlers[i] = new EventHandlerModel(HandlerTypes[i], typeof(Notice), RequestFlowLifetime.Singleton); + return new ParallelEventPlan(new EventModel(typeof(Notice), handlers), entries); + } + + private sealed class Notice(Task? operation = null) : IEvent + { + public Task Operation { get; } = operation ?? Task.CompletedTask; + } + + private sealed class AlphaHandler : IEventHandler + { + public Task HandleAsync(Notice @event, CancellationToken cancellationToken) => Task.CompletedTask; + } + + private sealed class BravoHandler : IEventHandler + { + public Task HandleAsync(Notice @event, CancellationToken cancellationToken) => @event.Operation; + } + + private sealed class CharlieHandler : IEventHandler + { + public Task HandleAsync(Notice @event, CancellationToken cancellationToken) => Task.CompletedTask; + } + + #endregion +} diff --git a/tests/RequestFlow.Tests.Unit/Events/SingleHandlerParallelEventTests.cs b/tests/RequestFlow.Tests.Unit/Events/SingleHandlerParallelEventTests.cs new file mode 100644 index 0000000..2e96d68 --- /dev/null +++ b/tests/RequestFlow.Tests.Unit/Events/SingleHandlerParallelEventTests.cs @@ -0,0 +1,172 @@ +using Microsoft.Extensions.DependencyInjection; + +namespace RequestFlow.Tests.Unit.Events; + +public sealed class SingleHandlerParallelEventTests +{ +#if NET8_0_OR_GREATER +#if DEBUG + [Fact(Skip = "Debug allocates the async state machine. Run this allocation test in Release.")] +#else + [Fact] +#endif + public async Task Given_One_Synchronous_Parallel_Handler_When_Publishing_Then_No_Dispatch_Allocation_Is_Added() + { + using var context = new Context(); + for (int i = 0; i < 64; i++) await context.Publish(); + + long before = GC.GetAllocatedBytesForCurrentThread(); + for (int i = 0; i < 64; i++) await context.Publish(); + long allocated = GC.GetAllocatedBytesForCurrentThread() - before; + + allocated.ShouldBe(0); + context.State.Calls.ShouldBe(128); + } +#endif + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Given_One_Parallel_Handler_When_It_Faults_Then_Publication_Preserves_The_Handler_Failure(bool delayed) + { + using var context = new Context(); + var completion = new TaskCompletionSource(); + var failure = new InvalidOperationException("failure"); + context.State.Operation = completion.Task; + if (!delayed) completion.SetException(failure); + + Task result = context.Publish(); + if (delayed) + { + result.IsCompleted.ShouldBeFalse(); + completion.SetException(failure); + } + var exception = await Should.ThrowAsync(() => result); + + result.IsFaulted.ShouldBeTrue(); + EventHandlerFailure entry = exception.Failures.ShouldHaveSingleItem(); + entry.Exception.ShouldBeSameAs(failure); + entry.HandlerType.ShouldBe(typeof(Handler)); + entry.DeclaredEventType.ShouldBe(typeof(Notice)); + exception.Message.ShouldContain("1 of 1 handlers"); + exception.SkippedHandlerCount.ShouldBe(0); + context.State.Calls.ShouldBe(1); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Given_A_Pre_Canceled_Publish_When_Selecting_One_Parallel_Handler_Then_The_Handler_Is_Not_Started(bool typedSelection) + { + using var context = new Context(typedSelection); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + + Task result = context.Publish(cancellation.Token); + Exception? caught = null; + try { await result; } + catch (Exception observed) { caught = observed; } + + result.IsCanceled.ShouldBeTrue(); + var exception = caught.ShouldBeOfType(); + exception.CancellationToken.ShouldBe(cancellation.Token); + exception.Failures.ShouldBeEmpty(); + context.State.Calls.ShouldBe(0); + } + + [Fact] + public async Task Given_Cancellation_After_The_Only_Parallel_Handler_Starts_When_It_Succeeds_Then_Publication_Succeeds() + { + using var context = new Context(); + using var cancellation = new CancellationTokenSource(); + var completion = new TaskCompletionSource(); + context.State.Operation = completion.Task; + + Task result = context.Publish(cancellation.Token); + cancellation.Cancel(); + completion.SetResult(null); + await result; + + result.Status.ShouldBe(TaskStatus.RanToCompletion); + context.State.Token.ShouldBe(cancellation.Token); + context.State.Calls.ShouldBe(1); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Given_A_Parallel_Handler_Cancellation_When_Publishing_Then_It_Remains_A_Handler_Failure(bool faultedCancellation) + { + using var context = new Context(); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + context.State.Operation = faultedCancellation + ? Task.FromException(new OperationCanceledException(cancellation.Token)) + : Task.FromCanceled(cancellation.Token); + + Task result = context.Publish(); + var exception = await Should.ThrowAsync(() => result); + + result.IsFaulted.ShouldBeTrue(); + exception.Failures.ShouldHaveSingleItem().Exception + .ShouldBeAssignableTo().CancellationToken.ShouldBe(cancellation.Token); + } + + #region Helpers + + private sealed class Context : IDisposable + { + private readonly ServiceProvider _provider; + private readonly IServiceScope _scope; + private readonly IEventPublisher _publisher; + private readonly Notice _notice = new(); + public State State { get; } = new(); + + public Context(bool typedSelection = false) + { + var services = new ServiceCollection(); + services.AddSingleton(State); + services.AddRequestFlow(options => + { + options.WithScopedHandlers().AddEventHandler(); + if (typedSelection) options.PublishEventsInParallel(); + else options.PublishEventsInParallel(); + }); + _provider = services.BuildServiceProvider(new ServiceProviderOptions + { + ValidateOnBuild = true, + ValidateScopes = true, + }); + _scope = _provider.CreateScope(); + _publisher = _scope.ServiceProvider.GetRequiredService(); + } + + public Task Publish(CancellationToken token = default) => _publisher.PublishAsync(_notice, token); + public void Dispose() + { + _scope.Dispose(); + _provider.Dispose(); + } + } + + public sealed class State + { + public Task Operation { get; set; } = Task.CompletedTask; + public int Calls { get; set; } + public CancellationToken Token { get; set; } + } + + public sealed class Notice : IEvent; + + public sealed class Handler(State state) : IEventHandler + { + public Task HandleAsync(Notice @event, CancellationToken cancellationToken) + { + state.Calls++; + state.Token = cancellationToken; + return state.Operation; + } + } + + #endregion +} diff --git a/tests/RequestFlow.Tests.Unit/Stages/Streaming/StreamChainAllocationTests.cs b/tests/RequestFlow.Tests.Unit/Stages/Streaming/StreamChainAllocationTests.cs index 28bcac9..2fc90e0 100644 --- a/tests/RequestFlow.Tests.Unit/Stages/Streaming/StreamChainAllocationTests.cs +++ b/tests/RequestFlow.Tests.Unit/Stages/Streaming/StreamChainAllocationTests.cs @@ -61,6 +61,18 @@ public async Task Given_An_Async_Iterator_Stream_Chain_When_Enumerating_More_Ite longWalk.ShouldBe(shortWalk); } + [Fact] + public async Task Given_No_Dispatch_Cancellation_Token_When_Enumerating_Then_It_Allocates_Less_Than_A_Cancelable_Dispatch() + { + IStreamDispatcher dispatcher = Build(); + using var source = new CancellationTokenSource(); + + long withoutToken = await MeasureAsync(dispatcher); + long withToken = await MeasureAsync(dispatcher, cancellationToken: source.Token); + + withoutToken.ShouldBeLessThan(withToken); + } + #region Helpers private static IStreamDispatcher Build(Action? configure = null) @@ -78,22 +90,23 @@ private static IStreamDispatcher Build(Action? configure = n // GC.GetAllocatedBytesForCurrentThread counts this thread only, so the handler completes // synchronously and every level stays on the calling thread. - private static async Task MeasureAsync(IStreamDispatcher dispatcher, int items = 8) + private static async Task MeasureAsync( + IStreamDispatcher dispatcher, int items = 8, CancellationToken cancellationToken = default) { for (int i = 0; i < 64; i++) { - await DrainAsync(dispatcher, items); + await DrainAsync(dispatcher, items, cancellationToken); } long before = GC.GetAllocatedBytesForCurrentThread(); - await DrainAsync(dispatcher, items); + await DrainAsync(dispatcher, items, cancellationToken); return GC.GetAllocatedBytesForCurrentThread() - before; } - private static async Task DrainAsync(IStreamDispatcher dispatcher, int items) + private static async Task DrainAsync(IStreamDispatcher dispatcher, int items, CancellationToken cancellationToken) { - await foreach (int item in dispatcher.Stream(new Ticks(items))) + await foreach (int item in dispatcher.Stream(new Ticks(items), cancellationToken)) { } } diff --git a/tests/RequestFlow.Tests.Unit/Stages/VoidStageFailureCompatibilityTests.cs b/tests/RequestFlow.Tests.Unit/Stages/VoidStageFailureCompatibilityTests.cs new file mode 100644 index 0000000..6cb1a09 --- /dev/null +++ b/tests/RequestFlow.Tests.Unit/Stages/VoidStageFailureCompatibilityTests.cs @@ -0,0 +1,84 @@ +using Microsoft.Extensions.DependencyInjection; + +namespace RequestFlow.Tests.Unit; + +public sealed class VoidStageFailureCompatibilityTests +{ + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task Given_A_Cancellation_Fault_In_A_Void_Pipeline_When_Sending_Then_The_Outer_Stage_And_Caller_Observe_Cancellation( + bool failInStage, bool completesLater) + { + using ServiceProvider provider = Build(); + using IServiceScope scope = provider.CreateScope(); + var dispatcher = scope.ServiceProvider.GetRequiredService(); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + var failure = new OperationCanceledException(cancellation.Token); + var request = new Signal(failInStage); + if (!completesLater) + request.Completion.SetException(failure); + + Task operation = dispatcher.SendAsync(request); + if (completesLater) + request.Completion.SetException(failure); + Exception? caught = null; + try { await operation; } + catch (Exception exception) { caught = exception; } + + caught.ShouldBeSameAs(failure); + operation.IsCanceled.ShouldBeTrue(); + request.Inner!.IsCanceled.ShouldBeTrue(); + request.FinallyCalls.ShouldBe(1); + } + + #region Helpers + + private static ServiceProvider Build() + { + var services = new ServiceCollection(); + services.AddRequestFlow(options => options.WithScopedHandlers() + .AddHandler() + .AddStage(typeof(ObservationStage<,>)) + .AddStage()); + return services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true }); + } + + public sealed class Signal(bool failInStage) : IRequest + { + public bool FailInStage { get; } = failInStage; + public TaskCompletionSource Completion { get; } = new(); + public Task? Inner { get; set; } + public int FinallyCalls { get; set; } + } + + public sealed class SignalHandler : IRequestHandler + { + public Task HandleAsync(Signal request, CancellationToken cancellationToken) + => request.Completion.Task; + } + + public sealed class FailureStage : IRequestStage + { + public Task HandleAsync(Signal request, Continuation next, CancellationToken cancellationToken) + => request.FailInStage ? request.Completion.Task : next.InvokeAsync(); + } + + public sealed class ObservationStage : IRequestStage + where TRequest : IRequest + { + public async Task HandleAsync(TRequest request, Continuation next, CancellationToken cancellationToken) + { + var signal = (Signal)(object)request; + Task inner = next.InvokeAsync(); + signal.Inner = inner; + try { return await inner.ConfigureAwait(false); } + finally { signal.FinallyCalls++; } + } + } + + #endregion +}