Repository navigation
Add MediatR-compatible streaming requests - #4
Merged
Merged
Conversation
IStreamRequest<T>, IStreamRequestHandler<,>, IStreamPipelineBehavior<,>, StreamHandlerDelegate<T> and ISender.CreateStream (typed and untyped), with the same shapes as MediatR so migrating code compiles unchanged. - Assembly scanning picks up stream handlers; AddOpenStreamBehavior, AddStreamBehavior<T>() and AddStreamBehavior(Type) register stream behaviors, first added outermost, deduplicated like pipeline behaviors. - Lazy: nothing is resolved until enumeration, and the handler is resolved inside the pipeline (1.3 semantics). - The CreateStream and WithCancellation tokens are combined and reach every behavior and the handler; the handler's stream is always enumerated with them, even behind a behavior that drops the token. - Scope per dispatch: one scope per enumeration, disposed when the stream ends, throws or is abandoned; re-established before every step so the handler's own sends join it after a yield. - Telemetry: a CreateStream span, current for the handler at every step but not for the consumer, and switchboard.stream.duration. - ValidateSwitchboard covers missing and duplicate stream handlers. - A covariant CreateStream runs the declared handler instead of failing and poisoning the wrapper cache (a MediatR bug). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Qh4oHys2D2TLycvt3yKVtX
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes the most visible gap in the migration table: Streaming (
IStreamRequest<>) ❌ → ✅ identical.API (namespace
Switchboard, shapes checked by reflection against MediatR 12.5 and 14.2)IStreamRequest<out TResponse>IBaseRequestIStreamRequestHandler<in TRequest, out TResponse>IAsyncEnumerable<TResponse> Handle(TRequest, CancellationToken),where TRequest : IStreamRequest<TResponse>IStreamPipelineBehavior<in TRequest, TResponse>IAsyncEnumerable<TResponse> Handle(TRequest, StreamHandlerDelegate<TResponse> next, CancellationToken)StreamHandlerDelegate<out TResponse>IAsyncEnumerable<TResponse>(). No token parameter, as in MediatR (unlikeRequestHandlerDelegate), so() => ...lambdas in migrated behavior tests still compileISender.CreateStream<TResponse>(IStreamRequest<TResponse>, CancellationToken = default)IAsyncEnumerable<TResponse>ISender.CreateStream(object, CancellationToken = default)IAsyncEnumerable<object?>SwitchboardConfiguration.AddOpenStreamBehavior(Type),AddStreamBehavior<TBehavior>(),AddStreamBehavior(Type)AddOpenBehavior/AddBehaviorAssembly scanning in
AddSwitchboardnow picks upIStreamRequestHandler<,>.Semantics
CreateStreamchecks its argument and returns. Behaviors and the handler are resolved by the firstMoveNextAsync. Each enumeration dispatches the request again. A null request, or an untyped call with a non-stream object, throws at the call (ArgumentNullException/ArgumentException), as MediatR does.next(). A behavior that answers the stream itself never constructs the handler. A missing handler fails on the firstMoveNextAsync, where behaviors can see it.AddSwitchboardcalls (TryAddEnumerable). Everything is transient.CreateStreamtoken and the consumer'sWithCancellationtoken are combined ([EnumeratorCancellation]on the outermost iterator), and the result is passed to every behavior and to the handler. The handler's stream is also always enumerated with that token. So a handler that returns a query (EF CoreAsAsyncEnumerable()), which only sees the enumerator token, is still cancelled behind a behavior that runsawait foreach (var x in next())with no token. A token a behavior passes vianext().WithCancellation(t)is combined in as well.MoveNextAsyncand disposed when the stream completes, throws, or is disposed early (break). A stream that is never enumerated never creates a scope. Subtle point: an async iterator resumes on the consumer'sExecutionContextafter everyyield. So theAsyncLocaldispatch scope is set again before each step; without that, a handler's injectedISenderwould open a fresh scope after its firstyieldinstead of joining. A test covers this, and it fails if the re-set is removed.CreateStream {Request}span taggedswitchboard.requestandswitchboard.stream.items(number of items yielded).Send: statusError,error.type, anexceptionevent, and cancellations are not marked as errors.switchboard.stream.durationhistogram. It is separate fromswitchboard.request.durationso long-lived streams don't skew request latency.Activity.Currentagain before every step, for the sameExecutionContextreason. Work the handler does between items is therefore traced as its child, while the consumer's code between items is not. Two tests cover this.ValidateSwitchboard()reports stream requests with no handler (OrphanStream has no handler. Add a class implementing IStreamRequestHandler<OrphanStream, Int32>.) and stream requests with two different handlers, under the existing options.Deliberate differences from MediatR
CreateStreamworks (a fix). TakeIStreamRequest<object> r = new GetOrderLines(), where the request declaresIStreamRequest<OrderLine>. Here,CreateStream(r)runs theOrderLinehandler. MediatR 12.5 looks forIStreamRequestHandler<GetOrderLines, object>and throws. It also caches that wrapper, so every later correctCreateStream(new GetOrderLines())throwsInvalidCastException. I checked this against the real MediatR 12.5 package. It is the same bug 1.2.2 fixed forSend. No item conversion is needed, becauseIAsyncEnumerable<out T>is covariant too.ServiceLifetime/ service+implementation overloads onAddOpenStreamBehavior/AddStreamBehavior. This mirrors Switchboard's existingAddOpenBehavior/AddBehavior, which already omit them. Behaviors are always transient.(I checked one more candidate difference and dropped it. MediatR also enumerates the handler with the token behind a token-dropping behavior, so that part is parity.)
Things for you to decide
ISendergained two members, and so didIMediator. Anyone who implementsISenderthemselves, such as theScopingSenderdecorator the README's 1.2 upgrade guide mentions, gets a compile error until they add both overloads. MediatR has them onISender, so this is needed for drop-in compatibility. The CHANGELOG calls it out under "Behavior notes". Default interface implementations would avoid the break, but MediatR doesn't use them and they'd make a broken stream silent, so I didn't.src/. By a strict count the library was already right at ~500 and is now ~650. I left the claim alone because it's your marketing copy, but it probably needs updating.<Version>change and no "New in 1.4" banner. The CHANGELOG has an[Unreleased]section that says the next release is a minor one (1.4.0).Tests
There are 39 new tests in four new files. Existing test files are untouched.
StreamTests.cs:IMediator, covariant (plus no cache poisoning)CreateStreamtoken and viaWithCancellation; both tokens reach behaviors and the handler; enumerator-token-only handler behind a token-dropping behaviorStreamScopePerDispatchTests.cs:breakand on a throwSendandCreateStreamafter ayieldjoin the scopeStreamRegistrationTests.cs:AddSwitchboardcalls, argument validationStreamTelemetryTests.cs(in the telemetry collection):Error; duration histogram witherror.typeI mutation-checked the three hard-to-see guarantees (the scope re-set, the
Activity.Currentre-set, and the handler enumerated with the token): removing each one makes its test fail.dotnet build -c Release(0 warnings),dotnet test -c Release(131/131 on net8.0, net9.0 and net10.0, 5 consecutive runs), anddotnet packwith package validation all pass locally.Generated by Claude Code