From 5ab18d6e0f8add4c93c7128158b1b3e8015117b9 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 09:11:53 -0700 Subject: [PATCH] Host the bus through dependency injection (AddObvs) services.AddObvs(obvs => obvs.AddService(name, service => service.UseEndpointProvider(...))) registers one singleton bus, resolvable as both IServiceBus and IServiceBusClient. A hosted service creates it when the host starts, so configuration errors stop startup; the container disposes it. Repeated AddObvs calls add to the same bus, AsClient/AsServer choose which endpoints are created, AddObvs(key, ...) registers a separate keyed bus, and an invalid configuration fails with one ObvsConfigurationException listing every problem and its fix. UseEndpointProvider bridges every existing transport until each gets its own Use() method. ServiceBus.Dispose is now safe to call twice: the container tracks the bus behind both of its registrations. Part of #84. Co-Authored-By: Claude Opus 5.5 --- Directory.Packages.props | 3 + Obvs.Tests/Hosting/TestAddObvs.cs | 276 ++++++++++++++++++ Obvs.Tests/Obvs.Tests.csproj | 1 + Obvs.Tests/TestServiceBus.cs | 14 + Obvs/Hosting/ObvsBuilder.cs | 79 +++++ Obvs/Hosting/ObvsConfigurationException.cs | 21 ++ Obvs/Hosting/ObvsHostedService.cs | 43 +++ Obvs/Hosting/ObvsServiceBuilder.cs | 64 ++++ .../ObvsServiceCollectionExtensions.cs | 88 ++++++ Obvs/IServiceBus.cs | 7 + Obvs/Obvs.csproj | 2 + 11 files changed, 598 insertions(+) create mode 100644 Obvs.Tests/Hosting/TestAddObvs.cs create mode 100644 Obvs/Hosting/ObvsBuilder.cs create mode 100644 Obvs/Hosting/ObvsConfigurationException.cs create mode 100644 Obvs/Hosting/ObvsHostedService.cs create mode 100644 Obvs/Hosting/ObvsServiceBuilder.cs create mode 100644 Obvs/Hosting/ObvsServiceCollectionExtensions.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index f75f379e..f9b32688 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -9,7 +9,9 @@ + + @@ -25,6 +27,7 @@ + diff --git a/Obvs.Tests/Hosting/TestAddObvs.cs b/Obvs.Tests/Hosting/TestAddObvs.cs new file mode 100644 index 00000000..d5d74c16 --- /dev/null +++ b/Obvs.Tests/Hosting/TestAddObvs.cs @@ -0,0 +1,276 @@ +using System; +using System.Reactive.Linq; +using System.Reactive.Threading.Tasks; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Obvs.Configuration; +using Obvs.Hosting; +using Obvs.Types; +using Xunit; + +namespace Obvs.Tests.Hosting +{ + public class TestAddObvs + { + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); + + [Fact] + public void ShouldResolveOneBusAsBothServiceBusAndClient() + { + var provider = new FakeEndpointProvider(typeof(IOrderMessage)); + using var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseEndpointProvider(_ => provider)))); + + var bus = host.Services.GetRequiredService(); + + Assert.Same(bus, host.Services.GetRequiredService()); + Assert.Same(bus, host.Services.GetRequiredService()); + } + + [Fact] + public async Task ShouldCreateTheBusWhenTheHostStarts() + { + var provider = new FakeEndpointProvider(typeof(IOrderMessage)); + using var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseEndpointProvider(_ => provider)))); + + Assert.Equal(0, provider.EndpointsCreated + provider.ClientsCreated); + + await host.StartAsync(TestContext.Current.CancellationToken); + + Assert.Equal(1, provider.EndpointsCreated); + Assert.Equal(1, provider.ClientsCreated); + } + + [Fact] + public async Task ShouldDeliverCommandsFromClientToServer() + { + var provider = new FakeEndpointProvider(typeof(IOrderMessage)); + using var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseEndpointProvider(_ => provider)))); + await host.StartAsync(TestContext.Current.CancellationToken); + + var received = host.Services.GetRequiredService() + .Commands.OfType().FirstAsync().ToTask(); + await host.Services.GetRequiredService().SendAsync(new PlaceOrder { OrderId = "o-1" }); + + var command = await received.WaitAsync(Timeout, TestContext.Current.CancellationToken); + Assert.Equal("o-1", command.OrderId); + } + + [Fact] + public async Task ShouldAddServicesFromRepeatedCallsToOneBus() + { + var orders = new FakeEndpointProvider(typeof(IOrderMessage)); + var shipping = new FakeEndpointProvider(typeof(IShippingMessage)); + using var host = BuildHost(services => services + .AddObvs(obvs => obvs.AddService("orders", service => service.UseEndpointProvider(_ => orders))) + .AddObvs(obvs => obvs.AddService("shipping", service => service.UseEndpointProvider(_ => shipping)))); + await host.StartAsync(TestContext.Current.CancellationToken); + + Assert.Same(host.Services.GetRequiredService(), host.Services.GetRequiredService()); + Assert.Equal(1, orders.EndpointsCreated); + Assert.Equal(1, shipping.EndpointsCreated); + } + + [Fact] + public async Task ShouldOnlyCreateClientEndpointsForAClientService() + { + var provider = new FakeEndpointProvider(typeof(IOrderMessage)); + using var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseEndpointProvider(_ => provider).AsClient()))); + + await host.StartAsync(TestContext.Current.CancellationToken); + + Assert.Equal(0, provider.EndpointsCreated); + Assert.Equal(1, provider.ClientsCreated); + } + + [Fact] + public async Task ShouldOnlyCreateServerEndpointsForAServerService() + { + var provider = new FakeEndpointProvider(typeof(IOrderMessage)); + using var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseEndpointProvider(_ => provider).AsServer()))); + + await host.StartAsync(TestContext.Current.CancellationToken); + + Assert.Equal(1, provider.EndpointsCreated); + Assert.Equal(0, provider.ClientsCreated); + } + + [Fact] + public async Task ShouldDisposeTheBusWhenTheContainerIsDisposed() + { + var provider = new FakeEndpointProvider(typeof(IOrderMessage)); + var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseEndpointProvider(_ => provider)))); + await host.StartAsync(TestContext.Current.CancellationToken); + await host.StopAsync(TestContext.Current.CancellationToken); + + Assert.False(provider.Endpoint.Messages.IsDisposed); + + host.Dispose(); + + Assert.True(provider.Endpoint.Messages.IsDisposed); + } + + [Fact] + public async Task ShouldDisposeTheBusOnlyOnce() + { + // Server only, so the shared fake endpoint is owned once by the bus + var provider = new FakeEndpointProvider(typeof(IOrderMessage)); + var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseEndpointProvider(_ => provider).AsServer()))); + await host.StartAsync(TestContext.Current.CancellationToken); + host.Services.GetRequiredService(); + await host.StopAsync(TestContext.Current.CancellationToken); + + host.Dispose(); + + Assert.Equal(1, provider.Endpoint.DisposeCount); + } + + [Fact] + public async Task ShouldGiveEachHostItsOwnBus() + { + // One provider per container, as a real transport's connection would be + Action configure = services => services.AddObvs(obvs => obvs + .AddService("orders", service => service.UseEndpointProvider(_ => new FakeEndpointProvider(typeof(IOrderMessage))))); + using var hostA = BuildHost(configure); + using var hostB = BuildHost(configure); + await hostA.StartAsync(TestContext.Current.CancellationToken); + await hostB.StartAsync(TestContext.Current.CancellationToken); + + var busA = hostA.Services.GetRequiredService(); + var busB = hostB.Services.GetRequiredService(); + Assert.NotSame(busA, busB); + + var receivedByB = busB.Commands.OfType().FirstAsync().ToTask(); + var receivedByA = busA.Commands.OfType().FirstAsync().ToTask(); + await hostA.Services.GetRequiredService().SendAsync(new PlaceOrder { OrderId = "o-1" }); + + await receivedByA.WaitAsync(Timeout, TestContext.Current.CancellationToken); + Assert.False(receivedByB.IsCompleted); + } + + [Fact] + public void ShouldResolveAKeyedBusSeparatelyFromTheDefaultBus() + { + using var host = BuildHost(services => services + .AddObvs(obvs => obvs.AddService("orders", service => service + .UseEndpointProvider(_ => new FakeEndpointProvider(typeof(IOrderMessage))))) + .AddObvs("analytics", obvs => obvs.AddService("shipping", service => service + .UseEndpointProvider(_ => new FakeEndpointProvider(typeof(IShippingMessage)))))); + + var defaultBus = host.Services.GetRequiredService(); + var analyticsBus = host.Services.GetRequiredKeyedService("analytics"); + + Assert.NotSame(defaultBus, analyticsBus); + Assert.Same(analyticsBus, host.Services.GetRequiredKeyedService("analytics")); + } + + [Fact] + public async Task ShouldFailToStartWhenAServiceHasNoTransport() + { + using var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => { }))); + + var exception = await Assert.ThrowsAsync( + () => host.StartAsync(TestContext.Current.CancellationToken)); + + var error = Assert.Single(exception.Errors); + Assert.Contains("'orders'", error); + Assert.Contains("UseEndpointProvider", error); + } + + [Fact] + public async Task ShouldFailToStartWhenTwoServicesShareAName() + { + using var host = BuildHost(services => services + .AddObvs(obvs => obvs.AddService("orders", service => service + .UseEndpointProvider(_ => new FakeEndpointProvider(typeof(IOrderMessage))))) + .AddObvs(obvs => obvs.AddService("orders", service => service + .UseEndpointProvider(_ => new FakeEndpointProvider(typeof(IShippingMessage)))))); + + var exception = await Assert.ThrowsAsync( + () => host.StartAsync(TestContext.Current.CancellationToken)); + + var error = Assert.Single(exception.Errors); + Assert.Contains("'orders'", error); + Assert.Contains("more than once", error); + } + + [Fact] + public async Task ShouldReportEveryConfigurationProblemAtOnce() + { + using var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => { }) + .AddService("shipping", service => { }))); + + var exception = await Assert.ThrowsAsync( + () => host.StartAsync(TestContext.Current.CancellationToken)); + + Assert.Equal(2, exception.Errors.Count); + Assert.Contains(exception.Errors, error => error.Contains("'orders'")); + Assert.Contains(exception.Errors, error => error.Contains("'shipping'")); + } + + private static IHost BuildHost(Action configure) + { + var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true }); + configure(builder.Services); + return builder.Build(); + } + + public interface IOrderMessage : IMessage { } + + public interface IShippingMessage : IMessage { } + + public class PlaceOrder : IOrderMessage, ICommand + { + public string OrderId { get; set; } + } + + // Hands out one shared fake endpoint as both server and client, so messages loop back in-process + private sealed class FakeEndpointProvider : IServiceEndpointProvider + { + public FakeEndpointProvider(Type serviceMessageType) + { + Endpoint = new CountingServiceEndpoint(serviceMessageType); + } + + public CountingServiceEndpoint Endpoint { get; } + public int EndpointsCreated { get; private set; } + public int ClientsCreated { get; private set; } + + public IServiceEndpoint CreateEndpoint() + { + EndpointsCreated++; + return Endpoint; + } + + public IServiceEndpointClient CreateEndpointClient() + { + ClientsCreated++; + return Endpoint; + } + } + + private sealed class CountingServiceEndpoint : FakeServiceEndpoint, IDisposable + { + public CountingServiceEndpoint(Type serviceType) : base(serviceType) + { + } + + public int DisposeCount { get; private set; } + + void IDisposable.Dispose() + { + DisposeCount++; + Dispose(); + } + } + } +} diff --git a/Obvs.Tests/Obvs.Tests.csproj b/Obvs.Tests/Obvs.Tests.csproj index 3c838ce6..019581fc 100644 --- a/Obvs.Tests/Obvs.Tests.csproj +++ b/Obvs.Tests/Obvs.Tests.csproj @@ -9,6 +9,7 @@ + diff --git a/Obvs.Tests/TestServiceBus.cs b/Obvs.Tests/TestServiceBus.cs index 42e2b243..14c8e325 100644 --- a/Obvs.Tests/TestServiceBus.cs +++ b/Obvs.Tests/TestServiceBus.cs @@ -25,6 +25,20 @@ public TestServiceBus(ITestOutputHelper testOutputHelper) _testOutputHelper = testOutputHelper; } + [Fact] + public void ShouldDisposeEndpointsOnlyOnceWhenDisposedTwice() + { + IServiceEndpoint serviceEndpoint = A.Fake(); + IServiceEndpointClient serviceEndpointClient = A.Fake(); + var serviceBus = new ServiceBus(new[] { serviceEndpointClient }, new[] { serviceEndpoint }); + + serviceBus.Dispose(); + serviceBus.Dispose(); + + A.CallTo(() => serviceEndpoint.Dispose()).MustHaveHappenedOnceExactly(); + A.CallTo(() => serviceEndpointClient.Dispose()).MustHaveHappenedOnceExactly(); + } + [Fact] public void ShouldOnlySubscribeToUnderlyingEndpointRequestsOnce() { diff --git a/Obvs/Hosting/ObvsBuilder.cs b/Obvs/Hosting/ObvsBuilder.cs new file mode 100644 index 00000000..9ecfebc8 --- /dev/null +++ b/Obvs/Hosting/ObvsBuilder.cs @@ -0,0 +1,79 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using Obvs.Configuration; +using Obvs.Types; + +namespace Obvs.Hosting +{ + /// + /// Configures the services on a bus. Obtained from services.AddObvs(...). + /// + public sealed class ObvsBuilder + { + private readonly List _services = new List(); + + internal ObvsBuilder() + { + } + + /// + /// Adds a service whose messages all implement . + /// + /// The service's name, unique on the bus. Transports use it to name destinations. + public ObvsBuilder AddService(string name, Action configure) + where TServiceMessage : class, IMessage + { + if (string.IsNullOrWhiteSpace(name)) + { + throw new ArgumentException("A service needs a name, which is used to name its destinations on the broker.", nameof(name)); + } + + var service = new ObvsServiceBuilder(name, typeof(TServiceMessage)); + configure?.Invoke(service); + _services.Add(service); + return this; + } + + internal ServiceBus Build(IServiceProvider serviceProvider) + { + Validate(); + + var endpoints = new List>(); + var endpointClients = new List>(); + foreach (var service in _services) + { + var provider = service.EndpointProviderFactory(serviceProvider); + if (service.IsServer) + { + endpoints.Add(provider.CreateEndpoint()); + } + if (service.IsClient) + { + endpointClients.Add(provider.CreateEndpointClient()); + } + } + + return new ServiceBus(endpointClients, endpoints); + } + + private void Validate() + { + var errors = new List(); + + errors.AddRange(_services + .GroupBy(service => service.Name) + .Where(group => group.Count() > 1) + .Select(group => $"Service name '{group.Key}' is used more than once. Each service on a bus needs a unique name.")); + + errors.AddRange(_services + .Where(service => service.EndpointProviderFactory == null) + .Select(service => $"Service '{service.Name}' has no transport. Call UseEndpointProvider(...) on it inside AddService(\"{service.Name}\", ...).")); + + if (errors.Count > 0) + { + throw new ObvsConfigurationException(errors); + } + } + } +} diff --git a/Obvs/Hosting/ObvsConfigurationException.cs b/Obvs/Hosting/ObvsConfigurationException.cs new file mode 100644 index 00000000..1c6b817f --- /dev/null +++ b/Obvs/Hosting/ObvsConfigurationException.cs @@ -0,0 +1,21 @@ +using System; +using System.Collections.Generic; +using System.Linq; + +namespace Obvs.Hosting +{ + /// + /// Thrown when the bus starts with an invalid configuration. Lists every problem found, each with how to fix it. + /// + public sealed class ObvsConfigurationException : Exception + { + public ObvsConfigurationException(IReadOnlyList errors) + : base("The Obvs service bus configuration is invalid:" + Environment.NewLine + + string.Join(Environment.NewLine, errors.Select(error => "- " + error))) + { + Errors = errors; + } + + public IReadOnlyList Errors { get; } + } +} diff --git a/Obvs/Hosting/ObvsHostedService.cs b/Obvs/Hosting/ObvsHostedService.cs new file mode 100644 index 00000000..32a3f793 --- /dev/null +++ b/Obvs/Hosting/ObvsHostedService.cs @@ -0,0 +1,43 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; + +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. + /// + internal sealed class ObvsHostedService : IHostedService + { + private readonly IServiceProvider _serviceProvider; + private readonly string _key; + + public ObvsHostedService(IServiceProvider serviceProvider, string key) + { + _serviceProvider = serviceProvider; + _key = key; + } + + public Task StartAsync(CancellationToken cancellationToken) + { + if (_key == null) + { + _serviceProvider.GetRequiredService(); + } + else + { + _serviceProvider.GetRequiredKeyedService(_key); + } + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken cancellationToken) + { + return Task.CompletedTask; + } + } +} diff --git a/Obvs/Hosting/ObvsServiceBuilder.cs b/Obvs/Hosting/ObvsServiceBuilder.cs new file mode 100644 index 00000000..36d636d1 --- /dev/null +++ b/Obvs/Hosting/ObvsServiceBuilder.cs @@ -0,0 +1,64 @@ +using System; +using Obvs.Configuration; + +namespace Obvs.Hosting +{ + /// + /// Configures one service on the bus: its transport, and whether this app is its client, its server or both. + /// + public sealed class ObvsServiceBuilder + { + internal ObvsServiceBuilder(string name, Type serviceMessageType) + { + Name = name; + ServiceMessageType = serviceMessageType; + } + + /// The service's name, unique on the bus. + public string Name { get; } + + /// The interface every message of this service implements. + public Type ServiceMessageType { get; } + + internal Func EndpointProviderFactory { get; private set; } + + internal bool IsClient { get; private set; } = true; + + internal bool IsServer { get; private set; } = true; + + /// + /// Uses an existing transport's endpoint provider, created from the container when the bus starts. + /// Transports will get their own Use<Transport>() methods; this works with any of them meanwhile. + /// + public ObvsServiceBuilder UseEndpointProvider(Func factory) + { + EndpointProviderFactory = factory ?? throw new ArgumentNullException(nameof(factory)); + return this; + } + + /// This app sends the service's commands and requests, and receives its events and responses. + public ObvsServiceBuilder AsClient() + { + return As(client: true, server: false); + } + + /// This app receives the service's commands and requests, and publishes its events and responses. + public ObvsServiceBuilder AsServer() + { + return As(client: false, server: true); + } + + /// This app is both client and server for the service. The default. + public ObvsServiceBuilder AsClientAndServer() + { + return As(client: true, server: true); + } + + private ObvsServiceBuilder As(bool client, bool server) + { + IsClient = client; + IsServer = server; + return this; + } + } +} diff --git a/Obvs/Hosting/ObvsServiceCollectionExtensions.cs b/Obvs/Hosting/ObvsServiceCollectionExtensions.cs new file mode 100644 index 00000000..22204bbe --- /dev/null +++ b/Obvs/Hosting/ObvsServiceCollectionExtensions.cs @@ -0,0 +1,88 @@ +using System; +using System.Collections.Generic; +using Microsoft.Extensions.Hosting; +using Obvs; +using Obvs.Hosting; + +// In this namespace, as is conventional for AddX extensions, so they're found without an extra using +namespace Microsoft.Extensions.DependencyInjection +{ + public static class ObvsServiceCollectionExtensions + { + /// + /// Adds the service bus to the container: one singleton, resolvable as both + /// and , started and stopped + /// with the host. Calling this more than once adds to the same bus. + /// + public static IServiceCollection AddObvs(this IServiceCollection services, Action configure) + { + return AddObvs(services, key: null, configure); + } + + /// + /// Adds a separate, keyed service bus, resolved with [FromKeyedServices(key)]. + /// Most apps need only the unkeyed bus, which can host many services. + /// + public static IServiceCollection AddObvs(this IServiceCollection services, string key, Action configure) + { + if (services == null) + { + throw new ArgumentNullException(nameof(services)); + } + + var registrations = GetOrAddRegistrations(services); + var registrationKey = key ?? DefaultBusKey; + if (!registrations.TryGetValue(registrationKey, out var builder)) + { + builder = new ObvsBuilder(); + registrations.Add(registrationKey, builder); + AddBus(services, key, builder); + } + + configure?.Invoke(builder); + return services; + } + + // The singletons are registered once per bus; later AddObvs calls only add services to its builder, + // which is read when the bus is first resolved + private static void AddBus(IServiceCollection services, string key, ObvsBuilder builder) + { + if (key == null) + { + services.AddSingleton(serviceProvider => builder.Build(serviceProvider)); + services.AddSingleton(serviceProvider => serviceProvider.GetRequiredService()); + } + else + { + services.AddKeyedSingleton(key, (serviceProvider, _) => builder.Build(serviceProvider)); + services.AddKeyedSingleton(key, (serviceProvider, _) => serviceProvider.GetRequiredKeyedService(key)); + } + + // Not AddHostedService, which would de-duplicate the hosted services of keyed buses + services.AddSingleton(serviceProvider => new ObvsHostedService(serviceProvider, key)); + } + + private static readonly object DefaultBusKey = new object(); + + private static Dictionary GetOrAddRegistrations(IServiceCollection services) + { + foreach (var descriptor in services) + { + if (descriptor.ServiceType == typeof(ObvsRegistrations) && descriptor.ImplementationInstance is ObvsRegistrations existing) + { + return existing.Builders; + } + } + + var registrations = new ObvsRegistrations(); + services.AddSingleton(registrations); + return registrations.Builders; + } + + // Kept in the service collection so repeated AddObvs calls find the same builders + private sealed class ObvsRegistrations + { + public Dictionary Builders { get; } = new Dictionary(); + } + } +} diff --git a/Obvs/IServiceBus.cs b/Obvs/IServiceBus.cs index f227554a..ca2618fb 100644 --- a/Obvs/IServiceBus.cs +++ b/Obvs/IServiceBus.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Linq; using System.Reactive.Linq; +using System.Threading; using System.Threading.Tasks; using Obvs.Configuration; using Obvs.Extensions; @@ -30,6 +31,7 @@ public interface IServiceBus : public class ServiceBus : IServiceBus, IDisposable { private readonly IServiceBus _serviceBus; + private int _disposed; // ReSharper disable once UnusedMember.Global public ServiceBus(IEnumerable> endpointClients, @@ -95,6 +97,11 @@ public Task ReplyAsync(IRequest request, IResponse response) public void Dispose() { + // Safe to call more than once: a container can track the same bus behind several registrations + if (Interlocked.Exchange(ref _disposed, 1) == 1) + { + return; + } ((IDisposable)_serviceBus).Dispose(); } } diff --git a/Obvs/Obvs.csproj b/Obvs/Obvs.csproj index 73bbccbd..72ed7dd1 100644 --- a/Obvs/Obvs.csproj +++ b/Obvs/Obvs.csproj @@ -14,6 +14,8 @@ + +