feat(mqtt): named-broker and broker-per-tenant support (GH-3307)

docs/guide/messaging/transports/mqtt.md

@@ -62,6 +62,40 @@ The MQTT transport does not really support the "Requeue" error handling policy i
 effectively an inline "Retry"
 :::
 
+## Connecting to Multiple MQTT Brokers
+
+If a single Wolverine application needs to talk to more than one MQTT broker, register the additional broker(s)
+with `AddNamedMqttBroker` using a `BrokerName`, then pin publishing or listening to a specific broker with the
+`*OnNamedBroker` overloads:
+
+```csharp
+opts.UseMqtt(mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("primary-broker")));
+
+// An additional, independent MQTT broker identified by name
+opts.AddNamedMqttBroker(new BrokerName("secondary"),
+    mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("secondary-broker")));
+
+// Publish a message type to a topic on a named broker
+opts.PublishMessage<OrderPlaced>()
+    .ToMqttTopicOnNamedBroker(new BrokerName("secondary"), "orders");
+
+// Listen to a topic on a named broker
+opts.ListenToMqttTopicOnNamedBroker(new BrokerName("secondary"), "orders");
+```
+
+::: info
+The Wolverine `Uri` scheme for any endpoint on a named broker is the broker name itself, so in the example above
+you would see endpoint URIs like `secondary://topic/orders`. The default broker keeps the canonical `mqtt://`
+scheme, which keeps the two brokers' endpoints from colliding.
+:::
+
+Each named broker is a completely separate `IManagedMqttClient` with its own per-node response topic, so
+request/reply works independently over each broker.
+
+Connecting to multiple named brokers is distinct from [broker-per-tenant multi-tenancy](#broker-per-tenant): a
+named broker is a statically-addressed second connection that you target explicitly, whereas per-tenant
+connections are selected at runtime from each message's tenant id.
+
 ## Broadcast to User Defined Topics
 
 As long as the MQTT transport is enabled in your application, you can explicitly publish messages to any named topic
@@ -388,6 +422,67 @@ builder.UseWolverine(opts =>
 });
 ```
 
+## Broker per Tenant
+
+Named brokers (above) are a *static* topology: you pin specific endpoints to a specific broker by name at
+configuration time. **Broker-per-tenant** is different — it is *runtime* routing. You declare one shared topic
+topology, and each tenant is served by its **own dedicated MQTT connection** (a distinct broker). Which
+connection a message goes to (and which connection an inbound message came from) is decided at runtime by the
+message's [tenant id](/guide/handlers/multi-tenancy), typically set through `DeliveryOptions.TenantId`:
+
+```csharp
+opts.UseMqtt(mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("shared-broker")))
+
+    // How should Wolverine route a message whose TenantId is null or unknown?
+    // FallbackToDefault (the default) uses the shared connection; TenantIdRequired
+    // throws; IgnoreUnknownTenants silently drops it.
+    .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault)
+
+    // Each tenant gets its OWN dedicated MQTT connection, but shares the
+    // topic topology declared below.
+    .AddTenant("tenant-west",
+        mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("west-broker")))
+
+    .AddTenant("tenant-east",
+        mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("east-broker")));
+
+// One shared topology; messages are routed to the right connection at runtime by
+// Envelope.TenantId (e.g. new DeliveryOptions { TenantId = "tenant-west" }).
+opts.PublishMessage<OrderPlaced>().ToMqttTopic("orders");
+opts.ListenToMqttTopic("orders");
+```
+
+To route a specific message to a tenant's connection, stamp the tenant id on the send:
+
+```csharp
+await bus.SendAsync(new OrderPlaced("blue"), new DeliveryOptions { TenantId = "tenant-west" });
+```
+
+Wolverine wraps the outbound endpoint in a `TenantedSender` that dispatches on `Envelope.TenantId`, and builds a
+compound listener that runs one subscription per tenant connection — each inbound envelope is stamped with the
+tenant id of the connection it was consumed from. This mirrors the
+[RabbitMQ](/guide/messaging/transports/rabbitmq/multi-tenancy) and
+[Azure Service Bus](/guide/messaging/transports/azureservicebus/multitenancy) broker-per-tenant support.
+
+::: warning Unique ClientId per tenant (mandatory)
+MQTT brokers forcibly disconnect a second connection that shares a `ClientId`. Wolverine therefore always gives
+each tenant connection a **unique** `ClientId` derived from the tenant id (`<clientId>-tenant-<tenantId>`), even
+if you pre-set one on the tenant options. This is required so tenant connections never kick each other.
+:::
+
+::: info Shared topology, isolated broker
+Unlike a shared-broker/topic-prefix multi-tenancy model, MQTT tenants use the **same topic string** on each
+tenant's **own broker** — isolation is purely a matter of which broker the tenant's connection points at. One
+consequence: [retained messages](#clearing-out-retained-messages) are per-broker, so a retained message on one
+tenant's broker is not visible to any other tenant.
+:::
+
+::: tip Named broker vs. broker-per-tenant
+Use a **named broker** when a *fixed set of endpoints* should always talk to a *specific* broker. Use
+**broker-per-tenant** when the *same logical endpoints* should be transparently routed to a *different connection
+per tenant* based on the runtime tenant id. They are independent features and can be combined.
+:::
+
 ## Interoperability
 
 ::: tip

src/Transports/MQTT/Wolverine.MQTT.Tests/mqtt_per_tenant_broker_tests.cs

@@ -0,0 +1,182 @@
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine.MQTT.Internals;
+using Wolverine.Tracking;
+using Wolverine.Transports.Sending;
+using Wolverine.Util;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Wolverine.MQTT.Tests;
+
+/// <summary>
+/// Integration coverage for broker-per-tenant MQTT (GH-3307). Two in-process <see cref="LocalMqttBroker"/>
+/// instances stand in for two brokers on two free ports — broker A is the default/shared connection and broker B
+/// is tenant "tenantB"'s own dedicated connection. Because MQTT tenants are always own-connection (no
+/// topic-prefix equivalent), isolation is purely by which broker the tenant's client is connected to. Proves:
+/// <list type="bullet">
+/// <item>a tenant message lands on the tenant's broker (B) and NOT the shared broker (A),</item>
+/// <item>a default/untenanted message falls back to the shared broker (A) and NOT the tenant broker (B),</item>
+/// <item>an inbound tenant message is consumed and stamped with its <c>TenantId</c>, and</item>
+/// <item>each tenant connection is given a unique ClientId so the broker never kicks it.</item>
+/// </list>
+/// </summary>
+[Collection("acceptance")]
+public class mqtt_per_tenant_broker_tests : IAsyncLifetime
+{
+    private readonly ITestOutputHelper _output;
+    private LocalMqttBroker _brokerA = null!;
+    private LocalMqttBroker _brokerB = null!;
+    private int _portA;
+    private int _portB;
+
+    public mqtt_per_tenant_broker_tests(ITestOutputHelper output) => _output = output;
+
+    public async Task InitializeAsync()
+    {
+        _portA = PortFinder.GetAvailablePort();
+        _portB = PortFinder.GetAvailablePort();
+
+        _brokerA = new LocalMqttBroker(_portA) { Logger = new XUnitLogger(_output, "MQTT-A") };
+        _brokerB = new LocalMqttBroker(_portB) { Logger = new XUnitLogger(_output, "MQTT-B") };
+
+        await _brokerA.StartAsync();
+        await _brokerB.StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await _brokerA.DisposeAsync();
+        await _brokerB.DisposeAsync();
+    }
+
+    [Fact]
+    public async Task tenant_message_is_published_to_the_tenant_broker_and_not_the_default()
+    {
+        var topic = $"tenant/{Guid.NewGuid():N}";
+
+        await using var subOnB = await RawMqttSubscriber.StartAsync(_portB, topic);
+        await using var subOnA = await RawMqttSubscriber.StartAsync(_portA, topic);
+
+        using var host = await buildSenderAsync(topic);
+
+        await host.MessageBus().SendAsync(new TenantColor("for-tenant-b"),
+            new DeliveryOptions { TenantId = "tenantB" });
+
+        // Landed on the tenant's broker (B)...
+        (await subOnB.WaitForMessageAsync(15.Seconds())).ShouldBeTrue();
+        // ...and NOT on the shared/default broker (A).
+        (await subOnA.WaitForMessageAsync(2.Seconds())).ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task default_message_falls_back_to_the_shared_broker()
+    {
+        var topic = $"tenant/{Guid.NewGuid():N}";
+
+        await using var subOnA = await RawMqttSubscriber.StartAsync(_portA, topic);
+        await using var subOnB = await RawMqttSubscriber.StartAsync(_portB, topic);
+
+        using var host = await buildSenderAsync(topic);
+
+        await host.MessageBus().SendAsync(new TenantColor("no-tenant"));
+
+        // Falls back to the shared/default broker (TenantedIdBehavior.FallbackToDefault)...
+        (await subOnA.WaitForMessageAsync(15.Seconds())).ShouldBeTrue();
+        // ...and NOT the tenant broker.
+        (await subOnB.WaitForMessageAsync(2.Seconds())).ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task tenant_message_is_consumed_and_stamped_with_the_tenant_id()
+    {
+        var topic = $"tenant/{Guid.NewGuid():N}";
+
+        using var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.ServiceName = "PerTenantInbound";
+                configureTransport(opts, topic);
+
+                opts.Policies.DisableConventionalLocalRouting();
+                opts.PublishMessage<TenantColor>().ToMqttTopic(topic).SendInline();
+                opts.ListenToMqttTopic(topic);
+            }).StartAsync();
+
+        // The default listener consumes broker A and the tenant listener consumes broker B; the message only
+        // exists on broker B, so only the tenant listener consumes it and stamps the tenant id.
+        var session = await host
+            .TrackActivity()
+            .Timeout(60.Seconds())
+            .WaitForMessageToBeReceivedAt<TenantColor>(host)
+            .ExecuteAndWaitAsync(c =>
+                c.SendAsync(new TenantColor("for-tenant-b"), new DeliveryOptions { TenantId = "tenantB" }));
+
+        var received = session.Received.SingleEnvelope<TenantColor>();
+        received.TenantId.ShouldBe("tenantB");
+        received.Message.ShouldBeOfType<TenantColor>().Color.ShouldBe("for-tenant-b");
+    }
+
+    [Fact]
+    public async Task tenant_client_gets_a_unique_client_id()
+    {
+        var topic = $"tenant/{Guid.NewGuid():N}";
+        using var host = await buildSenderAsync(topic);
+
+        var transport = host.GetRuntime().Options.Transports.GetOrCreate<MqttTransport>();
+
+        var tenant = transport.Tenants["tenantB"];
+        tenant.Client.ShouldNotBeNull();
+
+        // The tenant ClientId is unique and identifiable — required so the broker doesn't kick a duplicate.
+        tenant.Options.ClientOptions.ClientId.ShouldEndWith("-tenant-tenantB");
+        tenant.Options.ClientOptions.ClientId.ShouldNotBe(transport.Options.ClientOptions.ClientId);
+    }
+
+    [Fact]
+    public async Task tenant_aware_endpoint_resolves_a_TenantedSender()
+    {
+        var topic = $"tenant/{Guid.NewGuid():N}";
+        using var host = await buildSenderAsync(topic);
+
+        var runtime = host.GetRuntime();
+        var transport = runtime.Options.Transports.GetOrCreate<MqttTransport>();
+        var endpoint = transport.Topics[topic];
+
+        // buildSenderAsync publishes inline, so the endpoint resolves an InlineSendingAgent around our sender.
+        var agent = (InlineSendingAgent)runtime.Endpoints.GetOrBuildSendingAgent(endpoint.Uri);
+        agent.Sender.ShouldBeOfType<TenantedSender>();
+    }
+
+    private void configureTransport(WolverineOptions opts, string topic)
+    {
+        opts.UseMqttWithLocalBroker(_portA)
+            .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault)
+            .AddTenant("tenantB", builder =>
+                builder.WithClientOptions(o => o.WithTcpServer("127.0.0.1", _portB)));
+    }
+
+    private Task<IHost> buildSenderAsync(string topic)
+    {
+        return Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.ServiceName = "PerTenantSender";
+                configureTransport(opts, topic);
+
+                opts.Policies.DisableConventionalLocalRouting();
+                opts.PublishMessage<TenantColor>().ToMqttTopic(topic).SendInline();
+            }).StartAsync();
+    }
+}
+
+public record TenantColor(string Color);
+
+public static class TenantColorHandler
+{
+    public static void Handle(TenantColor message)
+    {
+        // no-op; presence lets Wolverine discover a handler so receive tests can track processing
+    }
+}