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
+ }
+ }
+}