From 31448a9169595aa18042ae431e5c3d02b3618c4e Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 11:33:28 -0700 Subject: [PATCH] Handlers, RequestAsync and MaxConcurrency ObvsServiceBuilder gains HandleCommand, HandleEvent and HandleRequest, each taking a delegate or an ICommandHandler/IEventHandler/IRequestHandler class. Each message is handled in its own DI scope, with a MessageContext for the service name, the scope's services and publishing or sending further messages. Request handlers' responses are sent back as replies. Handlers start with the host. By default each handles one message at a time, in order; HandlerOptions.MaxConcurrency allows more. A handler that throws is logged with the service and message type, and later messages are still handled. Stopping the host waits for running and queued messages, and cancels running handlers if shutdown runs out of time. Handlers on the wrong side of a service (a command handler on a client, an event handler on a server) or for another service's messages fail at startup with the fix. IServiceBusClient.RequestAsync sends a request and returns the first response, or throws TimeoutException, measured on an optional TimeProvider so timeouts can be tested with virtual time. Part of #87. Co-Authored-By: Claude Opus 5.5 --- Directory.Packages.props | 2 + .../Hosting/Handlers/HandlerTestSupport.cs | 64 ++++ Obvs.Tests/Hosting/Handlers/TestHandlers.cs | 320 ++++++++++++++++++ .../Hosting/Handlers/TestRequestAsync.cs | 63 ++++ Obvs.Tests/Obvs.Tests.csproj | 2 + Obvs/Hosting/HandlerOptions.cs | 11 + Obvs/Hosting/HandlerRunner.cs | 118 +++++++ Obvs/Hosting/MessageContext.cs | 48 +++ Obvs/Hosting/MessageHandlers.cs | 35 ++ Obvs/Hosting/ObvsBuilder.cs | 2 + Obvs/Hosting/ObvsHostedService.cs | 30 +- Obvs/Hosting/ObvsServiceBuilder.cs | 130 ++++++- .../ObvsServiceCollectionExtensions.cs | 2 +- Obvs/Obvs.csproj | 5 + Obvs/ServiceBusClientRequestExtensions.cs | 59 ++++ 15 files changed, 877 insertions(+), 14 deletions(-) create mode 100644 Obvs.Tests/Hosting/Handlers/HandlerTestSupport.cs create mode 100644 Obvs.Tests/Hosting/Handlers/TestHandlers.cs create mode 100644 Obvs.Tests/Hosting/Handlers/TestRequestAsync.cs create mode 100644 Obvs/Hosting/HandlerOptions.cs create mode 100644 Obvs/Hosting/HandlerRunner.cs create mode 100644 Obvs/Hosting/MessageContext.cs create mode 100644 Obvs/Hosting/MessageHandlers.cs create mode 100644 Obvs/ServiceBusClientRequestExtensions.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index 298b830..76adacd 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -14,6 +14,7 @@ + @@ -25,6 +26,7 @@ + diff --git a/Obvs.Tests/Hosting/Handlers/HandlerTestSupport.cs b/Obvs.Tests/Hosting/Handlers/HandlerTestSupport.cs new file mode 100644 index 0000000..477a240 --- /dev/null +++ b/Obvs.Tests/Hosting/Handlers/HandlerTestSupport.cs @@ -0,0 +1,64 @@ +using System; +using System.Collections.Concurrent; +using Microsoft.Extensions.Logging; +using Obvs.Types; + +namespace Obvs.Tests.Hosting.Handlers +{ + public interface IOrderMessage : IMessage { } + + public interface IShippingMessage : IMessage { } + + public record PlaceOrder(string OrderId) : IOrderMessage, ICommand; + + public record OrderPlaced(string OrderId) : IOrderMessage, IEvent; + + public record GetQuote(string Sku) : IOrderMessage, IRequest + { + public string RequestId { get; set; } + public string RequesterId { get; set; } + } + + public record Quote(string Sku, decimal Price) : IOrderMessage, IResponse + { + public string RequestId { get; set; } + public string RequesterId { get; set; } + } + + public record ShipOrder(string OrderId) : IShippingMessage, ICommand; + + public sealed record LogEntry(LogLevel Level, string Category, string Message, Exception Exception); + + // Captures everything logged, so tests can check what handlers report + public sealed class ListLoggerProvider : ILoggerProvider + { + public ConcurrentQueue Entries { get; } = new ConcurrentQueue(); + + public ILogger CreateLogger(string categoryName) => new ListLogger(this, categoryName); + + public void Dispose() + { + } + + private sealed class ListLogger : ILogger + { + private readonly ListLoggerProvider _provider; + private readonly string _category; + + public ListLogger(ListLoggerProvider provider, string category) + { + _provider = provider; + _category = category; + } + + public IDisposable BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func formatter) + { + _provider.Entries.Enqueue(new LogEntry(logLevel, _category, formatter(state, exception), exception)); + } + } + } +} diff --git a/Obvs.Tests/Hosting/Handlers/TestHandlers.cs b/Obvs.Tests/Hosting/Handlers/TestHandlers.cs new file mode 100644 index 0000000..ad4d489 --- /dev/null +++ b/Obvs.Tests/Hosting/Handlers/TestHandlers.cs @@ -0,0 +1,320 @@ +using System; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Obvs.Hosting; +using Obvs.Testing; +using Xunit; + +namespace Obvs.Tests.Hosting.Handlers +{ + public class TestHandlers + { + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + [Fact] + public async Task ShouldHandleCommandsWithADelegate() + { + await using var harness = await StartOrdersAsync(service => service + .HandleCommand((command, context, ct) => context.PublishAsync(new OrderPlaced(command.OrderId), ct))); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + + await harness.Published(placed => placed.OrderId == "o-1"); + } + + [Fact] + public async Task ShouldReplyWithTheRequestHandlersResponse() + { + await using var harness = await StartOrdersAsync(service => service + .HandleRequest((request, context, ct) => Task.FromResult(new Quote(request.Sku, 9.99m)))); + + var quote = await harness.Client.RequestAsync(new GetQuote("sku-1"), Timeout, Ct); + + Assert.Equal(new Quote("sku-1", 9.99m), quote with { RequestId = null, RequesterId = null }); + } + + [Fact] + public async Task ShouldHandleEventsAsAClient() + { + var handled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var harness = await StartOrdersAsync(service => service + .HandleEvent((placed, context, ct) => + { + handled.TrySetResult(placed); + return Task.CompletedTask; + })); + + await harness.Bus.PublishAsync(new OrderPlaced("o-1")); + + Assert.Equal("o-1", (await handled.Task.WaitAsync(Timeout, Ct)).OrderId); + } + + [Fact] + public async Task ShouldResolveClassHandlersInANewScopePerMessage() + { + await using var harness = await ObvsTestHarness.StartAsync(services => services + .AddScoped() + .AddObvs(obvs => obvs.AddService("orders", service => service + .UseInMemory() + .HandleCommand())), + cancellationToken: Ct); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + await harness.Client.SendAsync(new PlaceOrder("o-2")); + var first = await harness.Published(placed => placed.OrderId.StartsWith("o-1:")); + var second = await harness.Published(placed => placed.OrderId.StartsWith("o-2:")); + + // Each message gets its own scope, and the handler's context exposes that same scope + var firstScope = first.OrderId.Split(':'); + var secondScope = second.OrderId.Split(':'); + Assert.NotEqual(firstScope[1], secondScope[1]); + Assert.Equal("same-scope", firstScope[2]); + Assert.Equal("same-scope", secondScope[2]); + } + + [Fact] + public async Task ShouldHandleOneMessageAtATimeInOrderByDefault() + { + var running = 0; + var maxRunning = 0; + await using var harness = await StartOrdersAsync(service => service + .HandleCommand(async (command, context, ct) => + { + RecordConcurrency(ref running, ref maxRunning); + await Task.Delay(20, ct); + Interlocked.Decrement(ref running); + await context.PublishAsync(new OrderPlaced(command.OrderId), ct); + })); + var ids = Enumerable.Range(1, 5).Select(i => $"o-{i}").ToArray(); + + foreach (var id in ids) + { + await harness.Client.SendAsync(new PlaceOrder(id)); + } + await harness.Published(placed => placed.OrderId == ids.Last()); + + Assert.Equal(1, maxRunning); + Assert.Equal(ids, harness.Messages.OfType().Select(placed => placed.OrderId)); + } + + [Fact] + public async Task ShouldHandleUpToMaxConcurrencyMessagesAtOnce() + { + // Each handler waits until three are running at once: with a limit below three they would all + // wait and time out; with no limit, all six would run together + var running = 0; + var maxRunning = 0; + var threeRunning = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var harness = await StartOrdersAsync(service => service + .HandleCommand(async (command, context, ct) => + { + if (RecordConcurrency(ref running, ref maxRunning) == 3) + { + threeRunning.TrySetResult(true); + } + await threeRunning.Task.WaitAsync(Timeout, ct); + Interlocked.Decrement(ref running); + await context.PublishAsync(new OrderPlaced(command.OrderId), ct); + }, options => options.MaxConcurrency = 3)); + + for (var i = 1; i <= 6; i++) + { + await harness.Client.SendAsync(new PlaceOrder($"o-{i}")); + } + for (var i = 1; i <= 6; i++) + { + await harness.Published(placed => placed.OrderId == $"o-{i}"); + } + + Assert.Equal(3, maxRunning); + } + + [Fact] + public async Task ShouldKeepHandlingAfterAHandlerThrows() + { + await using var harness = await StartOrdersAsync(service => service + .HandleCommand((command, context, ct) => command.OrderId == "o-1" + ? throw new InvalidOperationException("boom") + : context.PublishAsync(new OrderPlaced(command.OrderId), ct))); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + await harness.Client.SendAsync(new PlaceOrder("o-2")); + + await harness.Published(placed => placed.OrderId == "o-2"); + } + + [Fact] + public async Task ShouldLogHandlerErrorsWithTheServiceAndMessage() + { + var logs = new ListLoggerProvider(); + await using var harness = await ObvsTestHarness.StartAsync(services => services + .AddLogging(logging => logging.AddProvider(logs)) + .AddObvs(obvs => obvs.AddService("orders", service => service + .UseInMemory() + .HandleCommand((command, context, ct) => throw new InvalidOperationException("boom")))), + cancellationToken: Ct); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + + var error = await WaitForAsync(() => logs.Entries.FirstOrDefault(entry => entry.Level == LogLevel.Error)); + Assert.IsType(error.Exception); + Assert.Contains("orders", error.Message); + Assert.Contains("PlaceOrder", error.Message); + } + + [Fact] + public async Task ShouldFinishRunningHandlersWhenTheHostStops() + { + var started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var finished = false; + using var host = BuildOrdersHost(service => service + .HandleCommand(async (command, context, ct) => + { + started.TrySetResult(true); + await release.Task; + finished = true; + })); + await host.StartAsync(Ct); + await host.Services.GetRequiredService().SendAsync(new PlaceOrder("o-1")); + await started.Task.WaitAsync(Timeout, Ct); + + var stopping = host.StopAsync(Ct); + await Task.Delay(100, Ct); + Assert.False(stopping.IsCompleted, "The host stopped while a handler was still running"); + + release.SetResult(true); + await stopping.WaitAsync(Timeout, Ct); + Assert.True(finished); + } + + [Fact] + public async Task ShouldCancelRunningHandlersWhenStoppingTimesOut() + { + var started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var host = BuildOrdersHost(service => service + .HandleCommand(async (command, context, ct) => + { + started.TrySetResult(true); + try + { + await Task.Delay(System.Threading.Timeout.Infinite, ct); + } + catch (OperationCanceledException) + { + cancelled.TrySetResult(true); + throw; + } + })); + await host.StartAsync(Ct); + await host.Services.GetRequiredService().SendAsync(new PlaceOrder("o-1")); + await started.Task.WaitAsync(Timeout, Ct); + + using var shutdownTimeout = new CancellationTokenSource(TimeSpan.FromMilliseconds(200)); + await host.StopAsync(shutdownTimeout.Token).WaitAsync(Timeout, Ct); + + Assert.True(await cancelled.Task.WaitAsync(Timeout, Ct)); + } + + [Fact] + public async Task ShouldFailToStartWhenACommandHandlerIsOnAClientOnlyService() + { + var exception = await Assert.ThrowsAsync(() => StartOrdersAsync(service => service + .AsClient() + .HandleCommand((command, context, ct) => Task.CompletedTask))); + + var error = Assert.Single(exception.Errors); + Assert.Contains("'orders'", error); + Assert.Contains("PlaceOrder", error); + Assert.Contains("AsServer()", error); + } + + [Fact] + public async Task ShouldFailToStartWhenAnEventHandlerIsOnAServerOnlyService() + { + var exception = await Assert.ThrowsAsync(() => StartOrdersAsync(service => service + .AsServer() + .HandleEvent((placed, context, ct) => Task.CompletedTask))); + + var error = Assert.Single(exception.Errors); + Assert.Contains("'orders'", error); + Assert.Contains("OrderPlaced", error); + Assert.Contains("AsClient()", error); + } + + [Fact] + public async Task ShouldFailToStartWhenAHandlerIsForAnotherServicesMessage() + { + var exception = await Assert.ThrowsAsync(() => StartOrdersAsync(service => service + .HandleCommand((command, context, ct) => Task.CompletedTask))); + + var error = Assert.Single(exception.Errors); + Assert.Contains("'orders'", error); + Assert.Contains("ShipOrder", error); + Assert.Contains("IOrderMessage", error); + } + + private static Task StartOrdersAsync(Action configure) + { + return ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs + .AddService("orders", service => configure(service.UseInMemory()))), + cancellationToken: Ct); + } + + private static IHost BuildOrdersHost(Action configure) + { + var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true }); + builder.Services.AddObvs(obvs => obvs.AddService("orders", service => configure(service.UseInMemory()))); + return builder.Build(); + } + + private static int RecordConcurrency(ref int running, ref int maxRunning) + { + var now = Interlocked.Increment(ref running); + int max; + while (now > (max = Volatile.Read(ref maxRunning)) && Interlocked.CompareExchange(ref maxRunning, now, max) != max) + { + } + return now; + } + + private static async Task WaitForAsync(Func find) where T : class + { + var deadline = DateTime.UtcNow + Timeout; + T found; + while ((found = find()) == null && DateTime.UtcNow < deadline) + { + await Task.Delay(20, Ct); + } + return found ?? throw new TimeoutException("The expected log entry wasn't written in time."); + } + + public sealed class ScopeMarker + { + public string Id { get; } = Guid.NewGuid().ToString("N"); + } + + public sealed class PlaceOrderHandler : ICommandHandler + { + private readonly ScopeMarker _scope; + + public PlaceOrderHandler(ScopeMarker scope) + { + _scope = scope; + } + + public Task HandleAsync(PlaceOrder command, MessageContext context, CancellationToken cancellationToken) + { + var sameScope = ReferenceEquals(_scope, context.Services.GetRequiredService()) ? "same-scope" : "other-scope"; + return context.PublishAsync(new OrderPlaced($"{command.OrderId}:{_scope.Id}:{sameScope}"), cancellationToken); + } + } + } +} diff --git a/Obvs.Tests/Hosting/Handlers/TestRequestAsync.cs b/Obvs.Tests/Hosting/Handlers/TestRequestAsync.cs new file mode 100644 index 0000000..1ef2238 --- /dev/null +++ b/Obvs.Tests/Hosting/Handlers/TestRequestAsync.cs @@ -0,0 +1,63 @@ +using System; +using System.Reactive.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Time.Testing; +using Obvs.Hosting; +using Obvs.Testing; +using Xunit; + +namespace Obvs.Tests.Hosting.Handlers +{ + // RequestAsync works on any bus; these use a plain Rx responder rather than a request handler + public class TestRequestAsync + { + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + [Fact] + public async Task ShouldReturnTheResponse() + { + await using var harness = await StartOrdersAsync(); + using var server = harness.Bus.Requests.OfType() + .Subscribe(request => _ = harness.Bus.ReplyAsync(request, new Quote(request.Sku, 9.99m))); + + var quote = await harness.Client.RequestAsync(new GetQuote("sku-1"), TimeSpan.FromSeconds(5), Ct); + + Assert.Equal(9.99m, quote.Price); + } + + [Fact] + public async Task ShouldTimeOutOnTheTimeProviderWhenNoResponseArrives() + { + var time = new FakeTimeProvider(); + await using var harness = await StartOrdersAsync(); + + var pending = harness.Client.RequestAsync(new GetQuote("sku-1"), TimeSpan.FromMinutes(5), Ct, time); + time.Advance(TimeSpan.FromMinutes(6)); + + // Completes because virtual time passed, long before five real minutes + var exception = await Assert.ThrowsAsync(() => pending).WaitAsync(TimeSpan.FromSeconds(2), Ct); + Assert.Contains("No Quote response to GetQuote", exception.Message); + } + + [Fact] + public async Task ShouldStopWaitingWhenCancelled() + { + await using var harness = await StartOrdersAsync(); + using var cancellation = new CancellationTokenSource(); + + var pending = harness.Client.RequestAsync(new GetQuote("sku-1"), TimeSpan.FromMinutes(5), cancellation.Token); + cancellation.Cancel(); + + await Assert.ThrowsAnyAsync(() => pending).WaitAsync(TimeSpan.FromSeconds(2), Ct); + } + + private static Task StartOrdersAsync() + { + return ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseInMemory())), + cancellationToken: Ct); + } + } +} diff --git a/Obvs.Tests/Obvs.Tests.csproj b/Obvs.Tests/Obvs.Tests.csproj index 019581f..1935ae7 100644 --- a/Obvs.Tests/Obvs.Tests.csproj +++ b/Obvs.Tests/Obvs.Tests.csproj @@ -10,6 +10,7 @@ + @@ -20,5 +21,6 @@ + \ No newline at end of file diff --git a/Obvs/Hosting/HandlerOptions.cs b/Obvs/Hosting/HandlerOptions.cs new file mode 100644 index 0000000..7640964 --- /dev/null +++ b/Obvs/Hosting/HandlerOptions.cs @@ -0,0 +1,11 @@ +namespace Obvs.Hosting +{ + public sealed class HandlerOptions + { + /// + /// How many messages the handler may process at once. One by default, so messages are handled + /// one at a time, in the order they arrive. + /// + public int MaxConcurrency { get; set; } = 1; + } +} diff --git a/Obvs/Hosting/HandlerRunner.cs b/Obvs/Hosting/HandlerRunner.cs new file mode 100644 index 0000000..07b74c2 --- /dev/null +++ b/Obvs/Hosting/HandlerRunner.cs @@ -0,0 +1,118 @@ +using System; +using System.Linq; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Obvs.Types; + +namespace Obvs.Hosting +{ + internal enum HandlerKind + { + Command, + Event, + Request, + } + + // What a HandleCommand/HandleEvent/HandleRequest call registered, until the host starts it + internal sealed class HandlerRegistration + { + public HandlerRegistration(HandlerKind kind, Type messageType, HandlerOptions options, Func start) + { + Kind = kind; + MessageType = messageType; + Options = options; + Start = start; + } + + public HandlerKind Kind { get; } + public Type MessageType { get; } + public HandlerOptions Options { get; } + public Func Start { get; } + } + + // Queues messages from the bus and runs the handler on them, MaxConcurrency at a time. With one worker, + // messages are handled one at a time in arrival order, as an Rx subscriber would see them. + internal sealed class HandlerRunner + { + private readonly IDisposable _subscription; + private readonly Channel _queue; + private readonly Task[] _workers; + private readonly CancellationTokenSource _cancelHandlers = new CancellationTokenSource(); + + private HandlerRunner(IDisposable subscription, Channel queue, Func handle, int maxConcurrency) + { + _subscription = subscription; + _queue = queue; + _workers = Enumerable.Range(0, maxConcurrency) + .Select(_ => Task.Run(() => WorkAsync(handle))) + .ToArray(); + } + + public static HandlerRunner Start(IObservable messages, Func handler, + HandlerOptions options, IServiceBus bus, string serviceName, IServiceProvider serviceProvider, ILogger logger) + where TMessage : class, IMessage + { + var queue = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = options.MaxConcurrency == 1 }); + var subscription = messages.Subscribe( + message => queue.Writer.TryWrite(message), + error => logger.LogError(error, "The {MessageType} stream on service '{ServiceName}' failed, so its handler stopped receiving.", typeof(TMessage).Name, serviceName)); + var scopes = serviceProvider.GetRequiredService(); + + async Task Handle(IMessage message, CancellationToken cancellationToken) + { + var scope = scopes.CreateAsyncScope(); + try + { + await handler((TMessage)message, new MessageContext(serviceName, scope.ServiceProvider, bus), cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Shutdown ran out of time and cancelled the handler + } + catch (Exception exception) + { + logger.LogError(exception, "Handling {MessageType} on service '{ServiceName}' failed.", message.GetType().Name, serviceName); + } + finally + { + await scope.DisposeAsync().ConfigureAwait(false); + } + } + + return new HandlerRunner(subscription, queue, Handle, options.MaxConcurrency); + } + + // Stops receiving, then lets queued and running messages finish. If stopping runs out of time, + // running handlers are cancelled and anything still queued is dropped. + public async Task StopAsync(CancellationToken cancellationToken) + { + _subscription.Dispose(); + _queue.Writer.TryComplete(); + + var finished = Task.WhenAll(_workers); + var timedOut = Task.Delay(Timeout.Infinite, cancellationToken); + if (await Task.WhenAny(finished, timedOut).ConfigureAwait(false) == timedOut) + { + _cancelHandlers.Cancel(); + } + } + + private async Task WorkAsync(Func handle) + { + while (await _queue.Reader.WaitToReadAsync().ConfigureAwait(false)) + { + while (!_cancelHandlers.IsCancellationRequested && _queue.Reader.TryRead(out var message)) + { + await handle(message, _cancelHandlers.Token).ConfigureAwait(false); + } + if (_cancelHandlers.IsCancellationRequested) + { + return; + } + } + } + } +} diff --git a/Obvs/Hosting/MessageContext.cs b/Obvs/Hosting/MessageContext.cs new file mode 100644 index 0000000..2d2ef29 --- /dev/null +++ b/Obvs/Hosting/MessageContext.cs @@ -0,0 +1,48 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Obvs.Types; + +namespace Obvs.Hosting +{ + /// + /// What a handler needs besides the message: the service it arrived on, the services of its DI scope, + /// and the bus, to publish or send further messages. + /// + public sealed class MessageContext + { + private readonly IServiceBus _bus; + + internal MessageContext(string serviceName, IServiceProvider services, IServiceBus bus) + { + ServiceName = serviceName; + Services = services; + _bus = bus; + } + + /// The name of the service the message arrived on. + public string ServiceName { get; } + + /// The services of this message's DI scope, disposed when the handler finishes. + public IServiceProvider Services { get; } + + /// Publishes an event. + public Task PublishAsync(IEvent ev, CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + return _bus.PublishAsync(ev); + } + + internal Task ReplyAsync(IRequest request, IResponse response) + { + return _bus.ReplyAsync(request, response); + } + + /// Sends a command. + public Task SendAsync(ICommand command, CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + return _bus.SendAsync(command); + } + } +} diff --git a/Obvs/Hosting/MessageHandlers.cs b/Obvs/Hosting/MessageHandlers.cs new file mode 100644 index 0000000..1701067 --- /dev/null +++ b/Obvs/Hosting/MessageHandlers.cs @@ -0,0 +1,35 @@ +using System.Threading; +using System.Threading.Tasks; +using Obvs.Types; + +namespace Obvs.Hosting +{ + /// + /// Handles a command. Register with HandleCommand<TCommand, THandler>(); a new instance is + /// resolved from its own DI scope for each command, so it can depend on scoped services. + /// + public interface ICommandHandler where TCommand : class, ICommand + { + Task HandleAsync(TCommand command, MessageContext context, CancellationToken cancellationToken); + } + + /// + /// Handles an event. Register with HandleEvent<TEvent, THandler>(); a new instance is + /// resolved from its own DI scope for each event. + /// + public interface IEventHandler where TEvent : class, IEvent + { + Task HandleAsync(TEvent ev, MessageContext context, CancellationToken cancellationToken); + } + + /// + /// Answers a request; the returned response is sent back to the requester. Register with + /// HandleRequest<TRequest, TResponse, THandler>(); a new instance is resolved from its own DI scope for each request. + /// + public interface IRequestHandler + where TRequest : class, IRequest + where TResponse : class, IResponse + { + Task HandleAsync(TRequest request, MessageContext context, CancellationToken cancellationToken); + } +} diff --git a/Obvs/Hosting/ObvsBuilder.cs b/Obvs/Hosting/ObvsBuilder.cs index 6ea05ae..5883239 100644 --- a/Obvs/Hosting/ObvsBuilder.cs +++ b/Obvs/Hosting/ObvsBuilder.cs @@ -62,6 +62,8 @@ public ObvsBuilder AddService(string name, Action ServiceBuilders => _services; + internal ServiceBus Build(IServiceProvider serviceProvider) { Validate(serviceProvider); diff --git a/Obvs/Hosting/ObvsHostedService.cs b/Obvs/Hosting/ObvsHostedService.cs index 32a3f79..77c257f 100644 --- a/Obvs/Hosting/ObvsHostedService.cs +++ b/Obvs/Hosting/ObvsHostedService.cs @@ -1,43 +1,49 @@ using System; +using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; namespace Obvs.Hosting { /// /// Creates the bus when the host starts, so it connects and subscribes before the app takes traffic, - /// and configuration errors stop startup instead of surfacing on first use. - /// The container, not this service, disposes the bus. + /// and configuration errors stop startup instead of surfacing on first use. Starts the services' + /// handlers, and on stop waits for running ones. The container, not this service, disposes the bus. /// internal sealed class ObvsHostedService : IHostedService { private readonly IServiceProvider _serviceProvider; private readonly string _key; + private readonly ObvsBuilder _builder; + private readonly List _handlers = new List(); - public ObvsHostedService(IServiceProvider serviceProvider, string key) + public ObvsHostedService(IServiceProvider serviceProvider, string key, ObvsBuilder builder) { _serviceProvider = serviceProvider; _key = key; + _builder = builder; } public Task StartAsync(CancellationToken cancellationToken) { - if (_key == null) - { - _serviceProvider.GetRequiredService(); - } - else - { - _serviceProvider.GetRequiredKeyedService(_key); - } + var bus = _key == null + ? _serviceProvider.GetRequiredService() + : _serviceProvider.GetRequiredKeyedService(_key); + + var logger = _serviceProvider.GetService()?.CreateLogger("Obvs.Hosting.Handlers") ?? NullLogger.Instance; + _handlers.AddRange(_builder.ServiceBuilders.SelectMany(service => service.Handlers + .Select(handler => handler.Start(bus, service.Name, _serviceProvider, logger)))); return Task.CompletedTask; } public Task StopAsync(CancellationToken cancellationToken) { - return Task.CompletedTask; + return Task.WhenAll(_handlers.Select(handler => handler.StopAsync(cancellationToken))); } } } diff --git a/Obvs/Hosting/ObvsServiceBuilder.cs b/Obvs/Hosting/ObvsServiceBuilder.cs index 2c8bfc9..814b238 100644 --- a/Obvs/Hosting/ObvsServiceBuilder.cs +++ b/Obvs/Hosting/ObvsServiceBuilder.cs @@ -1,6 +1,9 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Linq; +using System.Threading; +using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using Obvs.Configuration; using Obvs.Serialization; @@ -15,6 +18,7 @@ public sealed class ObvsServiceBuilder { private readonly ObvsBuilder _bus; private readonly List>> _validators = new List>>(); + private readonly List _handlers = new List(); internal ObvsServiceBuilder(string name, Type serviceMessageType, IServiceCollection services, ObvsBuilder bus) { @@ -65,7 +69,7 @@ public ObvsServiceBuilder ValidateWith(Func Validate(IServiceProvider serviceProvider) { - return _validators.SelectMany(validate => validate(serviceProvider) ?? Enumerable.Empty()); + return ValidateHandlers().Concat(_validators.SelectMany(validate => validate(serviceProvider) ?? Enumerable.Empty())); } internal Func> EndpointProviderFactory { get; private set; } @@ -84,6 +88,130 @@ public ObvsServiceBuilder UseEndpointProvider(Func + /// Handles the service's commands with , which runs + /// in its own DI scope per command. Handlers start with the host, and the host waits for running ones on stop. + /// Exceptions are logged and don't stop later commands being handled. Requires this app to be the service's server. + /// + public ObvsServiceBuilder HandleCommand(Func handler, Action configure = null) + where TCommand : class, ICommand + { + return AddHandler(HandlerKind.Command, handler, configure, bus => bus.Commands.OfType()); + } + + /// Handles the service's commands with a resolved per command. + public ObvsServiceBuilder HandleCommand(Action configure = null) + where TCommand : class, ICommand + where THandler : class, ICommandHandler + { + return HandleCommand((command, context, cancellationToken) => + ResolveHandler(context).HandleAsync(command, context, cancellationToken), configure); + } + + /// + /// Handles the service's events with , in its own DI scope per event. + /// Requires this app to be the service's client. + /// + public ObvsServiceBuilder HandleEvent(Func handler, Action configure = null) + where TEvent : class, IEvent + { + return AddHandler(HandlerKind.Event, handler, configure, bus => bus.Events.OfType()); + } + + /// Handles the service's events with a resolved per event. + public ObvsServiceBuilder HandleEvent(Action configure = null) + where TEvent : class, IEvent + where THandler : class, IEventHandler + { + return HandleEvent((ev, context, cancellationToken) => + ResolveHandler(context).HandleAsync(ev, context, cancellationToken), configure); + } + + /// + /// Answers the service's requests with the response returns, + /// in its own DI scope per request. Requires this app to be the service's server. + /// + public ObvsServiceBuilder HandleRequest(Func> handler, Action configure = null) + where TRequest : class, IRequest + where TResponse : class, IResponse + { + if (handler == null) + { + throw new ArgumentNullException(nameof(handler)); + } + + return AddHandler(HandlerKind.Request, async (request, context, cancellationToken) => + { + var response = await handler(request, context, cancellationToken).ConfigureAwait(false); + if (response != null) + { + await context.ReplyAsync(request, response).ConfigureAwait(false); + } + }, configure, bus => bus.Requests.OfType()); + } + + /// Answers the service's requests with a resolved per request. + public ObvsServiceBuilder HandleRequest(Action configure = null) + where TRequest : class, IRequest + where TResponse : class, IResponse + where THandler : class, IRequestHandler + { + return HandleRequest((request, context, cancellationToken) => + ResolveHandler(context).HandleAsync(request, context, cancellationToken), configure); + } + + internal IReadOnlyList Handlers => _handlers; + + private ObvsServiceBuilder AddHandler(HandlerKind kind, Func handler, + Action configure, Func> messages) + where TMessage : class, IMessage + { + if (handler == null) + { + throw new ArgumentNullException(nameof(handler)); + } + + var options = new HandlerOptions(); + configure?.Invoke(options); + _handlers.Add(new HandlerRegistration(kind, typeof(TMessage), options, (bus, serviceName, serviceProvider, logger) => + HandlerRunner.Start(messages(bus), handler, options, bus, serviceName, serviceProvider, logger))); + return this; + } + + // A registered handler is used as is; otherwise one is created from the message's scope + private static THandler ResolveHandler(MessageContext context) where THandler : class + { + return ActivatorUtilities.GetServiceOrCreateInstance(context.Services); + } + + private IEnumerable ValidateHandlers() + { + foreach (var handler in _handlers) + { + var messageName = handler.MessageType.Name; + if (!ServiceMessageType.IsAssignableFrom(handler.MessageType)) + { + yield return $"Service '{Name}' has a handler for {messageName}, which doesn't implement {ServiceMessageType.Name}. " + + $"Register the handler on the service that {messageName} belongs to."; + } + if (handler.Kind != HandlerKind.Event && !IsServer) + { + var kind = handler.Kind == HandlerKind.Command ? "commands" : "requests"; + yield return $"Service '{Name}' handles {messageName} {kind} but is configured AsClient(), and only servers receive {kind}. " + + "Call AsServer() or AsClientAndServer() instead."; + } + if (handler.Kind == HandlerKind.Event && !IsClient) + { + yield return $"Service '{Name}' handles {messageName} events but is configured AsServer(), and only clients receive events. " + + "Call AsClient() or AsClientAndServer() instead."; + } + if (handler.Options.MaxConcurrency < 1) + { + yield return $"Service '{Name}' handles {messageName} with MaxConcurrency {handler.Options.MaxConcurrency}. It must be at least 1."; + } + } + } + /// This app sends the service's commands and requests, and receives its events and responses. public ObvsServiceBuilder AsClient() { diff --git a/Obvs/Hosting/ObvsServiceCollectionExtensions.cs b/Obvs/Hosting/ObvsServiceCollectionExtensions.cs index 4cca78e..4754105 100644 --- a/Obvs/Hosting/ObvsServiceCollectionExtensions.cs +++ b/Obvs/Hosting/ObvsServiceCollectionExtensions.cs @@ -59,7 +59,7 @@ private static void AddBus(IServiceCollection services, string key, ObvsBuilder } // Not AddHostedService, which would de-duplicate the hosted services of keyed buses - services.AddSingleton(serviceProvider => new ObvsHostedService(serviceProvider, key)); + services.AddSingleton(serviceProvider => new ObvsHostedService(serviceProvider, key, builder)); } private static readonly object DefaultBusKey = new object(); diff --git a/Obvs/Obvs.csproj b/Obvs/Obvs.csproj index 72ed7dd..6f5c8e6 100644 --- a/Obvs/Obvs.csproj +++ b/Obvs/Obvs.csproj @@ -13,9 +13,14 @@ + + + + + diff --git a/Obvs/ServiceBusClientRequestExtensions.cs b/Obvs/ServiceBusClientRequestExtensions.cs new file mode 100644 index 0000000..7e1e8fc --- /dev/null +++ b/Obvs/ServiceBusClientRequestExtensions.cs @@ -0,0 +1,59 @@ +using System; +using System.Reactive.Linq; +using System.Reactive.Threading.Tasks; +using System.Threading; +using System.Threading.Tasks; +using Obvs.Types; + +namespace Obvs +{ + public static class ServiceBusClientRequestExtensions + { + /// + /// Sends a request and returns the first response of type . + /// Throws if none arrives within , measured on + /// (the system clock by default; pass a fake one to test timeouts). + /// + public static async Task RequestAsync(this IServiceBusClient client, TRequest request, TimeSpan timeout, + CancellationToken cancellationToken = default, TimeProvider timeProvider = null) + where TRequest : class, IRequest + where TResponse : class, IResponse + { + if (client == null) + { + throw new ArgumentNullException(nameof(client)); + } + if (request == null) + { + throw new ArgumentNullException(nameof(request)); + } + + using (var waiting = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken)) + { + // Both started before the first await, so the timeout runs from when the request was sent + var response = client.GetResponses(request).OfType().FirstAsync().ToTask(waiting.Token); + var expiry = Delay(timeout, timeProvider ?? TimeProvider.System, waiting.Token); + + var first = await Task.WhenAny(response, expiry).ConfigureAwait(false); + // Stops listening for responses, or stops the timer, whichever is still running + waiting.Cancel(); + cancellationToken.ThrowIfCancellationRequested(); + if (first == response) + { + return await response.ConfigureAwait(false); + } + } + + throw new TimeoutException($"No {typeof(TResponse).Name} response to {typeof(TRequest).Name} within {timeout}."); + } + + private static Task Delay(TimeSpan delay, TimeProvider timeProvider, CancellationToken cancellationToken) + { +#if NET8_0_OR_GREATER + return Task.Delay(delay, timeProvider, cancellationToken); +#else + return timeProvider.Delay(delay, cancellationToken); +#endif + } + } +}