From b1e4b0ed60be461a27042af4d166399b40a08c5b Mon Sep 17 00:00:00 2001 From: Illia Filippov Date: Sun, 2 Aug 2026 15:48:13 +0200 Subject: [PATCH] feat(stages)!: swap StageDelegate for IContinuation, resolve lazily An interface can grow an InvokeAsync overload that takes arguments. A delegate signature cannot, not without breaking every stage that exists. Each level now resolves from the container the first time it runs, the handler included, so a stage that short-circuits never builds what sits below it. A container failure comes out of the next.InvokeAsync() call that reached the broken level, where the stages around it can catch it. A dispatch through N stages allocates N objects instead of 2N+1. The per-level delegate and the per-dispatch stage array are both gone, and a repeated next call allocates nothing. Three stages cost 192 bytes on net10.0 against 416 before. The broken-contract failures throw HandlerNullTaskException, StageNullTaskException, and OverlappingNextCallException instead of a plain InvalidOperationException. Each carries the type at fault in a property. The two null-task types share an abstract NullTaskException base, so one catch clause covers both. BREAKING CHANGE: StageDelegate and StageDelegate are removed. A stage takes IContinuation or IContinuation and calls next.InvokeAsync() where it called next(). --- CHANGELOG.md | 14 + docs/exceptions.md | 46 +- docs/stages.md | 26 +- .../Exceptions/HandlerNullTaskException.cs | 16 + .../Exceptions/NullTaskException.cs | 14 + .../OverlappingNextCallException.cs | 18 + .../Exceptions/StageNullTaskException.cs | 18 + src/RequestFlow.Abstractions/IContinuation.cs | 32 ++ src/RequestFlow.Abstractions/IRequestStage.cs | 25 +- .../CqrsRequestFlowBuilderExtensions.cs | 2 +- src/RequestFlow/Dispatch/NullTaskGuard.cs | 12 +- .../Registration/RegistrationValidator.cs | 2 +- .../Registration/RequestFlowOptions.cs | 21 +- .../Registration/RequestFlowRegistry.cs | 65 ++- .../ServiceCollectionExtensions.cs | 7 +- src/RequestFlow/Stages/GuardSentinel.cs | 14 + src/RequestFlow/Stages/StageExecutor.cs | 152 ++++-- src/RequestFlow/Stages/StagedRequestPlan.cs | 17 +- .../Stages/StagedVoidRequestPlan.cs | 20 +- src/RequestFlow/Stages/TypedStageExecutor.cs | 21 +- src/RequestFlow/Stages/VoidStageExecutor.cs | 34 +- .../RequestDispatcherTests.cs | 12 +- .../RequestFlowRegistryTests.cs | 105 ++++ .../Stages/AddStageTests.cs | 30 +- .../Stages/StageClosingTests.cs | 12 +- .../Stages/StageExecutorTests.cs | 485 ++++++++++++++---- .../Stages/StageLifetimeTests.cs | 254 ++++++++- .../Stages/StagePipelineTests.cs | 32 +- .../Stages/StageSemanticsTests.cs | 71 ++- 29 files changed, 1243 insertions(+), 334 deletions(-) create mode 100644 src/RequestFlow.Abstractions/Exceptions/HandlerNullTaskException.cs create mode 100644 src/RequestFlow.Abstractions/Exceptions/NullTaskException.cs create mode 100644 src/RequestFlow.Abstractions/Exceptions/OverlappingNextCallException.cs create mode 100644 src/RequestFlow.Abstractions/Exceptions/StageNullTaskException.cs create mode 100644 src/RequestFlow.Abstractions/IContinuation.cs create mode 100644 src/RequestFlow/Stages/GuardSentinel.cs create mode 100644 tests/RequestFlow.Tests.Unit/RequestFlowRegistryTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ebd2d5..b90ec3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,20 @@ Releases are cut from this file. The `release` workflow reads the section matchi ## [Unreleased] +### Added + +- Dedicated exceptions for the three broken-contract failures that used to throw a plain `InvalidOperationException`: `HandlerNullTaskException` and `StageNullTaskException` for a null task returned from `HandleAsync`, and `OverlappingNextCallException` for a stage that calls `next` while its earlier call is still running. Each carries the type at fault in a property (`RequestType` or `StageType`) instead of only naming it in the message. The two null-task types share an abstract `NullTaskException` base, so one catch clause covers both, and all three still derive from `InvalidOperationException`. + +### Changed + +- A stage receives `next` as `IContinuation` (or `IContinuation` on the void form) instead of the `StageDelegate` delegate types, which are gone. Stage bodies call `await next.InvokeAsync()` where they called `await next()`. Nothing else about a stage changes. An interface also leaves room to add arguments to a future `InvokeAsync` overload, which a delegate signature cannot take without breaking every stage. +- A stage now resolves from the container when its level first runs rather than up front, and so does the handler. A stage that short-circuits builds neither the stages below it nor the handler behind them, which is the point for a cache stage sitting in front of a repository. A repeated `next` call still walks the instances the dispatch already resolved, the handler included. +- For a request with stages, a container failure building a stage or the handler now surfaces from inside the chain, out of the `next.InvokeAsync()` call that reached that level, where the stages wrapped around it can catch it. A retry stage with a broad `catch` will retry a missing registration. Turn on `ServiceProviderOptions.ValidateOnBuild` to keep registration mistakes at startup. + +### Performance + +- A dispatch through a chain of N stages allocates N objects instead of 2N+1: the per-level delegate and the per-dispatch stage array are both gone, and a stage that invokes `next` more than once no longer allocates on the repeat. Measured on net10.0 with a synchronous handler, a three-stage chain costs 192 bytes per dispatch against 416 before, and each further stage adds 56 bytes rather than 112. + ## [1.0.0-preview.3] - 2026-08-02 ### Added diff --git a/docs/exceptions.md b/docs/exceptions.md index e9eea53..1989d4c 100644 --- a/docs/exceptions.md +++ b/docs/exceptions.md @@ -9,12 +9,14 @@ Every exception RequestFlow throws, when it surfaces, and how to fix it. | `RequestFlowValidationException` | Startup validation | Any registration problem; one throw lists all of them | | `HandlerNotFoundException` | `SendAsync` | The dispatched request type has no registered handler | | `ResponseTypeMismatchException` | `SendAsync` | The call site's response type differs from the registered one | -| `InvalidOperationException` | `SendAsync` | A handler or stage returned a null task, or a stage overlapped two `next` calls | +| `HandlerNullTaskException` | `SendAsync` | A handler returned a null task from `HandleAsync` | +| `StageNullTaskException` | `SendAsync` | A stage returned a null task from `HandleAsync` | +| `OverlappingNextCallException` | `SendAsync` | A stage called `next` while its earlier call was still running | | `InvalidOperationException` | `WhereHandlerImplements` | A second handler filter added to one stage | | `ArgumentNullException` | All public entry points | A required argument is null | | `ArgumentException` | `RegisterGenericHandler` | `closingTypes` contains a null element | -The three RequestFlow types are sealed, live in the `RequestFlow` namespace in the `RequestFlow.Abstractions` package, and derive from `InvalidOperationException`. All of them signal programmer errors: fix the registration or the call site instead of catching them. +The RequestFlow types live in the `RequestFlow` namespace in the `RequestFlow.Abstractions` package and derive from `InvalidOperationException`. All are sealed except `NullTaskException`, the abstract base the two null-task types share. All of them signal programmer errors: fix the registration or the call site instead of catching them. ## RequestFlowValidationException @@ -127,18 +129,38 @@ public sealed class SyncInventory : IRequest, IRequest { } With a handler registered as `IRequestHandler`, the natural call `SendAsync(new SyncInventory())` cannot infer `TResponse` from two candidate interfaces. It silently binds the void `SendAsync(IRequest)` overload, asks for `NoResult`, and throws. The fix belongs in the model, not the call site: give each request type exactly one `IRequest` interface, and split it in two if both shapes are needed. -## InvalidOperationException +## HandlerNullTaskException -Plain `InvalidOperationException` signals a broken handler or stage contract. Three cases surface at dispatch, one at registration: +Thrown by `SendAsync` when a handler returns a null task from `HandleAsync`. The `RequestType` property holds the request whose handler returned it. -| Message starts with | Thrown from | Fix | -| ----------------------------------------------------------- | --------------------------- | -------------------------------------------------------------------------- | -| `The handler for '...' returned a null task from HandleAsync.` | `SendAsync` | Return a task from every path; use `Task.CompletedTask` or `Task.FromResult` for synchronous results | -| `Stage '...' returned a null task from HandleAsync...` | `SendAsync` | Return the task from `next`, or a completed task when short-circuiting | -| `Stage '...' called next while the task from its earlier call was still running.` | `SendAsync` | Await each `next` call before calling it again; each call runs the rest of the chain | -| `This stage already filters on '...'` | `AddStage` configure delegate | One `WhereHandlerImplements` per stage; give the target handlers one shared contract | +Return a task from every path: `Task.FromResult(value)` for a synchronous result, `Task.CompletedTask` for the void form. The usual source is a test double left without a configured return value. -The null-task checks exist so the failure names the handler or stage at fault instead of surfacing as a `NullReferenceException` at the await. The overlap check stops a stage from running the rest of the chain twice at the same time; a sequential second call, the retry shape, is allowed (see [stages.md](stages.md)). +## StageNullTaskException + +Thrown by `SendAsync` when a stage returns a null task from `HandleAsync`. The `StageType` property holds the stage class at fault. + +Return the task from `next.InvokeAsync()`, or a completed task when short-circuiting. + +Both null-task types derive from `NullTaskException`, so one catch clause covers a handler and a stage: + +```csharp +catch (NullTaskException e) +{ + // e is a HandlerNullTaskException or a StageNullTaskException +} +``` + +The base class is abstract with no public constructor, so those two are the only cases it ever holds. Both checks exist so the failure names the handler or stage at fault instead of surfacing as a `NullReferenceException` at the await. + +## OverlappingNextCallException + +Thrown by `SendAsync` when a stage invokes `next` while the task from its earlier call is still running. The `StageType` property holds the stage class at fault. + +Await each call before making the next one. The check stops a stage from running the rest of the chain twice at the same time; a sequential second call, the retry shape, is allowed (see [stages.md](stages.md)). + +## Plain InvalidOperationException + +One case is left with no type of its own. Adding a second `WhereHandlerImplements` to one stage throws from the `AddStage` configure delegate, with a message starting `This stage already filters on '...'`. A stage takes one handler filter, so give the target handlers one shared contract instead. ## Argument validation @@ -161,3 +183,5 @@ Handler and stage exceptions propagate as thrown. The dispatcher and the stage c Cancellation follows the same rule. The token passes to `HandleAsync` untouched, and an `OperationCanceledException` surfaces from the handler like any other exception. Container failures keep the container's own exception types. The dispatcher resolves the handler from the service provider on every dispatch, so a handler with a missing constructor dependency, or a scoped handler resolved from the root provider, throws the container's `InvalidOperationException` at dispatch time. See [lifetimes.md](lifetimes.md) for the lifetime rules that prevent these. + +On a request with stages, that failure lands inside the chain. Each level resolves when it first runs, so the container's exception comes out of the `next.InvokeAsync()` call that reached the broken level, and the stages wrapped around it can catch it like any other exception. A retry stage with a broad `catch` will retry a missing registration until it runs out of attempts. Catch the exceptions you mean to handle, and turn on `ServiceProviderOptions.ValidateOnBuild` so a registration mistake fails at startup instead. diff --git a/docs/stages.md b/docs/stages.md index 305e29f..fcff99a 100644 --- a/docs/stages.md +++ b/docs/stages.md @@ -4,7 +4,7 @@ A stage wraps the handler of every request it applies to: code before and after ## Writing a stage -Implement `IRequestStage`. The `next` delegate runs the rest of the chain, ending at the handler: +Implement `IRequestStage`. Invoking `next` runs the rest of the chain, ending at the handler: ```csharp using RequestFlow; @@ -13,10 +13,10 @@ public sealed class LoggingStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { Console.WriteLine($"Handling {typeof(TRequest).Name}"); - TResponse response = await next(); + TResponse response = await next.InvokeAsync(); Console.WriteLine($"Handled {typeof(TRequest).Name}"); return response; } @@ -25,9 +25,9 @@ public sealed class LoggingStage : IRequestStage : IRequestStage, IAudited { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { - TResponse response = await next(); + TResponse response = await next.InvokeAsync(); // write the audit record return response; } @@ -76,11 +76,11 @@ public sealed class ErrorTranslationStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { try { - return await next(); + return await next.InvokeAsync(); } catch (DomainException e) { @@ -106,13 +106,13 @@ The filter looks at the handler class, not the request, so a module can mark its ## Void requests -A stage for void requests implements `IRequestStage`, takes the parameterless `StageDelegate`, and returns plain `Task`: +A stage for void requests implements `IRequestStage`, takes the void form `IContinuation`, and returns plain `Task`: ```csharp public sealed class CacheClearGuard : IRequestStage { - public Task HandleAsync(ClearCache request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(ClearCache request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } ``` diff --git a/src/RequestFlow.Abstractions/Exceptions/HandlerNullTaskException.cs b/src/RequestFlow.Abstractions/Exceptions/HandlerNullTaskException.cs new file mode 100644 index 0000000..d77d8bd --- /dev/null +++ b/src/RequestFlow.Abstractions/Exceptions/HandlerNullTaskException.cs @@ -0,0 +1,16 @@ +using System; + +namespace RequestFlow; + +/// +/// Thrown by when the handler for the +/// dispatched request returns a null task from HandleAsync. +/// +public sealed class HandlerNullTaskException(Type requestType) + : NullTaskException($"The handler for '{requestType.FullName}' returned a null task from HandleAsync.") +{ + /// + /// The request type whose handler returned the null task. + /// + public Type RequestType { get; } = requestType; +} diff --git a/src/RequestFlow.Abstractions/Exceptions/NullTaskException.cs b/src/RequestFlow.Abstractions/Exceptions/NullTaskException.cs new file mode 100644 index 0000000..e1090bc --- /dev/null +++ b/src/RequestFlow.Abstractions/Exceptions/NullTaskException.cs @@ -0,0 +1,14 @@ +using System; + +namespace RequestFlow; + +/// +/// Base class for the exceptions thrown when a handler or a stage returns a null task +/// from HandleAsync. +/// +public abstract class NullTaskException : InvalidOperationException +{ + private protected NullTaskException(string message) + : base(message) + { } +} diff --git a/src/RequestFlow.Abstractions/Exceptions/OverlappingNextCallException.cs b/src/RequestFlow.Abstractions/Exceptions/OverlappingNextCallException.cs new file mode 100644 index 0000000..cce52b7 --- /dev/null +++ b/src/RequestFlow.Abstractions/Exceptions/OverlappingNextCallException.cs @@ -0,0 +1,18 @@ +using System; + +namespace RequestFlow; + +/// +/// Thrown by when a stage invokes next +/// while the task from its earlier call is still running. +/// +public sealed class OverlappingNextCallException(Type stageType) + : InvalidOperationException( + $"Stage '{stageType.FullName}' called next while the task from its earlier call was still running. " + + "Await that task before calling next again: each call runs the rest of the chain, so overlapping calls would run it twice at once.") +{ + /// + /// The stage type that called next twice over. + /// + public Type StageType { get; } = stageType; +} diff --git a/src/RequestFlow.Abstractions/Exceptions/StageNullTaskException.cs b/src/RequestFlow.Abstractions/Exceptions/StageNullTaskException.cs new file mode 100644 index 0000000..cb0da89 --- /dev/null +++ b/src/RequestFlow.Abstractions/Exceptions/StageNullTaskException.cs @@ -0,0 +1,18 @@ +using System; + +namespace RequestFlow; + +/// +/// Thrown by when a stage returns a null +/// task from HandleAsync. +/// +public sealed class StageNullTaskException(Type stageType) + : NullTaskException( + $"Stage '{stageType.FullName}' returned a null task from HandleAsync; " + + "return the task from next, or a completed task when short-circuiting.") +{ + /// + /// The stage type that returned the null task. + /// + public Type StageType { get; } = stageType; +} diff --git a/src/RequestFlow.Abstractions/IContinuation.cs b/src/RequestFlow.Abstractions/IContinuation.cs new file mode 100644 index 0000000..b89a02a --- /dev/null +++ b/src/RequestFlow.Abstractions/IContinuation.cs @@ -0,0 +1,32 @@ +using System.Threading.Tasks; + +namespace RequestFlow; + +/// +/// The rest of the stage chain below one stage, ending at the request's handler. Invoke it +/// again after its task completes to run the rest of the chain again; invoking it while an +/// earlier call is still running throws . +/// +/// +/// A repeated call walks the same stage instances: the chain resolves them once per +/// dispatch, not per call, so state a stage kept from the first pass is still there. +/// +/// The response the chain produces. +public interface IContinuation +{ + /// + /// Runs the rest of the chain. + /// + Task InvokeAsync(); +} + +/// +/// The void form of , under the same rules. +/// +public interface IContinuation +{ + /// + /// Runs the rest of the chain. + /// + Task InvokeAsync(); +} diff --git a/src/RequestFlow.Abstractions/IRequestStage.cs b/src/RequestFlow.Abstractions/IRequestStage.cs index 42fc6a9..3a61358 100644 --- a/src/RequestFlow.Abstractions/IRequestStage.cs +++ b/src/RequestFlow.Abstractions/IRequestStage.cs @@ -3,23 +3,6 @@ namespace RequestFlow; -/// -/// Runs the rest of the stage chain, ending at the request's handler. Call it again after -/// its task completes to run the rest of the chain again; calling it while an earlier call -/// is still running throws . -/// -/// -/// A repeated call walks the same stage instances: the chain resolves them once per -/// dispatch, not per call, so state a stage kept from the first pass is still there. -/// -/// The response the chain produces. -public delegate Task StageDelegate(); - -/// -/// The void form of , under the same rules. -/// -public delegate Task StageDelegate(); - /// /// Runs around the handler of every request this stage applies to. The implementing /// class's generic constraints decide which requests those are. @@ -30,10 +13,10 @@ public interface IRequestStage where TRequest : IRequest { /// - /// Wraps the rest of the chain for . Call + /// Wraps the rest of the chain for . Invoke /// to continue, or skip it to short-circuit. /// - Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken); + Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken); } /// @@ -45,8 +28,8 @@ public interface IRequestStage where TRequest : IRequest { /// - /// Wraps the rest of the chain for . Call + /// Wraps the rest of the chain for . Invoke /// to continue, or skip it to short-circuit. /// - Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken); + Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken); } diff --git a/src/RequestFlow.Cqrs/CqrsRequestFlowBuilderExtensions.cs b/src/RequestFlow.Cqrs/CqrsRequestFlowBuilderExtensions.cs index 5ff0853..cafa05e 100644 --- a/src/RequestFlow.Cqrs/CqrsRequestFlowBuilderExtensions.cs +++ b/src/RequestFlow.Cqrs/CqrsRequestFlowBuilderExtensions.cs @@ -4,7 +4,7 @@ using RequestFlow; using RequestFlow.Cqrs; -// Microsoft's own convention for registration extensions: AddCqrs is visible in Program.cs without an extra using. +// Same namespace convention as AddRequestFlow: AddCqrs needs no extra using. namespace Microsoft.Extensions.DependencyInjection; /// diff --git a/src/RequestFlow/Dispatch/NullTaskGuard.cs b/src/RequestFlow/Dispatch/NullTaskGuard.cs index 76cc6fc..290e3ca 100644 --- a/src/RequestFlow/Dispatch/NullTaskGuard.cs +++ b/src/RequestFlow/Dispatch/NullTaskGuard.cs @@ -5,7 +5,7 @@ namespace RequestFlow; /// -/// Rejects a null task returned by a handler with an +/// Rejects a null task returned by a handler with a /// that names the request type. /// internal static class NullTaskGuard @@ -14,7 +14,7 @@ internal static class NullTaskGuard public static Task ThrowIfNull(Task task, Type requestType) { if (task is null) - throw new InvalidOperationException(HandlerMessage(requestType)); + throw new HandlerNullTaskException(requestType); return task; } @@ -23,14 +23,8 @@ public static Task ThrowIfNull(Task task, Type public static Task ThrowIfNull(Task task, Type requestType) { if (task is null) - throw new InvalidOperationException(HandlerMessage(requestType)); + throw new HandlerNullTaskException(requestType); return task; } - - /// - /// The error message for a handler that returned a null task, shared by every dispatch path. - /// - public static string HandlerMessage(Type requestType) - => $"The handler for '{requestType.FullName}' returned a null task from HandleAsync."; } diff --git a/src/RequestFlow/Registration/RegistrationValidator.cs b/src/RequestFlow/Registration/RegistrationValidator.cs index dedae5b..088c2f4 100644 --- a/src/RequestFlow/Registration/RegistrationValidator.cs +++ b/src/RequestFlow/Registration/RegistrationValidator.cs @@ -168,7 +168,7 @@ private static bool ImplementsStageContract(Type stageType) // MakeGenericType substitutes positionally and StageClosing closes a one-parameter // definition over the request alone, so the interface's request argument has to be the // stage's own parameter for the closed type to name the dispatched request. A stage that - // breaks this closes into a type no request can match, or drags along a parameter its contract never uses. + // breaks this closes into a type no request can match. private static bool ClosesOverItsOwnParameters(Type stageType) { Type[] parameters = stageType.GetGenericArguments(); diff --git a/src/RequestFlow/Registration/RequestFlowOptions.cs b/src/RequestFlow/Registration/RequestFlowOptions.cs index a2dbf29..73083b8 100644 --- a/src/RequestFlow/Registration/RequestFlowOptions.cs +++ b/src/RequestFlow/Registration/RequestFlowOptions.cs @@ -124,17 +124,18 @@ public RequestFlowOptions RegisterGenericHandler(Type handlerType, params Type[] /// /// Registers to run around the handler of every request it - /// applies to. Pass an open generic definition such as typeof(LoggingStage<,>) - /// to let the stage's own generic constraints decide which requests it reaches, or a - /// closed stage class to target a single request contract. A closed stage is not - /// restricted to the one request type it names: TRequest is contravariant, so a - /// stage declared for a base request also wraps every request that derives from it. - /// narrows that set further. One stage type belongs to a chain - /// once, so a second call naming the same type is a duplicate whatever it filters on. - /// Registration order is execution order, outermost first. A null argument and a repeated - /// WhereHandlerImplements call throw here; an invalid stage surfaces as a - /// problem when the dispatch map is built. + /// applies to. Registration order is execution order, outermost first. /// + /// + /// Pass an open generic definition such as typeof(LoggingStage<,>) to let the + /// stage's own constraints decide which requests it reaches, or a closed stage class to + /// target one request contract. A closed stage is not restricted to the request type it + /// names: TRequest is contravariant, so it also wraps every request deriving from + /// that one, and narrows the set further. A stage type belongs + /// to a chain once, so a second call naming it is a duplicate whatever it filters on. An + /// invalid stage surfaces as a problem when + /// the dispatch map is built. + /// /// /// public RequestFlowOptions AddStage(Type stageType, Action? configure = null) diff --git a/src/RequestFlow/Registration/RequestFlowRegistry.cs b/src/RequestFlow/Registration/RequestFlowRegistry.cs index 27e6613..86a3bcc 100644 --- a/src/RequestFlow/Registration/RequestFlowRegistry.cs +++ b/src/RequestFlow/Registration/RequestFlowRegistry.cs @@ -147,7 +147,7 @@ .. RegistrationValidator.ValidateAliasedStages(_stageDeclarations, _handlers, Cl Dictionary plans = []; foreach (var handler in _handlers) { - plans[handler.RequestType] = CreatePlan(handler, stagePlans.StageTypesByRequest[handler.RequestType]); + plans[handler.RequestType] = CreatePlan(handler, stagePlans.ChainsByRequest[handler.RequestType]); } return new DispatchMap(plans); @@ -156,7 +156,7 @@ .. RegistrationValidator.ValidateAliasedStages(_stageDeclarations, _handlers, Cl // Ordering and chain shape are decided at freeze, never per AddRequestFlow call. private StagePlanSet BuildStagePlans() { - Dictionary stageTypesByRequest = []; + Dictionary chainsByRequest = []; HashSet appliedStageTypes = []; List ordered = []; @@ -172,15 +172,32 @@ private StagePlanSet BuildStagePlans() appliedStageTypes.Add(declaration.StageType); } - stageTypesByRequest[handler.RequestType] = ordered.ToArray(); + Type[] stageTypes = ordered.ToArray(); + chainsByRequest[handler.RequestType] = new StageChain(stageTypes, TypedShapesFor(handler, stageTypes)); } - return new StagePlanSet(stageTypesByRequest, appliedStageTypes); + return new StagePlanSet(chainsByRequest, appliedStageTypes); } - private static RequestPlanBase CreatePlan(HandlerRegistration handler, Type[] stageTypes) + // Only a void request can take stages of either contract shape, so it is the only chain + // that has to record which shape each level runs under. A stage that implements both is + // run as the two-parameter form, the shape that carries the response type. + private static bool[] TypedShapesFor(HandlerRegistration handler, Type[] stageTypes) { - if (stageTypes.Length == 0) + if (!handler.IsVoid || stageTypes.Length == 0) + return []; + + Type typedContract = typeof(IRequestStage<,>).MakeGenericType(handler.RequestType, typeof(NoResult)); + bool[] typedShapes = new bool[stageTypes.Length]; + for (int i = 0; i < stageTypes.Length; i++) + typedShapes[i] = typedContract.IsAssignableFrom(stageTypes[i]); + + return typedShapes; + } + + private static RequestPlanBase CreatePlan(HandlerRegistration handler, StageChain chain) + { + if (chain.StageTypes.Length == 0) { Type planType = handler.IsVoid ? typeof(VoidRequestPlan<>).MakeGenericType(handler.RequestType) @@ -188,23 +205,41 @@ private static RequestPlanBase CreatePlan(HandlerRegistration handler, Type[] st return (RequestPlanBase)Activator.CreateInstance(planType)!; } - Type stagedPlanType = handler.IsVoid - ? typeof(StagedVoidRequestPlan<>).MakeGenericType(handler.RequestType) - : typeof(StagedRequestPlan<,>).MakeGenericType(handler.RequestType, handler.ResponseType); - // Wrapped in an object array on purpose: Type[] converts to object[], so handing // stageTypes straight through would be read as one constructor argument per stage type. - return (RequestPlanBase)Activator.CreateInstance(stagedPlanType, [(object)stageTypes])!; + if (handler.IsVoid) + { + Type voidPlanType = typeof(StagedVoidRequestPlan<>).MakeGenericType(handler.RequestType); + return (RequestPlanBase)Activator.CreateInstance( + voidPlanType, [chain.StageTypes, chain.TypedShapes])!; + } + + Type stagedPlanType = typeof(StagedRequestPlan<,>) + .MakeGenericType(handler.RequestType, handler.ResponseType); + + return (RequestPlanBase)Activator.CreateInstance(stagedPlanType, [(object)chain.StageTypes])!; } } /// -/// The ordered stage types for each request type, plus the stage types that reached at least -/// one request. +/// The stage chain for each request type, plus the stage types that reached at least one +/// request. /// -internal sealed class StagePlanSet(Dictionary stageTypesByRequest, HashSet appliedStageTypes) +internal sealed class StagePlanSet(Dictionary chainsByRequest, HashSet appliedStageTypes) { - public Dictionary StageTypesByRequest { get; } = stageTypesByRequest; + public Dictionary ChainsByRequest { get; } = chainsByRequest; public ISet AppliedStageTypes { get; } = appliedStageTypes; } + +/// +/// One request's stages in execution order. records, per position, +/// whether the stage runs as ; it is empty for +/// a request whose stages can only take that one shape. +/// +internal sealed class StageChain(Type[] stageTypes, bool[] typedShapes) +{ + public Type[] StageTypes { get; } = stageTypes; + + public bool[] TypedShapes { get; } = typedShapes; +} diff --git a/src/RequestFlow/Registration/ServiceCollectionExtensions.cs b/src/RequestFlow/Registration/ServiceCollectionExtensions.cs index 156e175..0c968c6 100644 --- a/src/RequestFlow/Registration/ServiceCollectionExtensions.cs +++ b/src/RequestFlow/Registration/ServiceCollectionExtensions.cs @@ -4,7 +4,7 @@ using Microsoft.Extensions.DependencyInjection.Extensions; using RequestFlow; -// Microsoft's own convention for IServiceCollection extensions: AddRequestFlow is visible in Program.cs without an extra using. +// Microsoft's convention for IServiceCollection extensions: AddRequestFlow needs no extra using. namespace Microsoft.Extensions.DependencyInjection; /// @@ -111,9 +111,8 @@ private static void RegisterHandlers( // Walks the whole accumulated cross product on every call rather than a delta, so a stage // declared by an earlier call reaches requests scanned by a later one; the closing cache - // makes the repeated pairs cheap. Skipping a type that is already present leaves a stage - // the consumer registered themselves on its own lifetime; a keyed descriptor is a - // different service, so it does not count as present. + // makes the repeated pairs cheap. Skipping a type already present leaves a stage the + // consumer registered on its own lifetime; a keyed descriptor is a different service. private static void RegisterStages( IServiceCollection services, IReadOnlyList declarations, diff --git a/src/RequestFlow/Stages/GuardSentinel.cs b/src/RequestFlow/Stages/GuardSentinel.cs new file mode 100644 index 0000000..156f9d6 --- /dev/null +++ b/src/RequestFlow/Stages/GuardSentinel.cs @@ -0,0 +1,14 @@ +using System.Threading.Tasks; + +namespace RequestFlow; + +/// +/// The marker a guard slot holds while a next call is being claimed. +/// +internal static class GuardSentinel +{ + /// + /// Never completes, so a level that reads it treats the claiming call as still in flight. + /// + public static readonly Task Claimed = new TaskCompletionSource().Task; +} diff --git a/src/RequestFlow/Stages/StageExecutor.cs b/src/RequestFlow/Stages/StageExecutor.cs index 4ec98c7..7e96c42 100644 --- a/src/RequestFlow/Stages/StageExecutor.cs +++ b/src/RequestFlow/Stages/StageExecutor.cs @@ -5,16 +5,21 @@ namespace RequestFlow; /// -/// Runs the stage chain by recursion: each level hands its stage a continuation that enters -/// the level below. One instance per dispatch, holding the request and token for every level. +/// Runs the stage chain by recursion: each level hands its stage the level below it. One +/// instance per dispatch, holding the request and token for every level, and standing in for +/// the level the outermost stage re-enters. /// -internal abstract class StageExecutor +internal abstract class StageExecutor : IContinuation, IContinuation where TRequest : IRequest { private readonly int _stageCount; private readonly TRequest _request; private readonly CancellationToken _cancellationToken; + // The outermost stage re-enters through the executor, so its next-call state sits here + // rather than in a continuation of its own. + private NextCall _next; + protected StageExecutor(int stageCount, TRequest request, CancellationToken cancellationToken) { _stageCount = stageCount; @@ -25,73 +30,120 @@ protected StageExecutor(int stageCount, TRequest request, CancellationToken canc /// /// Runs the chain, starting at the outermost stage. /// - public Task RunAsync() => EnterAsync(0); + // Nothing re-enters the outermost level, so its stage needs no slot to be held in between. + public Task RunAsync() + => _stageCount == 0 ? StartHandlerAsync() : StartStageAsync(0, ResolveStage(0), this); + + /// + public Task InvokeAsync() => EnterBelowAsync(0, ref _next); + + /// + Task IContinuation.InvokeAsync() => InvokeAsync(); + + /// + /// The stage that runs at . Each level asks once per dispatch and + /// keeps the answer. + /// + protected abstract object ResolveStage(int index); + /// + /// Invokes the stage at with as the rest + /// of the chain. + /// protected abstract Task InvokeStageAsync( - int index, TRequest request, StageDelegate next, CancellationToken cancellationToken); + int index, object stage, IContinuation next, TRequest request, CancellationToken cancellationToken); + /// + /// Invokes the handler. Reached only once the chain runs to the bottom, so a stage that + /// short-circuits never builds it; the instance is kept so a stage that calls next again + /// reuses it. + /// protected abstract Task InvokeHandlerAsync(TRequest request, CancellationToken cancellationToken); /// - /// The runtime type of the stage at , used to name it in errors. + /// The type of the stage at , used to name it in errors. /// protected abstract Type StageTypeAt(int index); - private Task EnterAsync(int index) + private Task StartStageAsync(int index, object stage, IContinuation next) { - Task task = index < _stageCount - ? InvokeStageAsync(index, _request, new Continuation(this, index + 1).InvokeAsync, _cancellationToken) - : InvokeHandlerAsync(_request, _cancellationToken); - + Task task = InvokeStageAsync(index, stage, next, _request, _cancellationToken); if (task is null) - throw new InvalidOperationException(NullTaskMessage(index)); + throw new StageNullTaskException(StageTypeAt(index)); return task; } - // Which level a next call enters has to live in the delegate itself: the moment a stage - // suspends on anything, shared executor state stops saying which frame is calling. Each - // level therefore gets its own continuation, which also carries the state that catches a - // stage calling next again while its earlier call is still running. - private sealed class Continuation(StageExecutor executor, int index) - { - // Never completes, so callers that read it while a claim is held treat it as a call - // still in flight and throw. - private static readonly Task Claimed = new TaskCompletionSource().Task; + private Task StartHandlerAsync() + => NullTaskGuard.ThrowIfNull(InvokeHandlerAsync(_request, _cancellationToken), typeof(TRequest)); - private Task? _running; + /// + /// Enters the level under the stage at , under the guard in + /// , which belongs to that stage's calls. + /// + private Task EnterBelowAsync(int callerIndex, ref NextCall slot) + { + if (!TryClaim(ref slot.InFlight, out Task? prior)) + throw new OverlappingNextCallException(StageTypeAt(callerIndex)); - public Task InvokeAsync() + try { - // The compare-exchange makes the check and the claim one atomic step; of two - // simultaneous callers, the loser sees either the sentinel or a value that moved. - Task? running = Volatile.Read(ref _running); - if (running is { IsCompleted: false } - || Interlocked.CompareExchange(ref _running, Claimed, running) != running) - throw new InvalidOperationException(executor.OverlappingNextMessage(index - 1)); - - try - { - Task task = executor.EnterAsync(index); - Volatile.Write(ref _running, task); - return task; - } - catch - { - // A synchronous throw releases the claim so an outer retry stage may call next again. - Volatile.Write(ref _running, running); - throw; - } + Task task = StartBelowAsync(callerIndex + 1, ref slot.Below); + Volatile.Write(ref slot.InFlight, task); + return task; } + catch + { + // A synchronous throw releases the claim so an outer retry stage may call next again. + Volatile.Write(ref slot.InFlight, prior); + throw; + } + } + + private Task StartBelowAsync(int index, ref Continuation? below) + => index == _stageCount + ? StartHandlerAsync() + : (below ??= new Continuation(this, index)).EnterAsync(); + + // The compare-exchange makes the check and the claim one atomic step; of two simultaneous + // callers, the loser sees either the sentinel or a value that moved. + private static bool TryClaim(ref Task? guard, out Task? prior) + { + prior = Volatile.Read(ref guard); + return prior is not { IsCompleted: false } + && Interlocked.CompareExchange(ref guard, GuardSentinel.Claimed, prior) == prior; + } + + /// + /// One stage's next-call state: the guard that admits one call at a time, and the level its + /// calls enter. + /// + // Always reached by ref. A copy would guard storage nobody reads. + private struct NextCall + { + public Task? InFlight; + public Continuation? Below; } - private string OverlappingNextMessage(int caller) - => $"Stage '{StageTypeAt(caller).FullName}' called next while the task from its earlier call was still running. " + - "Await that task before calling next again: each call runs the rest of the chain, so overlapping calls would run it twice at once."; + // Which level a next call enters has to live in the object the call reaches: the moment a + // stage suspends, shared executor state stops saying which frame is calling. Each level + // below the outermost therefore gets its own continuation, holding its stage and guard state. + private sealed class Continuation(StageExecutor executor, int index) + : IContinuation, IContinuation + { + private object? _stage; + private NextCall _next; + + /// + public Task InvokeAsync() => executor.EnterBelowAsync(index, ref _next); - private string NullTaskMessage(int index) - => index < _stageCount - ? $"Stage '{StageTypeAt(index).FullName}' returned a null task from HandleAsync; " + - "return the task from next, or a completed task when short-circuiting." - : NullTaskGuard.HandlerMessage(typeof(TRequest)); + /// + Task IContinuation.InvokeAsync() => InvokeAsync(); + + // The stage above may call next again, so the level keeps the stage it resolved the + // first time and every later pass runs that same instance. The guard admits one call at + // a time, which is also what publishes the slot to the next caller's thread. + internal Task EnterAsync() + => executor.StartStageAsync(index, _stage ??= executor.ResolveStage(index), this); + } } diff --git a/src/RequestFlow/Stages/StagedRequestPlan.cs b/src/RequestFlow/Stages/StagedRequestPlan.cs index c8a751f..899a6c6 100644 --- a/src/RequestFlow/Stages/StagedRequestPlan.cs +++ b/src/RequestFlow/Stages/StagedRequestPlan.cs @@ -7,8 +7,9 @@ namespace RequestFlow; /// /// Closed plan for one request/response pair wrapped in stages. The ordered stage types are -/// fixed when the dispatch map freezes; the instances resolve from the supplied provider on -/// each call, so DI lifetimes hold. +/// fixed when the dispatch map freezes; each stage instance and the handler resolve from the +/// supplied provider when their level first runs, so DI lifetimes hold and a level the chain +/// never reaches is never built. /// internal sealed class StagedRequestPlan(Type[] stageTypes) : RequestPlan where TRequest : IRequest @@ -16,14 +17,6 @@ internal sealed class StagedRequestPlan(Type[] stageTypes) /// public override Task ExecuteAsync( IRequest request, IServiceProvider services, CancellationToken cancellationToken) - { - var stages = new IRequestStage[stageTypes.Length]; - for (int i = 0; i < stages.Length; i++) - stages[i] = (IRequestStage)services.GetRequiredService(stageTypes[i]); - - var handler = services.GetRequiredService>(); - - return new TypedStageExecutor( - stages, handler, (TRequest)request, cancellationToken).RunAsync(); - } + => new TypedStageExecutor( + stageTypes, services, (TRequest)request, cancellationToken).RunAsync(); } diff --git a/src/RequestFlow/Stages/StagedVoidRequestPlan.cs b/src/RequestFlow/Stages/StagedVoidRequestPlan.cs index c9331b1..7ca12ba 100644 --- a/src/RequestFlow/Stages/StagedVoidRequestPlan.cs +++ b/src/RequestFlow/Stages/StagedVoidRequestPlan.cs @@ -6,23 +6,17 @@ namespace RequestFlow; /// -/// Closed plan for one void request wrapped in stages. The ordered stage types are fixed when -/// the dispatch map freezes; the instances resolve from the supplied provider on each call, so -/// DI lifetimes hold. +/// Closed plan for one void request wrapped in stages. The ordered stage types and the +/// contract shape each one runs under are fixed when the dispatch map freezes; each stage +/// instance and the handler resolve from the supplied provider when their level first runs, so +/// DI lifetimes hold and a level the chain never reaches is never built. /// -internal sealed class StagedVoidRequestPlan(Type[] stageTypes) : RequestPlan +internal sealed class StagedVoidRequestPlan(Type[] stageTypes, bool[] typedShapes) : RequestPlan where TRequest : IRequest { /// public override Task ExecuteAsync( IRequest request, IServiceProvider services, CancellationToken cancellationToken) - { - var stages = new object[stageTypes.Length]; - for (int i = 0; i < stages.Length; i++) - stages[i] = services.GetRequiredService(stageTypes[i]); - - var handler = services.GetRequiredService>(); - - return new VoidStageExecutor(stages, handler, (TRequest)request, cancellationToken).RunAsync(); - } + => new VoidStageExecutor( + stageTypes, typedShapes, services, (TRequest)request, cancellationToken).RunAsync(); } diff --git a/src/RequestFlow/Stages/TypedStageExecutor.cs b/src/RequestFlow/Stages/TypedStageExecutor.cs index 56a056b..f16624f 100644 --- a/src/RequestFlow/Stages/TypedStageExecutor.cs +++ b/src/RequestFlow/Stages/TypedStageExecutor.cs @@ -1,6 +1,7 @@ using System; using System.Threading; using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; namespace RequestFlow; @@ -8,22 +9,28 @@ namespace RequestFlow; /// Stage chain that terminates at . /// internal sealed class TypedStageExecutor( - IRequestStage[] stages, - IRequestHandler handler, + Type[] stageTypes, + IServiceProvider services, TRequest request, CancellationToken cancellationToken) - : StageExecutor(stages.Length, request, cancellationToken) + : StageExecutor(stageTypes.Length, request, cancellationToken) where TRequest : IRequest { + private IRequestHandler? _handler; + + /// + protected override object ResolveStage(int index) => services.GetRequiredService(stageTypes[index]); + /// protected override Task InvokeStageAsync( - int index, TRequest request, StageDelegate next, CancellationToken cancellationToken) - => stages[index].HandleAsync(request, next, cancellationToken); + int index, object stage, IContinuation next, TRequest request, CancellationToken cancellationToken) + => ((IRequestStage)stage).HandleAsync(request, next, cancellationToken); /// protected override Task InvokeHandlerAsync(TRequest request, CancellationToken cancellationToken) - => handler.HandleAsync(request, cancellationToken); + => (_handler ??= services.GetRequiredService>()) + .HandleAsync(request, cancellationToken); /// - protected override Type StageTypeAt(int index) => stages[index].GetType(); + protected override Type StageTypeAt(int index) => stageTypes[index]; } diff --git a/src/RequestFlow/Stages/VoidStageExecutor.cs b/src/RequestFlow/Stages/VoidStageExecutor.cs index b03ea26..709ba50 100644 --- a/src/RequestFlow/Stages/VoidStageExecutor.cs +++ b/src/RequestFlow/Stages/VoidStageExecutor.cs @@ -1,38 +1,48 @@ using System; using System.Threading; using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; namespace RequestFlow; /// /// Stage chain that terminates at the standalone . -/// Its stages come in both contract shapes, so the array is untyped and each level picks. +/// Its stages come in both contract shapes, so which shape each level runs is settled when the +/// dispatch map freezes and read from here. /// internal sealed class VoidStageExecutor( - object[] stages, - IRequestHandler handler, + Type[] stageTypes, + bool[] typedShapes, + IServiceProvider services, TRequest request, CancellationToken cancellationToken) - : StageExecutor(stages.Length, request, cancellationToken) + : StageExecutor(stageTypes.Length, request, cancellationToken) where TRequest : IRequest { + private IRequestHandler? _handler; + + /// + protected override object ResolveStage(int index) => services.GetRequiredService(stageTypes[index]); + /// protected override Task InvokeStageAsync( - int index, TRequest request, StageDelegate next, CancellationToken cancellationToken) + int index, object stage, IContinuation next, TRequest request, CancellationToken cancellationToken) { - object stage = stages[index]; - if (stage is IRequestStage typed) - return typed.HandleAsync(request, next, cancellationToken); + if (typedShapes[index]) + return ((IRequestStage)stage).HandleAsync(request, next, cancellationToken); - // The void shape wraps the same continuation, so both forms share its guard state. + // Task converts to the void shape's Task return, so one level object serves + // both forms. Both reach the same level and share its guard state. return NoResultBridge.CompleteOrNull( - ((IRequestStage)stage).HandleAsync(request, new StageDelegate(next.Invoke), cancellationToken)); + ((IRequestStage)stage).HandleAsync(request, (IContinuation)next, cancellationToken)); } /// protected override Task InvokeHandlerAsync(TRequest request, CancellationToken cancellationToken) - => NoResultBridge.CompleteOrNull(handler.HandleAsync(request, cancellationToken)); + => NoResultBridge.CompleteOrNull( + (_handler ??= services.GetRequiredService>()) + .HandleAsync(request, cancellationToken)); /// - protected override Type StageTypeAt(int index) => stages[index].GetType(); + protected override Type StageTypeAt(int index) => stageTypes[index]; } diff --git a/tests/RequestFlow.Tests.Unit/RequestDispatcherTests.cs b/tests/RequestFlow.Tests.Unit/RequestDispatcherTests.cs index 50278e5..4e9a9cc 100644 --- a/tests/RequestFlow.Tests.Unit/RequestDispatcherTests.cs +++ b/tests/RequestFlow.Tests.Unit/RequestDispatcherTests.cs @@ -109,11 +109,11 @@ public async Task Given_Handler_That_Returns_A_Null_Task_When_Sending_Request_Th _pingHandler.HandleAsync(Arg.Any(), Arg.Any()) .Returns(default(Task)!); - var exception = await Should.ThrowAsync( + var exception = await Should.ThrowAsync( () => _sut.SendAsync(new Ping("bob"))); - exception.Message.ShouldContain(nameof(Ping)); - exception.Message.ShouldContain("null task"); + exception.RequestType.ShouldBe(typeof(Ping)); + exception.ShouldBeAssignableTo(); } [Fact] @@ -122,11 +122,11 @@ public async Task Given_Void_Handler_That_Returns_A_Null_Task_When_Sending_Reque _logHandler.HandleAsync(Arg.Any(), Arg.Any()) .Returns(default(Task)!); - var exception = await Should.ThrowAsync( + var exception = await Should.ThrowAsync( () => _sut.SendAsync(new Log("hi"))); - exception.Message.ShouldContain(nameof(Log)); - exception.Message.ShouldContain("null task"); + exception.RequestType.ShouldBe(typeof(Log)); + exception.ShouldBeAssignableTo(); } [Fact] diff --git a/tests/RequestFlow.Tests.Unit/RequestFlowRegistryTests.cs b/tests/RequestFlow.Tests.Unit/RequestFlowRegistryTests.cs new file mode 100644 index 0000000..3e773a3 --- /dev/null +++ b/tests/RequestFlow.Tests.Unit/RequestFlowRegistryTests.cs @@ -0,0 +1,105 @@ +using Microsoft.Extensions.DependencyInjection; +using RequestFlow; + +namespace RequestFlow.Tests.Unit; + +public sealed class RequestFlowRegistryTests +{ + [Fact] + public void Given_One_Applicable_Stage_When_Building_The_Dispatch_Map_Then_Request_Gets_The_Staged_Plan() + { + DispatchMap map = BuildMap(o => o.AddStage(typeof(WrapStage<,>))); + + map.TryGet(typeof(Echo), out RequestPlanBase? plan); + + plan.ShouldBeOfType>(); + } + + [Fact] + public void Given_One_Applicable_Stage_When_Building_The_Dispatch_Map_Then_Void_Request_Gets_The_Staged_Void_Plan() + { + DispatchMap map = BuildMap(o => o.AddStage(typeof(WrapStage<,>))); + + map.TryGet(typeof(Purge), out RequestPlanBase? plan); + + plan.ShouldBeOfType>(); + } + + [Fact] + public void Given_Two_Applicable_Stages_When_Building_The_Dispatch_Map_Then_Request_Gets_The_General_Staged_Plan() + { + DispatchMap map = BuildMap(o => o.AddStage(typeof(WrapStage<,>)).AddStage(typeof(ExtraStage<,>))); + + map.TryGet(typeof(Echo), out RequestPlanBase? plan); + + plan.ShouldBeOfType>(); + } + + [Fact] + public void Given_Two_Applicable_Stages_When_Building_The_Dispatch_Map_Then_Void_Request_Gets_The_General_Staged_Void_Plan() + { + DispatchMap map = BuildMap(o => o.AddStage(typeof(WrapStage<,>)).AddStage(typeof(ExtraStage<,>))); + + map.TryGet(typeof(Purge), out RequestPlanBase? plan); + + plan.ShouldBeOfType>(); + } + + [Fact] + public void Given_No_Stages_When_Building_The_Dispatch_Map_Then_Request_Gets_The_Plain_Plan() + { + DispatchMap map = BuildMap(); + + map.TryGet(typeof(Echo), out RequestPlanBase? plan); + + plan.ShouldBeOfType>(); + } + + #region Helpers + + private static DispatchMap BuildMap(Action? configure = null) + { + var services = new ServiceCollection(); + services.AddRequestFlow(o => + { + o.RegisterHandlersFromAssemblyContaining(); + configure?.Invoke(o); + }); + + return services.BuildServiceProvider().GetRequiredService(); + } + + public sealed record Echo(string Text) : IRequest; + + public sealed record Purge : IRequest; + + public sealed class EchoHandler : IRequestHandler + { + public Task HandleAsync(Echo request, CancellationToken cancellationToken) + => Task.FromResult(request.Text); + } + + public sealed class PurgeHandler : IRequestHandler + { + public Task HandleAsync(Purge request, CancellationToken cancellationToken) + => Task.CompletedTask; + } + + public sealed class WrapStage : IRequestStage + where TRequest : IRequest + { + public Task HandleAsync( + TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); + } + + public sealed class ExtraStage : IRequestStage + where TRequest : IRequest + { + public Task HandleAsync( + TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); + } + + #endregion +} diff --git a/tests/RequestFlow.Tests.Unit/Stages/AddStageTests.cs b/tests/RequestFlow.Tests.Unit/Stages/AddStageTests.cs index 8a18283..037b4c3 100644 --- a/tests/RequestFlow.Tests.Unit/Stages/AddStageTests.cs +++ b/tests/RequestFlow.Tests.Unit/Stages/AddStageTests.cs @@ -207,29 +207,29 @@ private interface ITag private sealed class LoggingStage : IRequestStage where TRequest : IRequest { - public Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } private sealed class PingAuditStage : IRequestStage { - public Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } // One type parameter that the contract never uses as its request, so validation rejects // it even though it implements IRequestStage. private sealed class OneParameterStage : IRequestStage { - public Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } private sealed class ResponseBoundStage : IRequestStage where TRequest : IRequest { - public Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } // Implements the contract, but with the parameters transposed, so closing it over a @@ -237,21 +237,21 @@ public Task HandleAsync(TRequest request, StageDelegate next, Ca private sealed class SwappedStage : IRequestStage where TRequest : IRequest { - public Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } private sealed class VoidOnlyStage : IRequestStage where TRequest : IRequest { - public Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } private sealed class WipeAuditStage : IRequestStage { - public Task HandleAsync(Wipe request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(Wipe request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } private sealed class NotAStage @@ -259,7 +259,7 @@ private sealed class NotAStage private abstract class AbstractStage : IRequestStage { - public abstract Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken); + public abstract Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken); } #endregion diff --git a/tests/RequestFlow.Tests.Unit/Stages/StageClosingTests.cs b/tests/RequestFlow.Tests.Unit/Stages/StageClosingTests.cs index 497b474..4d7cedf 100644 --- a/tests/RequestFlow.Tests.Unit/Stages/StageClosingTests.cs +++ b/tests/RequestFlow.Tests.Unit/Stages/StageClosingTests.cs @@ -187,21 +187,21 @@ public Task HandleAsync(Log request, CancellationToken cancellationToken) private sealed class LoggingStage : IRequestStage where TRequest : IRequest { - public Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } private sealed class TaggedOnlyStage : IRequestStage where TRequest : IRequest, ITag { - public Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } private sealed class PingAuditStage : IRequestStage { - public Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) - => next(); + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } #endregion diff --git a/tests/RequestFlow.Tests.Unit/Stages/StageExecutorTests.cs b/tests/RequestFlow.Tests.Unit/Stages/StageExecutorTests.cs index abbd812..c0237cd 100644 --- a/tests/RequestFlow.Tests.Unit/Stages/StageExecutorTests.cs +++ b/tests/RequestFlow.Tests.Unit/Stages/StageExecutorTests.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.DependencyInjection; using RequestFlow; namespace RequestFlow.Tests.Unit; @@ -7,7 +8,7 @@ public sealed class StageExecutorTests [Fact] public async Task Given_No_Stages_When_Running_Executor_Then_Handler_Produces_Response() { - var sut = new TypedStageExecutor([], _pingHandler, new Ping("hi"), CancellationToken.None); + var sut = PingExecutor(); string result = await sut.RunAsync(); @@ -18,8 +19,8 @@ public async Task Given_No_Stages_When_Running_Executor_Then_Handler_Produces_Re public async Task Given_Two_Stages_When_Running_Executor_Then_First_Registered_Stage_Is_Outermost() { List log = []; - IRequestStage[] stages = [new RecordingStage("outer", log), new RecordingStage("inner", log)]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new RecordingStage("outer", log), new RecordingStage("inner", log)]; + var sut = PingExecutor(stages); await sut.RunAsync(); @@ -30,8 +31,8 @@ public async Task Given_Two_Stages_When_Running_Executor_Then_First_Registered_S public async Task Given_Stage_That_Awaits_Before_Calling_Next_When_Running_Executor_Then_Chain_Completes() { List log = []; - IRequestStage[] stages = [new AwaitBeforeNextStage("outer", log)]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new AwaitBeforeNextStage("outer", log)]; + var sut = PingExecutor(stages); string result = await sut.RunAsync(); @@ -43,8 +44,8 @@ public async Task Given_Stage_That_Awaits_Before_Calling_Next_When_Running_Execu public async Task Given_Two_Stages_That_Await_Before_Calling_Next_When_Running_Executor_Then_First_Registered_Stage_Is_Outermost() { List log = []; - IRequestStage[] stages = [new AwaitBeforeNextStage("outer", log), new AwaitBeforeNextStage("inner", log)]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new AwaitBeforeNextStage("outer", log), new AwaitBeforeNextStage("inner", log)]; + var sut = PingExecutor(stages); await sut.RunAsync(); @@ -54,8 +55,8 @@ public async Task Given_Two_Stages_That_Await_Before_Calling_Next_When_Running_E [Fact] public async Task Given_Stage_That_Skips_Next_When_Running_Executor_Then_Handler_Is_Not_Invoked() { - IRequestStage[] stages = [new ShortCircuitStage("cached")]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new ShortCircuitStage("cached")]; + var sut = PingExecutor(stages); string result = await sut.RunAsync(); @@ -68,8 +69,8 @@ public async Task Given_Throwing_Handler_When_Running_Executor_Then_Exception_Pr { _pingHandler.HandleAsync(Arg.Any(), Arg.Any()) .Returns(Task.FromException(new InvalidTimeZoneException("no such zone"))); - IRequestStage[] stages = [new RecordingStage("outer", [])]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new RecordingStage("outer", [])]; + var sut = PingExecutor(stages); var exception = await Should.ThrowAsync(() => sut.RunAsync()); @@ -81,7 +82,9 @@ public async Task Given_Cancellation_Token_When_Running_Executor_Then_Stage_And_ { using var cts = new CancellationTokenSource(); var stage = new TokenCapturingStage(); - var sut = new TypedStageExecutor([stage], _pingHandler, new Ping("hi"), cts.Token); + object[] stages = [stage]; + var sut = new TypedStageExecutor( + StageTypes(stages), ChainProvider(_pingHandler, stages), new Ping("hi"), cts.Token); await sut.RunAsync(); @@ -94,8 +97,8 @@ public async Task Given_Void_Handler_And_One_Stage_When_Running_Executor_Then_Ha { var logHandler = Substitute.For>(); List log = []; - IRequestStage[] stages = [new RecordingVoidStage(log)]; - var sut = new VoidStageExecutor(stages, logHandler, new Log("hi"), CancellationToken.None); + object[] stages = [new RecordingVoidStage(log)]; + var sut = LogExecutor(logHandler, stages); NoResult result = await sut.RunAsync(); @@ -108,8 +111,8 @@ public async Task Given_Void_Handler_And_One_Stage_When_Running_Executor_Then_Ha public async Task Given_Stage_That_Calls_Next_Twice_When_Running_Executor_Then_Inner_Chain_Runs_Again() { List log = []; - IRequestStage[] stages = [new DoubleNextStage("outer", log), new RecordingStage("inner", log)]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new DoubleNextStage("outer", log), new RecordingStage("inner", log)]; + var sut = PingExecutor(stages); await sut.RunAsync(); @@ -123,8 +126,8 @@ public async Task Given_Asynchronously_Completing_Handler_When_Stage_Calls_Next_ _pingHandler.HandleAsync(Arg.Any(), Arg.Any()) .Returns(call => YieldThenReturnAsync(call.Arg().Text + ":handled")); List log = []; - IRequestStage[] stages = [new DoubleNextStage("outer", log), new RecordingStage("inner", log)]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new DoubleNextStage("outer", log), new RecordingStage("inner", log)]; + var sut = PingExecutor(stages); await sut.RunAsync(); @@ -140,8 +143,8 @@ public async Task Given_Failing_Handler_When_Outer_Stage_Retries_Then_Inner_Chai ? Task.FromException(new InvalidTimeZoneException("transient")) : Task.FromResult("second")); List log = []; - IRequestStage[] stages = [new RetryOnceStage(log), new RecordingStage("inner", log)]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new RetryOnceStage(log), new RecordingStage("inner", log)]; + var sut = PingExecutor(stages); string result = await sut.RunAsync(); @@ -153,8 +156,8 @@ public async Task Given_Failing_Handler_When_Outer_Stage_Retries_Then_Inner_Chai public async Task Given_Stage_That_Throws_Before_Returning_A_Task_When_Outer_Stage_Retries_Then_It_Runs_Again() { var flaky = new ThrowOnFirstAttemptStage(); - IRequestStage[] stages = [new RetryOnceStage([]), flaky]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new RetryOnceStage([]), flaky]; + var sut = PingExecutor(stages); string result = await sut.RunAsync(); @@ -168,16 +171,52 @@ public async Task Given_Stage_That_Calls_Next_Again_Before_The_First_Call_Comple var pending = new TaskCompletionSource(); _pingHandler.HandleAsync(Arg.Any(), Arg.Any()).Returns(pending.Task); List log = []; - IRequestStage[] stages = [new ConcurrentNextStage(), new RecordingStage("inner", log)]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new ConcurrentNextStage(), new RecordingStage("inner", log)]; + var sut = PingExecutor(stages); - var exception = await Should.ThrowAsync(() => sut.RunAsync()); + var exception = await Should.ThrowAsync(() => sut.RunAsync()); - exception.Message.ShouldContain(nameof(ConcurrentNextStage)); + exception.StageType.ShouldBe(typeof(ConcurrentNextStage)); exception.Message.ShouldContain("still running"); log.ShouldBe(["inner:enter"]); } + // The outermost level and the levels below it hold their guard state in different places, so + // an inner stage is a separate case from the outer one rather than a repeat of it. + [Fact] + public async Task Given_Inner_Stage_That_Calls_Next_Again_Before_The_First_Call_Completes_When_Running_Executor_Then_Throws() + { + var pending = new TaskCompletionSource(); + _pingHandler.HandleAsync(Arg.Any(), Arg.Any()).Returns(pending.Task); + List log = []; + object[] stages = [new RecordingStage("outer", log), new ConcurrentNextStage()]; + var sut = PingExecutor(stages); + + var exception = await Should.ThrowAsync(() => sut.RunAsync()); + + exception.StageType.ShouldBe(typeof(ConcurrentNextStage)); + exception.Message.ShouldContain("still running"); + log.ShouldBe(["outer:enter"]); + } + + // A level keeps its guard state for the whole dispatch, so re-entry does not clear it. A + // stage that walked away from a call still in flight overlaps with itself when an outer + // retry sends it back down. + [Fact] + public async Task Given_Stage_That_Abandoned_A_Pending_Next_Call_When_An_Outer_Stage_Retries_It_Then_Throws() + { + var pending = new TaskCompletionSource(); + _pingHandler.HandleAsync(Arg.Any(), Arg.Any()).Returns(pending.Task); + var abandoning = new AbandonPendingNextStage(); + object[] stages = [new RetryOnceStage([]), abandoning]; + var sut = PingExecutor(stages); + + var exception = await Should.ThrowAsync(() => sut.RunAsync()); + + exception.StageType.ShouldBe(typeof(AbandonPendingNextStage)); + abandoning.Attempts.ShouldBe(2); + } + [Fact] public async Task Given_Stage_That_Calls_Next_From_Two_Threads_At_Once_When_Running_Executor_Then_Exactly_One_Call_Proceeds() { @@ -197,7 +236,38 @@ public async Task Given_Stage_That_Calls_Next_From_Two_Threads_At_Once_When_Runn return gate.Task; }); var stage = new SimultaneousNextStage(gate, () => Interlocked.Increment(ref guardThrows)); - var sut = new TypedStageExecutor([stage], handler, new Ping("hi"), CancellationToken.None); + var sut = PingExecutorFor(handler, stage); + + await sut.RunAsync(); + } + + handlerRuns.ShouldBe(attempts); + guardThrows.ShouldBe(attempts); + } + + [Fact] + public async Task Given_Inner_Stage_That_Calls_Next_From_Two_Threads_At_Once_When_Running_Executor_Then_Exactly_One_Call_Proceeds() + { + const int attempts = 1000; + int handlerRuns = 0; + int guardThrows = 0; + + for (int i = 0; i < attempts; i++) + { + var gate = new TaskCompletionSource(); + var handler = Substitute.For>(); + handler.HandleAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + Interlocked.Increment(ref handlerRuns); + return gate.Task; + }); + IRequestStage[] stages = + [ + new RecordingStage("outer", []), + new SimultaneousNextStage(gate, () => Interlocked.Increment(ref guardThrows)), + ]; + var sut = PingExecutorFor(handler, stages); await sut.RunAsync(); } @@ -206,6 +276,23 @@ public async Task Given_Stage_That_Calls_Next_From_Two_Threads_At_Once_When_Runn guardThrows.ShouldBe(attempts); } + // A void-form stage reaches the same guard state as a two-parameter one, so its second call + // has to be admitted once the first has completed. + [Fact] + public async Task Given_Void_Form_Stage_That_Calls_Next_Twice_When_Running_Void_Executor_Then_Inner_Chain_Runs_Again() + { + var handler = Substitute.For>(); + handler.HandleAsync(Arg.Any(), Arg.Any()).Returns(Task.CompletedTask); + List log = []; + object[] stages = [new DoubleNextVoidStage(log), new RecordingVoidStage(log)]; + var sut = LogExecutor(handler, stages); + + await sut.RunAsync(); + + log.ShouldBe(["void:enter", "enter", "exit", "enter", "exit", "void:exit"]); + await handler.Received(2).HandleAsync(Arg.Any(), Arg.Any()); + } + [Fact] public async Task Given_Void_Form_Stage_That_Calls_Next_From_Two_Threads_At_Once_When_Running_Void_Executor_Then_Exactly_One_Call_Proceeds() { @@ -224,7 +311,38 @@ public async Task Given_Void_Form_Stage_That_Calls_Next_From_Two_Threads_At_Once return gate.Task; }); object[] stages = [new SimultaneousNextVoidStage(gate, () => Interlocked.Increment(ref guardThrows))]; - var sut = new VoidStageExecutor(stages, handler, new Log("hi"), CancellationToken.None); + var sut = LogExecutor(handler, stages); + + await sut.RunAsync(); + } + + handlerRuns.ShouldBe(attempts); + guardThrows.ShouldBe(attempts); + } + + [Fact] + public async Task Given_Inner_Void_Form_Stage_That_Calls_Next_From_Two_Threads_At_Once_When_Running_Void_Executor_Then_Exactly_One_Call_Proceeds() + { + const int attempts = 1000; + int handlerRuns = 0; + int guardThrows = 0; + + for (int i = 0; i < attempts; i++) + { + var gate = new TaskCompletionSource(); + var handler = Substitute.For>(); + handler.HandleAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + Interlocked.Increment(ref handlerRuns); + return gate.Task; + }); + object[] stages = + [ + new VoidFormStage([]), + new SimultaneousNextVoidStage(gate, () => Interlocked.Increment(ref guardThrows)), + ]; + var sut = LogExecutor(handler, stages); await sut.RunAsync(); } @@ -236,32 +354,33 @@ public async Task Given_Void_Form_Stage_That_Calls_Next_From_Two_Threads_At_Once [Fact] public void Given_Stage_That_Returns_A_Null_Task_When_Running_Executor_Then_Throws_Naming_The_Stage() { - IRequestStage[] stages = [new NullTaskStage()]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new NullTaskStage()]; + var sut = PingExecutor(stages); - InvalidOperationException exception = Should.Throw(() => sut.RunAsync()); + StageNullTaskException exception = Should.Throw(() => sut.RunAsync()); - exception.Message.ShouldContain(nameof(NullTaskStage)); - exception.Message.ShouldContain("null task"); + exception.StageType.ShouldBe(typeof(NullTaskStage)); + exception.ShouldBeAssignableTo(); } [Fact] public void Given_Handler_That_Returns_A_Null_Task_When_Running_Executor_Then_Throws_Naming_The_Request() { - var sut = new TypedStageExecutor([], new NilHandler(), new Nil(), CancellationToken.None); + var sut = new TypedStageExecutor( + [], ChainProvider>(new NilHandler(), []), new Nil(), CancellationToken.None); - InvalidOperationException exception = Should.Throw(() => sut.RunAsync()); + HandlerNullTaskException exception = Should.Throw(() => sut.RunAsync()); - exception.Message.ShouldContain(nameof(Nil)); - exception.Message.ShouldContain("null task"); + exception.RequestType.ShouldBe(typeof(Nil)); + exception.ShouldBeAssignableTo(); } [Fact] public async Task Given_Stage_That_Returned_A_Null_Task_When_Outer_Stage_Retries_Then_It_Runs_Again() { var flaky = new NullTaskOnFirstAttemptStage(); - IRequestStage[] stages = [new RetryOnceStage([]), flaky]; - var sut = new TypedStageExecutor(stages, _pingHandler, new Ping("hi"), CancellationToken.None); + object[] stages = [new RetryOnceStage([]), flaky]; + var sut = PingExecutor(stages); string result = await sut.RunAsync(); @@ -275,7 +394,7 @@ public async Task Given_Void_Form_Stage_When_Running_Void_Executor_Then_It_Wraps var logHandler = Substitute.For>(); List log = []; object[] stages = [new VoidFormStage(log)]; - var sut = new VoidStageExecutor(stages, logHandler, new Log("hi"), CancellationToken.None); + var sut = LogExecutor(logHandler, stages); NoResult result = await sut.RunAsync(); @@ -290,7 +409,7 @@ public async Task Given_Void_Form_Stage_That_Awaits_Before_Calling_Next_When_Run var logHandler = Substitute.For>(); List log = []; object[] stages = [new AwaitBeforeNextVoidStage(log)]; - var sut = new VoidStageExecutor(stages, logHandler, new Log("hi"), CancellationToken.None); + var sut = LogExecutor(logHandler, stages); NoResult result = await sut.RunAsync(); @@ -300,12 +419,12 @@ public async Task Given_Void_Form_Stage_That_Awaits_Before_Calling_Next_When_Run } [Fact] - public async Task Given_Both_Stage_Forms_When_Running_Void_Executor_Then_Array_Order_Is_Execution_Order() + public async Task Given_Both_Stage_Forms_When_Running_Void_Executor_Then_Registration_Order_Is_Execution_Order() { var logHandler = Substitute.For>(); List log = []; object[] stages = [new RecordingVoidStage(log), new VoidFormStage(log)]; - var sut = new VoidStageExecutor(stages, logHandler, new Log("hi"), CancellationToken.None); + var sut = LogExecutor(logHandler, stages); await sut.RunAsync(); @@ -317,7 +436,7 @@ public async Task Given_Void_Form_Stage_That_Skips_Next_When_Running_Void_Execut { var logHandler = Substitute.For>(); object[] stages = [new ShortCircuitVoidStage()]; - var sut = new VoidStageExecutor(stages, logHandler, new Log("hi"), CancellationToken.None); + var sut = LogExecutor(logHandler, stages); await sut.RunAsync(); @@ -329,23 +448,22 @@ public void Given_Void_Form_Stage_That_Returns_A_Null_Task_When_Running_Void_Exe { var logHandler = Substitute.For>(); object[] stages = [new NullTaskVoidStage()]; - var sut = new VoidStageExecutor(stages, logHandler, new Log("hi"), CancellationToken.None); + var sut = LogExecutor(logHandler, stages); - InvalidOperationException exception = Should.Throw(() => sut.RunAsync()); + StageNullTaskException exception = Should.Throw(() => sut.RunAsync()); - exception.Message.ShouldContain(nameof(NullTaskVoidStage)); - exception.Message.ShouldContain("null task"); + exception.StageType.ShouldBe(typeof(NullTaskVoidStage)); } [Fact] public void Given_Void_Handler_That_Returns_A_Null_Task_When_Running_Void_Executor_Then_Throws_Naming_The_Request() { - var sut = new VoidStageExecutor([], new SilentHandler(), new Silent(), CancellationToken.None); + var sut = new VoidStageExecutor( + [], [], ChainProvider>(new SilentHandler(), []), new Silent(), CancellationToken.None); - InvalidOperationException exception = Should.Throw(() => sut.RunAsync()); + HandlerNullTaskException exception = Should.Throw(() => sut.RunAsync()); - exception.Message.ShouldContain(nameof(Silent)); - exception.Message.ShouldContain("null task"); + exception.RequestType.ShouldBe(typeof(Silent)); } [Fact] @@ -357,6 +475,99 @@ public async Task Given_Synchronously_Completed_Task_When_Bridging_To_No_Result_ await result; } + // A one-stage chain is the boundary case: the outermost stage's next reaches the handler + // with no level in between, so it exercises the executor's own guard state rather than a + // continuation's. + [Fact] + public async Task Given_One_Stage_When_Running_Executor_Then_Stage_Wraps_The_Handler() + { + List log = []; + var sut = PingExecutor(new RecordingStage("only", log)); + + string result = await sut.RunAsync(); + + result.ShouldBe("hi:handled"); + log.ShouldBe(["only:enter", "only:exit"]); + } + + [Fact] + public async Task Given_One_Stage_That_Calls_Next_Twice_When_Running_Executor_Then_Handler_Runs_Again() + { + List log = []; + var sut = PingExecutor(new DoubleNextStage("only", log)); + + await sut.RunAsync(); + + log.ShouldBe(["only:enter", "only:exit"]); + await _pingHandler.Received(2).HandleAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task Given_One_Stage_That_Calls_Next_Again_Before_The_First_Call_Completes_When_Running_Executor_Then_Throws() + { + var pending = new TaskCompletionSource(); + _pingHandler.HandleAsync(Arg.Any(), Arg.Any()).Returns(pending.Task); + var sut = PingExecutor(new ConcurrentNextStage()); + + var exception = await Should.ThrowAsync(() => sut.RunAsync()); + + exception.StageType.ShouldBe(typeof(ConcurrentNextStage)); + exception.Message.ShouldContain("still running"); + } + + [Fact] + public void Given_One_Stage_That_Returns_A_Null_Task_When_Running_Executor_Then_Throws_Naming_The_Stage() + { + var sut = PingExecutor(new NullTaskStage()); + + StageNullTaskException exception = Should.Throw(() => sut.RunAsync()); + + exception.StageType.ShouldBe(typeof(NullTaskStage)); + } + + [Fact] + public void Given_One_Stage_And_Handler_That_Returns_A_Null_Task_When_Running_Executor_Then_Throws_Naming_The_Request() + { + object[] stages = [new NilPassThroughStage()]; + var sut = new TypedStageExecutor( + StageTypes(stages), + ChainProvider>(new NilHandler(), stages), + new Nil(), + CancellationToken.None); + + HandlerNullTaskException exception = Should.Throw(() => sut.RunAsync()); + + exception.RequestType.ShouldBe(typeof(Nil)); + } + + [Fact] + public async Task Given_One_Typed_Form_Stage_When_Running_Void_Executor_Then_It_Wraps_The_Handler() + { + var logHandler = Substitute.For>(); + List log = []; + var sut = LogExecutor(logHandler, new RecordingVoidStage(log)); + + NoResult result = await sut.RunAsync(); + + result.ShouldBe(NoResult.Value); + log.ShouldBe(["enter", "exit"]); + await logHandler.Received(1).HandleAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task Given_One_Void_Form_Stage_That_Calls_Next_Twice_When_Running_Void_Executor_Then_Handler_Runs_Again() + { + var handler = Substitute.For>(); + handler.HandleAsync(Arg.Any(), Arg.Any()).Returns(Task.CompletedTask); + List log = []; + var sut = LogExecutor(handler, new DoubleNextVoidStage(log)); + + await sut.RunAsync(); + + log.ShouldBe(["void:enter", "void:exit"]); + await handler.Received(2).HandleAsync(Arg.Any(), Arg.Any()); + } + #region Initialization private readonly IRequestHandler _pingHandler; @@ -372,6 +583,69 @@ public StageExecutorTests() #region Helpers + // A chain resolves each stage from DI by its registered type, so a test chain gives every + // level its own stage type, the same way validation forces a real chain to. The handler goes + // in the same provider, because the bottom level resolves it there too. + private static ServiceProvider ChainProvider(THandler handler, object[] stages) + where THandler : class + { + var services = new ServiceCollection(); + services.AddSingleton(handler); + foreach (object stage in stages) + { + services.AddSingleton(stage.GetType(), stage); + } + + return services.BuildServiceProvider(); + } + + private static Type[] StageTypes(object[] stages) + { + Type[] types = new Type[stages.Length]; + for (int i = 0; i < stages.Length; i++) + { + types[i] = stages[i].GetType(); + } + + return types; + } + + // The freeze settles which contract shape each void level runs under; these tests stand in + // for it by reading the shape off the instances they were handed. + private static bool[] TypedShapes(object[] stages) + where TRequest : IRequest + { + bool[] shapes = new bool[stages.Length]; + for (int i = 0; i < stages.Length; i++) + { + shapes[i] = stages[i] is IRequestStage; + } + + return shapes; + } + + private TypedStageExecutor PingExecutor(params object[] stages) + => PingExecutorFor(_pingHandler, stages); + + private static TypedStageExecutor PingExecutorFor( + IRequestHandler handler, params object[] stages) + => new(StageTypes(stages), ChainProvider(handler, stages), new Ping("hi"), CancellationToken.None); + + private static VoidStageExecutor LogExecutor(IRequestHandler handler, params object[] stages) + => new( + StageTypes(stages), + TypedShapes(stages), + ChainProvider(handler, stages), + new Log("hi"), + CancellationToken.None); + + // Position markers, so two levels of the same stage class are two registrable types. + private sealed class Outer + { } + + private sealed class Inner + { } + // Public so NSubstitute can proxy handler interfaces closed over these types. public sealed record Ping(string Text) : IRequest; @@ -410,12 +684,12 @@ public Task HandleAsync(Silent request, CancellationToken cancellationToken) => null!; } - private sealed class RecordingStage(string name, List log) : IRequestStage + private sealed class RecordingStage(string name, List log) : IRequestStage { - public async Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { log.Add($"{name}:enter"); - string response = await next(); + string response = await next.InvokeAsync(); log.Add($"{name}:exit"); return response; } @@ -423,10 +697,10 @@ public async Task HandleAsync(Ping request, StageDelegate next, private sealed class RecordingVoidStage(List log) : IRequestStage { - public async Task HandleAsync(Log request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Log request, IContinuation next, CancellationToken cancellationToken) { log.Add("enter"); - NoResult response = await next(); + NoResult response = await next.InvokeAsync(); log.Add("exit"); return response; } @@ -434,58 +708,84 @@ public async Task HandleAsync(Log request, StageDelegate nex private sealed class ShortCircuitStage(string response) : IRequestStage { - public Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) => Task.FromResult(response); } private sealed class DoubleNextStage(string name, List log) : IRequestStage { - public async Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { log.Add($"{name}:enter"); - await next(); - string response = await next(); + await next.InvokeAsync(); + string response = await next.InvokeAsync(); log.Add($"{name}:exit"); return response; } } + private sealed class DoubleNextVoidStage(List log) : IRequestStage + { + public async Task HandleAsync(Log request, IContinuation next, CancellationToken cancellationToken) + { + log.Add("void:enter"); + await next.InvokeAsync(); + await next.InvokeAsync(); + log.Add("void:exit"); + } + } + private sealed class RetryOnceStage(List log) : IRequestStage { - public async Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { log.Add("retry:attempt"); try { - return await next(); + return await next.InvokeAsync(); } catch (Exception) { log.Add("retry:attempt"); - return await next(); + return await next.InvokeAsync(); } } } + // The timeout shape: it starts the rest of the chain, gives up on it, and leaves that call in + // flight rather than awaiting it out. + private sealed class AbandonPendingNextStage : IRequestStage + { + public int Attempts { get; private set; } + + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) + { + Attempts++; + _ = next.InvokeAsync(); + + return Task.FromException(new TimeoutException("gave up")); + } + } + private sealed class ThrowOnFirstAttemptStage : IRequestStage { public int Attempts { get; private set; } - public Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { Attempts++; - return Attempts == 1 ? throw new InvalidOperationException("sync boom") : next(); + return Attempts == 1 ? throw new InvalidOperationException("sync boom") : next.InvokeAsync(); } } // Suspends on work of its own before delegating, the shape of a validation or caching stage. - private sealed class AwaitBeforeNextStage(string name, List log) : IRequestStage + private sealed class AwaitBeforeNextStage(string name, List log) : IRequestStage { - public async Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { await Task.Yield(); log.Add($"{name}:enter"); - string response = await next(); + string response = await next.InvokeAsync(); log.Add($"{name}:exit"); return response; } @@ -493,11 +793,11 @@ public async Task HandleAsync(Ping request, StageDelegate next, private sealed class AwaitBeforeNextVoidStage(List log) : IRequestStage { - public async Task HandleAsync(Log request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Log request, IContinuation next, CancellationToken cancellationToken) { await Task.Yield(); log.Add("void:enter"); - await next(); + await next.InvokeAsync(); log.Add("void:exit"); } } @@ -505,10 +805,10 @@ public async Task HandleAsync(Log request, StageDelegate next, CancellationToken // Starts a second walk of the chain while the first is still suspended on the handler. private sealed class ConcurrentNextStage : IRequestStage { - public async Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { - Task first = next(); - Task second = next(); + Task first = next.InvokeAsync(); + Task second = next.InvokeAsync(); return await first.ConfigureAwait(false) + await second.ConfigureAwait(false); } @@ -520,7 +820,7 @@ public async Task HandleAsync(Ping request, StageDelegate next, private sealed class SimultaneousNextStage(TaskCompletionSource gate, Action onGuardThrow) : IRequestStage { - public async Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { using var barrier = new Barrier(2); Task?[] calls = new Task?[2]; @@ -530,9 +830,9 @@ Task Caller(int slot) => Task.Run(() => barrier.SignalAndWait(); try { - calls[slot] = next(); + calls[slot] = next.InvokeAsync(); } - catch (InvalidOperationException) + catch (OverlappingNextCallException) { onGuardThrow(); } @@ -555,7 +855,7 @@ Task Caller(int slot) => Task.Run(() => private sealed class SimultaneousNextVoidStage(TaskCompletionSource gate, Action onGuardThrow) : IRequestStage { - public async Task HandleAsync(Log request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Log request, IContinuation next, CancellationToken cancellationToken) { using var barrier = new Barrier(2); Task?[] calls = new Task?[2]; @@ -565,9 +865,9 @@ Task Caller(int slot) => Task.Run(() => barrier.SignalAndWait(); try { - calls[slot] = next(); + calls[slot] = next.InvokeAsync(); } - catch (InvalidOperationException) + catch (OverlappingNextCallException) { onGuardThrow(); } @@ -586,40 +886,47 @@ Task Caller(int slot) => Task.Run(() => private sealed class NullTaskStage : IRequestStage { - public Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) => null!; } + // Delegates straight to next, so the null task the handler returns is the one reported. + private sealed class NilPassThroughStage : IRequestStage + { + public Task HandleAsync(Nil request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); + } + private sealed class NullTaskOnFirstAttemptStage : IRequestStage { public int Attempts { get; private set; } - public Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { Attempts++; - return Attempts == 1 ? null! : next(); + return Attempts == 1 ? null! : next.InvokeAsync(); } } private sealed class VoidFormStage(List log) : IRequestStage { - public async Task HandleAsync(Log request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(Log request, IContinuation next, CancellationToken cancellationToken) { log.Add("void:enter"); - await next(); + await next.InvokeAsync(); log.Add("void:exit"); } } private sealed class ShortCircuitVoidStage : IRequestStage { - public Task HandleAsync(Log request, StageDelegate next, CancellationToken cancellationToken) + public Task HandleAsync(Log request, IContinuation next, CancellationToken cancellationToken) => Task.CompletedTask; } private sealed class NullTaskVoidStage : IRequestStage { - public Task HandleAsync(Log request, StageDelegate next, CancellationToken cancellationToken) + public Task HandleAsync(Log request, IContinuation next, CancellationToken cancellationToken) => null!; } @@ -633,10 +940,10 @@ private sealed class TokenCapturingStage : IRequestStage { public CancellationToken CapturedToken { get; private set; } - public Task HandleAsync(Ping request, StageDelegate next, CancellationToken cancellationToken) + public Task HandleAsync(Ping request, IContinuation next, CancellationToken cancellationToken) { CapturedToken = cancellationToken; - return next(); + return next.InvokeAsync(); } } diff --git a/tests/RequestFlow.Tests.Unit/Stages/StageLifetimeTests.cs b/tests/RequestFlow.Tests.Unit/Stages/StageLifetimeTests.cs index a258cf4..232e1bd 100644 --- a/tests/RequestFlow.Tests.Unit/Stages/StageLifetimeTests.cs +++ b/tests/RequestFlow.Tests.Unit/Stages/StageLifetimeTests.cs @@ -48,6 +48,88 @@ public async Task Given_Transient_Stage_When_Sending_Twice_In_One_Scope_Then_A_N StageMarkers[0].ShouldBeSameAs(StageMarkers[1]); } + [Fact] + public async Task Given_Stage_That_Skips_Next_When_Sending_Request_Then_The_Stage_Below_Is_Never_Constructed() + { + ServiceProvider provider = BuildTraceChain(typeof(SkipNextStage)); + + await SendTraceAsync(provider); + + CountingStageConstructions.ShouldBe(0); + } + + [Fact] + public async Task Given_Stage_That_Calls_Next_Twice_When_Sending_Request_Then_The_Stage_Below_Is_Constructed_Once() + { + ServiceProvider provider = BuildTraceChain(typeof(DoubleNextStage)); + + await SendTraceAsync(provider); + + CountingStageConstructions.ShouldBe(1); + } + + [Fact] + public async Task Given_Stage_That_Skips_Next_When_Sending_Request_Then_The_Handler_Is_Never_Constructed() + { + ServiceProvider provider = BuildTraceChain(typeof(SkipNextStage)); + + await SendTraceAsync(provider); + + TraceHandlerConstructions.ShouldBe(0); + } + + [Fact] + public async Task Given_Stage_That_Calls_Next_Twice_When_Sending_Request_Then_The_Handler_Is_Constructed_Once() + { + ServiceProvider provider = BuildTraceChain(typeof(DoubleNextStage)); + + await SendTraceAsync(provider); + + TraceHandlerConstructions.ShouldBe(1); + } + + [Fact] + public async Task Given_Void_Stage_That_Skips_Next_When_Sending_Request_Then_The_Handler_Is_Never_Constructed() + { + ServiceProvider provider = BuildVoidTraceChain(typeof(SkipNextVoidStage)); + + await SendVoidTraceAsync(provider); + + VoidTraceHandlerConstructions.ShouldBe(0); + } + + [Fact] + public async Task Given_Void_Stage_That_Calls_Next_Twice_When_Sending_Request_Then_The_Handler_Is_Constructed_Once() + { + ServiceProvider provider = BuildVoidTraceChain(typeof(DoubleNextVoidStage)); + + await SendVoidTraceAsync(provider); + + VoidTraceHandlerConstructions.ShouldBe(1); + } + + // Resolving a level lazily puts the container's failure inside the chain, where the stages + // above it can see it, instead of ahead of it where it escaped SendAsync untouched. + [Fact] + public async Task Given_Stage_That_Cannot_Be_Constructed_When_An_Outer_Stage_Wraps_It_Then_That_Stage_Observes_The_Failure() + { + ServiceProvider provider = BuildStageChain(typeof(CatchingStage), typeof(UnbuildableStage)); + + string result = await SendTraceForResultAsync(provider); + + result.ShouldBe("caught"); + } + + [Fact] + public async Task Given_Handler_That_Cannot_Be_Constructed_When_A_Stage_Wraps_It_Then_That_Stage_Observes_The_Failure() + { + ServiceProvider provider = BuildStageChain(typeof(CatchingUnbuildableStage)); + + string result = await SendUnbuildableForResultAsync(provider); + + result.ShouldBe("caught"); + } + [Fact] public void Given_Consumer_Registered_Stage_When_Adding_Request_Flow_Then_The_Consumer_Registration_Is_Kept() { @@ -72,12 +154,18 @@ public void Given_Consumer_Registered_Stage_When_Adding_Request_Flow_Then_The_Co private static readonly List StageMarkers = []; private static readonly List HandlerMarkers = []; private static readonly List StageInstances = []; + private static int CountingStageConstructions; + private static int TraceHandlerConstructions; + private static int VoidTraceHandlerConstructions; public StageLifetimeTests() { StageMarkers.Clear(); HandlerMarkers.Clear(); StageInstances.Clear(); + CountingStageConstructions = 0; + TraceHandlerConstructions = 0; + VoidTraceHandlerConstructions = 0; } #endregion @@ -116,6 +204,50 @@ private static async Task SendTwiceInOneScopeAsync(ServiceProvider provider) await dispatcher.SendAsync(new Probe()); } + private static ServiceProvider BuildStageChain(params Type[] stageTypes) + { + var services = new ServiceCollection(); + services.AddScoped(); + services.AddRequestFlow(o => + { + o.RegisterHandlersFromAssemblyContaining(); + foreach (Type stageType in stageTypes) + { + o.AddStage(stageType); + } + }); + + return services.BuildServiceProvider(); + } + + private static ServiceProvider BuildTraceChain(Type outerStageType) + => BuildStageChain(outerStageType, typeof(CountingStage)); + + private static async Task SendTraceAsync(ServiceProvider provider) + => await SendTraceForResultAsync(provider); + + private static async Task SendTraceForResultAsync(ServiceProvider provider) + { + using IServiceScope scope = provider.CreateScope(); + + return await scope.ServiceProvider.GetRequiredService().SendAsync(new Trace()); + } + + private static async Task SendUnbuildableForResultAsync(ServiceProvider provider) + { + using IServiceScope scope = provider.CreateScope(); + + return await scope.ServiceProvider.GetRequiredService().SendAsync(new Unbuildable()); + } + + private static ServiceProvider BuildVoidTraceChain(Type stageType) => BuildStageChain(stageType); + + private static async Task SendVoidTraceAsync(ServiceProvider provider) + { + using IServiceScope scope = provider.CreateScope(); + await scope.ServiceProvider.GetRequiredService().SendAsync(new VoidTrace()); + } + public sealed class ScopeMarker { } @@ -134,12 +266,130 @@ public sealed class MarkerStage(ScopeMarker marker) : IRequ where TRequest : IRequest { public Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { StageMarkers.Add(marker); StageInstances.Add(this); - return next(); + return next.InvokeAsync(); + } + } + + public sealed record Trace : IRequest; + + public sealed class TraceHandler : IRequestHandler + { + public TraceHandler() + => TraceHandlerConstructions++; + + public Task HandleAsync(Trace request, CancellationToken cancellationToken) + => Task.FromResult("traced"); + } + + public sealed record VoidTrace : IRequest; + + public sealed class VoidTraceHandler : IRequestHandler + { + public VoidTraceHandler() + => VoidTraceHandlerConstructions++; + + public Task HandleAsync(VoidTrace request, CancellationToken cancellationToken) + => Task.CompletedTask; + } + + public sealed class SkipNextVoidStage : IRequestStage + { + public Task HandleAsync(VoidTrace request, IContinuation next, CancellationToken cancellationToken) + => Task.CompletedTask; + } + + public sealed class DoubleNextVoidStage : IRequestStage + { + public async Task HandleAsync(VoidTrace request, IContinuation next, CancellationToken cancellationToken) + { + await next.InvokeAsync(); + await next.InvokeAsync(); + } + } + + public sealed class SkipNextStage : IRequestStage + { + public Task HandleAsync(Trace request, IContinuation next, CancellationToken cancellationToken) + => Task.FromResult("short-circuited"); + } + + public sealed class DoubleNextStage : IRequestStage + { + public async Task HandleAsync( + Trace request, IContinuation next, CancellationToken cancellationToken) + { + await next.InvokeAsync(); + return await next.InvokeAsync(); + } + } + + public sealed class CountingStage : IRequestStage + { + public CountingStage() + => CountingStageConstructions++; + + public Task HandleAsync(Trace request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); + } + + // Nothing registers this, so any level that asks the container for it fails to build. + public sealed class MissingDependency + { } + + public sealed class UnbuildableStage : IRequestStage + { + public UnbuildableStage(MissingDependency dependency) + { } + + public Task HandleAsync(Trace request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); + } + + public sealed class CatchingStage : IRequestStage + { + public async Task HandleAsync( + Trace request, IContinuation next, CancellationToken cancellationToken) + { + try + { + return await next.InvokeAsync(); + } + catch (InvalidOperationException) + { + return "caught"; + } + } + } + + public sealed record Unbuildable : IRequest; + + public sealed class UnbuildableHandler : IRequestHandler + { + public UnbuildableHandler(MissingDependency dependency) + { } + + public Task HandleAsync(Unbuildable request, CancellationToken cancellationToken) + => Task.FromResult("unreachable"); + } + + public sealed class CatchingUnbuildableStage : IRequestStage + { + public async Task HandleAsync( + Unbuildable request, IContinuation next, CancellationToken cancellationToken) + { + try + { + return await next.InvokeAsync(); + } + catch (InvalidOperationException) + { + return "caught"; + } } } diff --git a/tests/RequestFlow.Tests.Unit/Stages/StagePipelineTests.cs b/tests/RequestFlow.Tests.Unit/Stages/StagePipelineTests.cs index b414972..96a4308 100644 --- a/tests/RequestFlow.Tests.Unit/Stages/StagePipelineTests.cs +++ b/tests/RequestFlow.Tests.Unit/Stages/StagePipelineTests.cs @@ -316,10 +316,10 @@ public sealed class RecordingStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("Recording:enter"); - TResponse response = await next(); + TResponse response = await next.InvokeAsync(); Trace.Add("Recording:exit"); return response; } @@ -329,10 +329,10 @@ public sealed class SecondStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("Second:enter"); - TResponse response = await next(); + TResponse response = await next.InvokeAsync(); Trace.Add("Second:exit"); return response; } @@ -342,10 +342,10 @@ public sealed class TaggedOnlyStage : IRequestStage, ITag { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("TaggedOnly:enter"); - TResponse response = await next(); + TResponse response = await next.InvokeAsync(); Trace.Add("TaggedOnly:exit"); return response; } @@ -355,10 +355,10 @@ public sealed class UnreachableStage : IRequestStage, INothingImplementsThis { public Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("Unreachable:enter"); - return next(); + return next.InvokeAsync(); } } @@ -367,10 +367,10 @@ public sealed class ResponseBoundStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("ResponseBound:enter"); - string response = await next(); + string response = await next.InvokeAsync(); Trace.Add("ResponseBound:exit"); return response; } @@ -379,10 +379,10 @@ public async Task HandleAsync( public sealed class VoidRecordingStage : IRequestStage where TRequest : IRequest { - public async Task HandleAsync(TRequest request, StageDelegate next, CancellationToken cancellationToken) + public async Task HandleAsync(TRequest request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("VoidRecording:enter"); - await next(); + await next.InvokeAsync(); Trace.Add("VoidRecording:exit"); } } @@ -392,10 +392,10 @@ public async Task HandleAsync(TRequest request, StageDelegate next, Cancellation public sealed class NotificationStage : IRequestStage { public async Task HandleAsync( - Notification request, StageDelegate next, CancellationToken cancellationToken) + Notification request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("Notification:enter"); - string response = await next(); + string response = await next.InvokeAsync(); Trace.Add("Notification:exit"); return response; } @@ -404,10 +404,10 @@ public async Task HandleAsync( public sealed class PingOnlyStage : IRequestStage { public async Task HandleAsync( - Ping request, StageDelegate next, CancellationToken cancellationToken) + Ping request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("PingOnly:enter"); - string response = await next(); + string response = await next.InvokeAsync(); Trace.Add("PingOnly:exit"); return response; } diff --git a/tests/RequestFlow.Tests.Unit/Stages/StageSemanticsTests.cs b/tests/RequestFlow.Tests.Unit/Stages/StageSemanticsTests.cs index 0e67a72..1d993d7 100644 --- a/tests/RequestFlow.Tests.Unit/Stages/StageSemanticsTests.cs +++ b/tests/RequestFlow.Tests.Unit/Stages/StageSemanticsTests.cs @@ -125,6 +125,18 @@ public async Task Given_Stage_On_A_Void_Request_When_Sending_Request_Then_Return await task; } + // Which shape a void request's stage runs under is settled once, when the dispatch map + // freezes, so a stage that implements both has to land on the same one every dispatch. + [Fact] + public async Task Given_Stage_Implementing_Both_Contract_Shapes_When_Sending_Void_Request_Then_The_Typed_Shape_Runs() + { + IRequestDispatcher dispatcher = Build(o => o.AddStage(typeof(BothShapesStage))); + + await dispatcher.SendAsync(new Wipe()); + + Trace.ShouldBe(["BothShapes:typed"]); + } + #region Initialization // The container instantiates stages, so the trace and counters have to be static; the @@ -222,9 +234,9 @@ public sealed class ThrowAfterNextStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { - await next(); + await next.InvokeAsync(); throw new TimeoutException("after next"); } } @@ -233,7 +245,7 @@ public sealed class ThrowingStage : IRequestStage { public Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) => throw new InvalidTimeZoneException("from stage"); } @@ -241,28 +253,28 @@ public sealed class PassThroughStage : IRequestStage { public Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) - => next(); + TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } public sealed class SecondPassThroughStage : IRequestStage where TRequest : IRequest { public Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) - => next(); + TRequest request, IContinuation next, CancellationToken cancellationToken) + => next.InvokeAsync(); } public sealed class TokenForwardingStage : IRequestStage where TRequest : IRequest { public Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { if (cancellationToken.IsCancellationRequested) Trace.Add("Token:cancelled"); - return next(); + return next.InvokeAsync(); } } @@ -271,11 +283,11 @@ public sealed class AwaitBeforeNextStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { await Task.Yield(); Trace.Add("Await:enter"); - TResponse response = await next(); + TResponse response = await next.InvokeAsync(); Trace.Add("Await:exit"); return response; } @@ -285,10 +297,10 @@ public sealed class TracingStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { Trace.Add("Tracing:enter"); - TResponse response = await next(); + TResponse response = await next.InvokeAsync(); Trace.Add("Tracing:exit"); return response; } @@ -298,10 +310,27 @@ public sealed class DoubleNextStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) + { + await next.InvokeAsync(); + return await next.InvokeAsync(); + } + } + + // Implements the two-parameter and the void contract at once, which only a void request can + // offer both of. Each records which one ran. + public sealed class BothShapesStage : IRequestStage, IRequestStage + { + public Task HandleAsync(Wipe request, IContinuation next, CancellationToken cancellationToken) + { + Trace.Add("BothShapes:typed"); + return next.InvokeAsync(); + } + + public Task HandleAsync(Wipe request, IContinuation next, CancellationToken cancellationToken) { - await next(); - return await next(); + Trace.Add("BothShapes:void"); + return next.InvokeAsync(); } } @@ -309,15 +338,15 @@ public sealed class RetryOnceStage : IRequestStage { public async Task HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { try { - return await next(); + return await next.InvokeAsync(); } catch (InvalidOperationException) { - return await next(); + return await next.InvokeAsync(); } } } @@ -328,10 +357,10 @@ public sealed class CountingStage : IRequestStage HandleAsync( - TRequest request, StageDelegate next, CancellationToken cancellationToken) + TRequest request, IContinuation next, CancellationToken cancellationToken) { Entries++; - return next(); + return next.InvokeAsync(); } }