Repository navigation
v7: Handler-based API and awaitable RequestAsync alongside Rx #87
Description
Today: the only way to consume messages is IObservable<T> (Subscribe, OfType, ObserveOn), and requests return an observable from GetResponses. That's powerful, but easy to misuse: subscriptions get dropped or never disposed, and replies get sent from the wrong place. Async methods take no CancellationToken.
Proposal:
- Handlers:
bus.Handle<PlaceOrder>(async (cmd, ctx, ct) => ...), andICommandHandler<T>,IEventHandler<T>andIRequestHandler<TReq, TResp>registered through DI. - Requests:
await bus.RequestAsync<GetQuote, Quote>(request, timeout, ct). - Cancellation: a
CancellationTokenon every async method. - Streams:
IAsyncEnumerable<T>views of message streams. - Observable-first, not observable-only:
IObservablestays in the core as the engine. Handlers,RequestAsyncandIAsyncEnumerableare thin adapters over it, and Rx stays first class for composing streams.
Concurrency control (from #37 by @fernando-almeida): each handler has a MaxConcurrency option, defaulting to processing one message at a time per handler, which is what Rx subscriptions do today. This gives the control #37 wanted from passing an IScheduler to IMessageSource, without changing the interface every transport implements. Rx users can still ObserveOn their own scheduler.
Done when: the examples and the new testing kit use handlers and RequestAsync, and all public async methods take a CancellationToken, and a test shows MaxConcurrency limits how many messages a handler processes at once.
Activity
Handlers, RequestAsync and MaxConcurrency (#114)
christopherread commentedon Oct 26, 2026
Status on the
v7branch.Merged
HandleCommand,HandleEventandHandleRequest;ICommandHandler<T>,IEventHandler<T>andIRequestHandler<TReq, TResp>;MessageContext, withPublishAsync,SendAsyncand replies, each taking aCancellationToken;HandlerOptions.MaxConcurrency, which defaults to one at a time, in order, and has a test showing it limits concurrency;RequestAsync<TReq, TResp>(request, timeout, ct), usingTimeProvider.Left
IAsyncEnumerable<T>views of the message streams:client.ReadEventsAsync<T>(ct)andbus.ReadCommandsAsync<T>(ct)(v7: read a bus's events and commands with await foreach (#87), and a guide to Obvs.Testing (#88) #148). Calling themEvents<T>andCommands<T>wasn't possible, because the bus's properties of those names hide extension methods.CancellationTokenon the bus's own async methods:PublishAsync,SendAsyncandReplyAsynctake an optional token (v7: CancellationToken on the bus's PublishAsync, SendAsync and ReplyAsync #138).IMessagePublisher<T>.PublishAsyncand the endpoint interfaces (v7: pass the cancellation token on to the transports (#87) #143).TestContext.Current.CancellationTokenin the tests that call those methods, which cleared every xUnit1051 warning (v7: pass the cancellation token on to the transports (#87) #143).Examples/to handlers andRequestAsync. They still use Rx subscriptions.MaxConcurrencyabove 1 now takes effect withMaxConcurrentCalls(Obvs.AzureSB: concurrent delivery, lock settings, Basic tier and credentials without Manage #132).Merge pull request #143 from christopherread/v7-transport-cancellation
Merge pull request #148 from christopherread/v7-async-streams
christopherread commentedon Oct 27, 2026
Done in #114 (handlers,
RequestAsyncandMaxConcurrency), #126 (settling messages after their handler), #138 and #143 (cancellation tokens through the bus and on to the transports), and #148 (ReadEventsAsyncandReadCommandsAsyncforawait foreach). Merged intov7.