diff --git a/docs/guide/messaging/transports/mqtt.md b/docs/guide/messaging/transports/mqtt.md index 52b9acd92..3d0ba4f36 100644 --- a/docs/guide/messaging/transports/mqtt.md +++ b/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() + .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().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 (`-tenant-`), 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 diff --git a/src/Transports/MQTT/Wolverine.MQTT.Tests/mqtt_per_tenant_broker_tests.cs b/src/Transports/MQTT/Wolverine.MQTT.Tests/mqtt_per_tenant_broker_tests.cs new file mode 100644 index 000000000..1acebf7fb --- /dev/null +++ b/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; + +/// +/// Integration coverage for broker-per-tenant MQTT (GH-3307). Two in-process +/// 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: +/// +/// a tenant message lands on the tenant's broker (B) and NOT the shared broker (A), +/// a default/untenanted message falls back to the shared broker (A) and NOT the tenant broker (B), +/// an inbound tenant message is consumed and stamped with its TenantId, and +/// each tenant connection is given a unique ClientId so the broker never kicks it. +/// +/// +[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().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(host) + .ExecuteAndWaitAsync(c => + c.SendAsync(new TenantColor("for-tenant-b"), new DeliveryOptions { TenantId = "tenantB" })); + + var received = session.Received.SingleEnvelope(); + received.TenantId.ShouldBe("tenantB"); + received.Message.ShouldBeOfType().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(); + + 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(); + 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(); + } + + 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 buildSenderAsync(string topic) + { + return Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.ServiceName = "PerTenantSender"; + configureTransport(opts, topic); + + opts.Policies.DisableConventionalLocalRouting(); + opts.PublishMessage().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 + } +} diff --git a/src/Transports/MQTT/Wolverine.MQTT.Tests/named_broker_tests.cs b/src/Transports/MQTT/Wolverine.MQTT.Tests/named_broker_tests.cs new file mode 100644 index 000000000..c4c1aa734 --- /dev/null +++ b/src/Transports/MQTT/Wolverine.MQTT.Tests/named_broker_tests.cs @@ -0,0 +1,304 @@ +using System.Text; +using JasperFx.Core; +using Microsoft.Extensions.Hosting; +using MQTTnet; +using MQTTnet.Extensions.ManagedClient; +using Shouldly; +using Wolverine.Configuration; +using Wolverine.MQTT.Internals; +using Wolverine.Tracking; +using Wolverine.Util; +using Xunit; +using Xunit.Abstractions; + +namespace Wolverine.MQTT.Tests; + +/// +/// Registration-level coverage for named MQTT brokers that needs no running broker, so it always runs in CI. +/// A named broker is a second, independent whose Protocol (and therefore its +/// endpoints' URI scheme) is the broker name, so its endpoints never collide with the default mqtt:// +/// broker. +/// +public class NamedMqttBrokerRegistrationTests +{ + private readonly BrokerName theName = new("secondary"); + + [Fact] + public void adds_a_distinct_transport_instance_per_broker() + { + var options = new WolverineOptions(); + options.UseMqttWithLocalBroker(1883); + options.AddNamedMqttBroker(theName, builder => + builder.WithClientOptions(o => o.WithTcpServer("127.0.0.1", 1884))); + + var transports = options.Transports.OfType().ToList(); + transports.Count.ShouldBe(2); + transports.Select(x => x.Protocol).OrderBy(x => x).ShouldBe(["mqtt", "secondary"]); + } + + [Fact] + public void named_broker_endpoints_use_the_broker_name_as_their_uri_scheme() + { + var options = new WolverineOptions(); + options.UseMqttWithLocalBroker(1883); + options.AddNamedMqttBroker(theName, builder => + builder.WithClientOptions(o => o.WithTcpServer("127.0.0.1", 1884))); + + var named = options.Transports.OfType().Single(x => x.Protocol == "secondary"); + named.Topics["incoming/one"].Uri.ShouldBe(new Uri("secondary://topic/incoming/one")); + + // ...and the default broker keeps the canonical mqtt:// scheme. + var @default = options.Transports.OfType().Single(x => x.Protocol == "mqtt"); + @default.Topics["incoming/one"].Uri.ShouldBe(new Uri("mqtt://topic/incoming/one")); + } + + [Fact] + public void listen_on_named_broker_registers_the_listener_on_the_named_transport_only() + { + var options = new WolverineOptions(); + options.UseMqttWithLocalBroker(1883); + options.AddNamedMqttBroker(theName, builder => + builder.WithClientOptions(o => o.WithTcpServer("127.0.0.1", 1884))); + + options.ListenToMqttTopicOnNamedBroker(theName, "incoming/one"); + + var named = options.Transports.OfType().Single(x => x.Protocol == "secondary"); + named.Topics["incoming/one"].IsListener.ShouldBeTrue(); + } +} + +/// +/// End-to-end coverage that a named MQTT broker actually talks to a different broker than the default +/// one. Two in-process instances stand in for two brokers (A = default, B = named) +/// on two free ports — no Docker needed. Proves: +/// +/// a message published on the named broker lands on broker B and not broker A, +/// a default publish lands on broker A and not broker B, +/// a message published and consumed on the named broker round-trips and arrives stamped with the +/// broker's own URI scheme (the receive pipeline sets Destination from the listener endpoint's URI), and +/// a full request/reply round-trips over the named broker, proving the reply is routed back to broker B +/// rather than the default broker A. +/// +/// +[Collection("acceptance")] +public class named_broker_tests : IAsyncLifetime +{ + private readonly ITestOutputHelper _output; + private readonly BrokerName theName = new("secondary"); + private LocalMqttBroker _brokerA = null!; + private LocalMqttBroker _brokerB = null!; + private int _portA; + private int _portB; + + public named_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 named_broker_send_lands_on_the_named_broker() + { + var topic = $"named/{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, useNamedBroker: true); + + await host.MessageBus().SendAsync(new NamedColor("on-named-broker")); + + // Landed on broker B (the named broker's connection)... + (await subOnB.WaitForMessageAsync(15.Seconds())).ShouldBeTrue(); + // ...and NOT on the default broker A. + (await subOnA.WaitForMessageAsync(2.Seconds())).ShouldBeFalse(); + } + + [Fact] + public async Task default_broker_send_lands_on_the_default_broker() + { + var topic = $"named/{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, useNamedBroker: false); + + await host.MessageBus().SendAsync(new NamedColor("on-default-broker")); + + (await subOnA.WaitForMessageAsync(15.Seconds())).ShouldBeTrue(); + (await subOnB.WaitForMessageAsync(2.Seconds())).ShouldBeFalse(); + } + + [Fact] + public async Task round_trips_a_message_over_the_named_broker() + { + var topic = $"named/{Guid.NewGuid():N}"; + + // A single host both publishes and listens on the named broker (broker B). On receipt the pipeline stamps + // Destination from the listener endpoint's URI, so the consumed envelope carries the named broker's + // "secondary" scheme rather than the default "mqtt". + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.UseMqttWithLocalBroker(_portA); + opts.AddNamedMqttBroker(theName, builder => + builder.WithClientOptions(o => o.WithTcpServer("127.0.0.1", _portB))); + + opts.Policies.DisableConventionalLocalRouting(); + opts.PublishMessage().ToMqttTopicOnNamedBroker(theName, topic).SendInline(); + opts.ListenToMqttTopicOnNamedBroker(theName, topic); + }).StartAsync(); + + var session = await host + .TrackActivity() + .Timeout(30.Seconds()) + .WaitForMessageToBeReceivedAt(host) + .ExecuteAndWaitAsync(c => c.SendAsync(new NamedColor("round-trip"))); + + var received = session.Received.SingleEnvelope(); + received.Message.ShouldBeOfType().Color.ShouldBe("round-trip"); + received.Destination!.Scheme.ShouldBe("secondary"); + } + + [Fact] + public async Task request_reply_round_trips_over_the_named_broker() + { + var topic = $"named/rr/{Guid.NewGuid():N}"; + + // Request/reply entirely over the named broker (broker B). The reply is routed by the reply-uri's scheme, + // which the pipeline corrects to the receiving endpoint's scheme ("secondary") — so the response travels + // back over broker B, not the default broker A. If reply routing resolved to the default broker, + // InvokeAndWaitAsync would time out. + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.UseMqttWithLocalBroker(_portA); + opts.AddNamedMqttBroker(theName, builder => + builder.WithClientOptions(o => o.WithTcpServer("127.0.0.1", _portB))); + + opts.Policies.DisableConventionalLocalRouting(); + opts.PublishMessage().ToMqttTopicOnNamedBroker(theName, topic); + opts.ListenToMqttTopicOnNamedBroker(theName, topic); + }).StartAsync(); + + var (_, response) = await host + .TrackActivity() + .Timeout(30.Seconds()) + .InvokeAndWaitAsync(new NamedPing("named-rr")); + + response.ShouldNotBeNull(); + response.Name.ShouldBe("named-rr"); + } + + private Task buildSenderAsync(string topic, bool useNamedBroker) + { + return Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.UseMqttWithLocalBroker(_portA); + opts.AddNamedMqttBroker(theName, builder => + builder.WithClientOptions(o => o.WithTcpServer("127.0.0.1", _portB))); + + opts.Policies.DisableConventionalLocalRouting(); + + if (useNamedBroker) + { + opts.PublishMessage().ToMqttTopicOnNamedBroker(theName, topic).SendInline(); + } + else + { + opts.PublishMessage().ToMqttTopic(topic).SendInline(); + } + }).StartAsync(); + } +} + +/// +/// A minimal raw MQTTnet subscriber used to assert which physical broker a Wolverine publish actually reached, +/// independent of Wolverine's own routing. +/// +internal class RawMqttSubscriber : IAsyncDisposable +{ + private readonly IManagedMqttClient _client; + private readonly List _messages = new(); + + private RawMqttSubscriber(IManagedMqttClient client) => _client = client; + + public static async Task StartAsync(int port, string topic) + { + var client = new MqttFactory().CreateManagedMqttClient(); + var subscriber = new RawMqttSubscriber(client); + + client.ApplicationMessageReceivedAsync += e => + { + lock (subscriber._messages) + { + subscriber._messages.Add(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment)); + } + + return Task.CompletedTask; + }; + + await client.StartAsync(new ManagedMqttClientOptionsBuilder() + .WithClientOptions(o => o.WithTcpServer("127.0.0.1", port)).Build()); + await client.SubscribeAsync(topic); + + return subscriber; + } + + public async Task WaitForMessageAsync(TimeSpan timeout) + { + var deadline = DateTimeOffset.UtcNow.Add(timeout); + while (DateTimeOffset.UtcNow < deadline) + { + lock (_messages) + { + if (_messages.Count > 0) return true; + } + + await Task.Delay(50); + } + + return false; + } + + public async ValueTask DisposeAsync() + { + await _client.StopAsync(); + _client.Dispose(); + } +} + +public record NamedColor(string Color); + +public static class NamedColorHandler +{ + public static void Handle(NamedColor message) + { + // no-op; presence lets Wolverine discover a handler so receive tests can track processing + } +} + +public record NamedPing(string Name); + +public record NamedPong(string Name); + +public class NamedPingHandler +{ + public NamedPong Handle(NamedPing ping) => new(ping.Name); +} diff --git a/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttHealthCheck.cs b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttHealthCheck.cs index 325790465..dc5ea4704 100644 --- a/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttHealthCheck.cs +++ b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttHealthCheck.cs @@ -9,7 +9,7 @@ internal class MqttHealthCheck : WolverineTransportHealthCheck public MqttHealthCheck(MqttTransport transport) => _transport = transport; public override string TransportName => "MQTT"; - public override string Protocol => "mqtt"; + public override string Protocol => _transport.Protocol; public override Task CheckHealthAsync(CancellationToken cancellationToken = default) { diff --git a/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttListener.cs b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttListener.cs index 9455be67d..604189d4f 100644 --- a/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttListener.cs +++ b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttListener.cs @@ -3,6 +3,7 @@ using JasperFx.Core; using Microsoft.Extensions.Logging; using MQTTnet.Client; +using MQTTnet.Extensions.ManagedClient; using Wolverine.Runtime; using Wolverine.Transports; @@ -18,17 +19,24 @@ internal class MqttListener : IListener, IReportConnectionState private readonly IReceiver _receiver; private readonly MqttTopic _topic; + // Broker-per-tenant (GH-3307): the managed client this listener consumes from — the default/shared client + // for the untenanted listener, or a tenant's own dedicated client. ConnectionState reports on this specific + // connection so a dropped tenant connection is surfaced independently of the default one. + private readonly IManagedMqttClient _client; + // GH-3231: surface the managed MQTT client's connection state so external monitors can detect a listener whose // broker connection has dropped (and not resubscribed) while it still reports Accepting. public TransportConnectionState ConnectionState => - _broker.Client.IsConnected ? TransportConnectionState.Connected : TransportConnectionState.Disconnected; + _client.IsConnected ? TransportConnectionState.Connected : TransportConnectionState.Disconnected; - public MqttListener(MqttTransport broker, ILogger logger, MqttTopic topic, IReceiver receiver) + public MqttListener(MqttTransport broker, ILogger logger, MqttTopic topic, IReceiver receiver, + IManagedMqttClient client) { _broker = broker; _logger = logger; _topic = topic; _receiver = receiver; + _client = client; Address = topic.Uri; TopicName = topic.ListeningTopic; diff --git a/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTenant.cs b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTenant.cs new file mode 100644 index 000000000..9db5090ff --- /dev/null +++ b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTenant.cs @@ -0,0 +1,40 @@ +using MQTTnet.Extensions.ManagedClient; + +namespace Wolverine.MQTT.Internals; + +/// +/// Represents a single Wolverine tenant that is served by its own dedicated MQTT connection while sharing the +/// topic topology declared on the parent transport. Unlike NATS, MQTT tenants are always +/// own-connection: there is no subject/topic-prefix equivalent, so isolation is purely a matter of which broker +/// the tenant's dedicated is connected to. Outbound routing is by +/// via the framework's +/// ; inbound the tenant's own listener stamps the +/// tenant id. Mirrors the RabbitMQ / Azure Service Bus tenant model. +/// +internal class MqttTenant +{ + public MqttTenant(string tenantId) + { + TenantId = tenantId ?? throw new ArgumentNullException(nameof(tenantId)); + } + + public string TenantId { get; } + + /// + /// The tenant's own connection options. Seeded from the parent transport at registration time and then + /// overridden with the tenant-specific broker configuration. + /// + public ManagedMqttClientOptions Options { get; set; } = new() + { ClientOptions = new MQTTnet.Client.MqttClientOptions() }; + + /// + /// Optional JWT authentication configuration for the tenant's own connection. + /// + public MqttJwtAuthenticationOptions? Jwt { get; set; } + + /// + /// The tenant's dedicated managed MQTT client. Created and owned by the transport during + /// , and stopped with the transport. + /// + internal IManagedMqttClient? Client { get; set; } +} diff --git a/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTopicSender.cs b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTopicSender.cs new file mode 100644 index 000000000..a1e9aee8b --- /dev/null +++ b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTopicSender.cs @@ -0,0 +1,47 @@ +using MQTTnet.Extensions.ManagedClient; +using Wolverine.Transports.Sending; + +namespace Wolverine.MQTT.Internals; + +/// +/// A standalone bound to a specific . Broker-per-tenant +/// (GH-3307) uses one of these per tenant (bound to the tenant's dedicated client) beneath a +/// , plus one bound to the default client as the fallback path. +/// +/// It enqueues directly to the managed client (fire-and-forget) and deliberately does NOT implement +/// : does not forward RegisterCallback to the +/// senders beneath it, so a callback-requiring sender there would silently drop every message (GH-2361). +/// +internal class MqttTopicSender : ISender +{ + private readonly MqttTopic _topic; + private readonly IManagedMqttClient _client; + + public MqttTopicSender(MqttTopic topic, IManagedMqttClient client) + { + _topic = topic; + _client = client; + } + + public bool SupportsNativeScheduledSend => false; + public Uri Destination => _topic.Uri; + + public async Task PingAsync() + { + try + { + await _client.PingAsync(); + return true; + } + catch (Exception) + { + return false; + } + } + + public ValueTask SendAsync(Envelope envelope) + { + var message = _topic.BuildMessage(envelope); + return new ValueTask(_client.EnqueueAsync(message)); + } +} diff --git a/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTransport.cs b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTransport.cs index 9b074c990..66faf34a5 100644 --- a/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTransport.cs +++ b/src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTransport.cs @@ -10,6 +10,7 @@ using Wolverine.Runtime; using Wolverine.Runtime.Routing; using Wolverine.Transports; +using Wolverine.Transports.Sending; namespace Wolverine.MQTT.Internals; @@ -17,23 +18,77 @@ public class MqttTransport : TransportBase, IAsyncDisposable { [IgnoreDescription] public LightweightCache Topics { get; } = new(); - private List _listeners = new(); - private ImHashMap _topicListeners = ImHashMap.Empty; - private bool _subscribed; + + // Named broker + broker-per-tenant (GH-3307): listener bookkeeping is per-managed-client rather than + // transport-global, because each tenant runs on its own IManagedMqttClient and messages received on a + // tenant connection must resolve to that tenant's (tenant-id stamping) listener, not the default one. + private readonly Dictionary _clientGroups = new(); private ILogger _logger = null!; private CancellationTokenSource? _refreshCts; + // Broker-per-tenant (GH-3307): CancellationTokenSources for each tenant client's JWT re-auth loop, so they + // can be cancelled on shutdown. + private readonly List _tenantRefreshCts = new(); + public static string TopicForUri(Uri uri) { if (uri == null) return string.Empty; return uri.LocalPath.Trim('/'); } - public MqttTransport() : base("mqtt", "MQTT Transport", ["mqtt"]) + public MqttTransport() : this("mqtt") + { + } + + /// + /// Constructor used when connecting to more than one MQTT broker from a single application. The + /// doubles as the additional broker's URI scheme so its endpoints don't + /// collide with the default mqtt:// broker. Reached through + /// when a is supplied. + /// + public MqttTransport(string protocol) : base(protocol, "MQTT Transport", [protocol]) { Topics.OnMissing = topicName => new MqttTopic(topicName, this, EndpointRole.Application); } + /// + /// Broker-per-tenant registrations (GH-3307). Each tenant owns its own dedicated + /// pointed at its own broker while sharing the topic topology declared on the parent transport; outbound is + /// routed by and inbound listeners stamp the tenant id. + /// + [IgnoreDescription] + internal LightweightCache Tenants { get; } = new(name => new MqttTenant(name)); + + /// + /// Controls how an outbound message whose tenant id is null or unregistered is routed when broker-per-tenant + /// multi-tenancy is in effect. Defaults to . + /// + public TenantedIdBehavior TenantedIdBehavior { get; set; } = TenantedIdBehavior.FallbackToDefault; + + /// + /// The managed client that a tenant's traffic should flow over, falling back to the default/shared client for + /// tenants that were never registered (or the untenanted path). + /// + internal IManagedMqttClient GetTenantClient(MqttTenant tenant) => tenant.Client ?? Client; + + private sealed class ClientListenerGroup + { + public List Listeners { get; } = new(); + public ImHashMap TopicListeners = ImHashMap.Empty; + public bool Subscribed; + } + + private ClientListenerGroup groupFor(IManagedMqttClient client) + { + if (!_clientGroups.TryGetValue(client, out var group)) + { + group = new ClientListenerGroup(); + _clientGroups[client] = group; + } + + return group; + } + protected override IEnumerable endpoints() { return Topics; @@ -83,29 +138,120 @@ public override async ValueTask InitializeAsync(IWolverineRuntime runtime) Client.ConnectedAsync += onClientConnected; Client.DisconnectedAsync += onClientDisconnected; await Client.StartAsync(Options); + + // Broker-per-tenant (GH-3307): each tenant gets its own dedicated managed client, connected to the + // tenant's own broker. MQTT brokers forcibly disconnect a second connection that shares a ClientId, so + // each tenant client is given a UNIQUE ClientId derived from the tenant id (see buildTenantClientAsync). + foreach (var tenant in Tenants) + { + tenant.Client = await buildTenantClientAsync(runtime, mqttFactory, tenant); + } + foreach (var endpoint in Topics) { endpoint.Compile(runtime); } } + private async Task buildTenantClientAsync(IWolverineRuntime runtime, MqttFactory mqttFactory, + MqttTenant tenant) + { + var logger = new MqttNetLogger(runtime.LoggerFactory.CreateLogger()); + var client = mqttFactory.CreateManagedMqttClient(logger); + + var options = tenant.Options; + options.ClientOptions.ProtocolVersion = MqttProtocolVersion.V500; + + // Enforce a unique ClientId for the tenant connection even if the user pre-set one on the tenant options. + // Two managed clients sharing a ClientId make the broker kick one of them, so tenant traffic would drop. + var baseClientId = options.ClientOptions.ClientId; + options.ClientOptions.ClientId = + (string.IsNullOrEmpty(baseClientId) ? Guid.NewGuid().ToString("N") : baseClientId) + + "-tenant-" + tenant.TenantId; + + if (tenant.Jwt is not null) + { + options.ClientOptions.AuthenticationMethod = "OAUTH2-JWT"; + options.ClientOptions.AuthenticationData = await tenant.Jwt.GetTokenCallBack(); + client.ConnectedAsync += buildTenantJwtRefreshHandler(client, tenant.Jwt); + } + + await client.StartAsync(options); + + return client; + } + + // Broker-per-tenant JWT re-authentication loop, scoped to a single tenant client. Mirrors the default + // client's onClientConnected refresh, but keyed to the tenant's own connection and token callback. + private Func buildTenantJwtRefreshHandler(IManagedMqttClient client, + MqttJwtAuthenticationOptions jwt) + { + CancellationTokenSource? refreshCts = null; + return async arg => + { + if (arg.ConnectResult.ResultCode != MqttClientConnectResultCode.Success) + { + return; + } + + if (refreshCts is not null) + { + await refreshCts.CancelAsync(); + refreshCts.Dispose(); + } + + refreshCts = new CancellationTokenSource(); + lock (_tenantRefreshCts) + { + _tenantRefreshCts.Add(refreshCts); + } + + var ct = refreshCts.Token; + _ = Task.Run(async () => + { + using var periodicTimer = new PeriodicTimer(jwt.RefreshPeriod); + try + { + while (await periodicTimer.WaitForNextTickAsync(ct).ConfigureAwait(false)) + { + if (!client.IsConnected) continue; + + await client.InternalClient + .SendExtendedAuthenticationExchangeDataAsync( + new MqttExtendedAuthenticationExchangeData + { + AuthenticationData = await jwt.GetTokenCallBack(), + ReasonCode = MQTTnet.Protocol.MqttAuthenticateReasonCode.ReAuthenticate + }, ct) + .ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Shutdown requested + } + }, ct); + }; + } + public WolverineTransportHealthCheck BuildHealthCheck(IWolverineRuntime runtime) { return new MqttHealthCheck(this); } - private void startSubscribing() + private void startSubscribing(IManagedMqttClient client) { - if (_subscribed) return; + var group = groupFor(client); + if (group.Subscribed) return; - Client.ApplicationMessageReceivedAsync += receiveAsync; - _subscribed = true; + client.ApplicationMessageReceivedAsync += arg => receiveAsync(client, arg); + group.Subscribed = true; } - private Task receiveAsync(MqttApplicationMessageReceivedEventArgs arg) + private Task receiveAsync(IManagedMqttClient client, MqttApplicationMessageReceivedEventArgs arg) { var topicName = arg.ApplicationMessage.Topic; - if (tryFindListener(topicName, out var listener)) + if (tryFindListener(client, topicName, out var listener)) { return listener.ReceiveAsync(arg); } @@ -168,17 +314,18 @@ private async Task onClientDisconnected(MqttClientDisconnectedEventArgs arg) } } - internal bool tryFindListener(string topicName, out MqttListener listener) + internal bool tryFindListener(IManagedMqttClient client, string topicName, out MqttListener listener) { - if (_topicListeners.TryFind(topicName, out listener)) + var group = groupFor(client); + if (group.TopicListeners.TryFind(topicName, out listener)) { return listener is not null; } - listener = (_listeners.FirstOrDefault(x => x.TopicName == topicName) ?? _listeners.FirstOrDefault(x => + listener = (group.Listeners.FirstOrDefault(x => x.TopicName == topicName) ?? group.Listeners.FirstOrDefault(x => MqttTopicFilterComparer.Compare(topicName, x.TopicName) == MqttTopicFilterCompareResult.IsMatch))!; - _topicListeners = _topicListeners.AddOrUpdate(topicName, listener!); + group.TopicListeners = group.TopicListeners.AddOrUpdate(topicName, listener!); return listener is not null; @@ -223,21 +370,52 @@ public async ValueTask DisposeAsync() await _refreshCts.CancelAsync(); _refreshCts.Dispose(); } + + CancellationTokenSource[] tenantCts; + lock (_tenantRefreshCts) + { + tenantCts = _tenantRefreshCts.ToArray(); + _tenantRefreshCts.Clear(); + } + + foreach (var cts in tenantCts) + { + try + { + await cts.CancelAsync(); + cts.Dispose(); + } + catch (ObjectDisposedException) + { + } + } + if (Client is not null) await Client.StopAsync(); + + // Broker-per-tenant (GH-3307): each tenant owns its own managed client; stop them too. + foreach (var tenant in Tenants) + { + if (tenant.Client is not null) + { + await tenant.Client.StopAsync(); + } + } } catch (ObjectDisposedException) { } } - internal async ValueTask SubscribeToTopicAsync(string topicName, MqttListener listener, MqttTopic mqttTopic) + internal async ValueTask SubscribeToTopicAsync(string topicName, MqttListener listener, MqttTopic mqttTopic, + IManagedMqttClient client) { - _listeners.Add(listener); + var group = groupFor(client); + group.Listeners.Add(listener); - await Client.SubscribeAsync(topicName, mqttTopic.QualityOfServiceLevel); + await client.SubscribeAsync(topicName, mqttTopic.QualityOfServiceLevel); - startSubscribing(); + startSubscribing(client); } private int _senderIndex = 0; diff --git a/src/Transports/MQTT/Wolverine.MQTT/MqttEndpointUri.cs b/src/Transports/MQTT/Wolverine.MQTT/MqttEndpointUri.cs index 895a3eeea..1a9c6bef2 100644 --- a/src/Transports/MQTT/Wolverine.MQTT/MqttEndpointUri.cs +++ b/src/Transports/MQTT/Wolverine.MQTT/MqttEndpointUri.cs @@ -16,7 +16,23 @@ public static class MqttEndpointUri /// Thrown when is null, empty, or whitespace. public static Uri Topic(string topicName) { + return Topic("mqtt", topicName); + } + + /// + /// Builds a URI referencing an MQTT topic endpoint in the canonical form + /// {protocol}://topic/{topicName}. The is the transport's URI scheme, + /// which for a named broker (see AddNamedMqttBroker) is the broker name rather than mqtt. + /// Slashes inside the topic name are preserved as path separators. + /// + /// The transport's URI scheme (e.g. mqtt or a named broker's name). + /// The MQTT topic name (may contain slashes). + /// A of the form {protocol}://topic/{topicName}. + /// Thrown when or is null, empty, or whitespace. + public static Uri Topic(string protocol, string topicName) + { + ArgumentException.ThrowIfNullOrWhiteSpace(protocol); ArgumentException.ThrowIfNullOrWhiteSpace(topicName); - return new Uri("mqtt://topic/" + topicName.Trim('/')); + return new Uri($"{protocol}://topic/" + topicName.Trim('/')); } } diff --git a/src/Transports/MQTT/Wolverine.MQTT/MqttTopic.cs b/src/Transports/MQTT/Wolverine.MQTT/MqttTopic.cs index aad68582b..9ebcff7ef 100644 --- a/src/Transports/MQTT/Wolverine.MQTT/MqttTopic.cs +++ b/src/Transports/MQTT/Wolverine.MQTT/MqttTopic.cs @@ -19,7 +19,7 @@ public class MqttTopic : Endpoint, ISender, ITopicEndpoint public string TopicName { get; } private CancellationToken _cancellation; - public MqttTopic(string topicName, MqttTransport parent, EndpointRole role) : base(new Uri("mqtt://topic/" + topicName.Trim('/')), role) + public MqttTopic(string topicName, MqttTransport parent, EndpointRole role) : base(MqttEndpointUri.Topic(parent.Protocol, topicName), role) { TopicName = topicName.Trim('/'); Parent = parent; @@ -76,8 +76,29 @@ public override async ValueTask BuildListenerAsync(IWolverineRuntime var logger = runtime.LoggerFactory.CreateLogger(); - var listener = new MqttListener(Parent, logger, this, receiver); - await Parent.SubscribeToTopicAsync(TopicName, listener, this); + var listener = new MqttListener(Parent, logger, this, receiver, Parent.Client); + await Parent.SubscribeToTopicAsync(TopicName, listener, this, Parent.Client); + + // Broker-per-tenant (GH-3307): the shared listener consumes the default connection. Each tenant runs its + // own listener on its own dedicated client, stamping the tenant id onto inbound envelopes via + // TenantIdRule. Per-envelope completion routes back over the receiving connection through + // Envelope.Listener — the same CompoundListener multi-tenancy pattern used by RabbitMQ / NATS / SQS. + if (Parent.Tenants.Any() && TenancyBehavior == TenancyBehavior.TenantAware) + { + var compound = new CompoundListener(Uri); + compound.Inner.Add(listener); + + foreach (var tenant in Parent.Tenants) + { + var client = Parent.GetTenantClient(tenant); + var tenantReceiver = new ReceiverWithRules(receiver, [new TenantIdRule(tenant.TenantId)]); + var tenantListener = new MqttListener(Parent, logger, this, tenantReceiver, client); + await Parent.SubscribeToTopicAsync(TopicName, tenantListener, this, client); + compound.Inner.Add(tenantListener); + } + + return compound; + } return listener; } @@ -87,6 +108,27 @@ public override async ValueTask BuildListenerAsync(IWolverineRuntime protected override ISender CreateSender(IWolverineRuntime runtime) { Compile(runtime); + + // Broker-per-tenant (GH-3307): route by Envelope.TenantId to a per-tenant sender bound to that tenant's + // own connection, falling back to the shared/default connection for the untenanted path. + // + // Both the tenant senders AND the default sender they fall back to are simple fire-and-forget + // MqttTopicSenders: TenantedSender intentionally does NOT implement ISenderRequiresCallback (GH-2361) and + // does not forward RegisterCallback to the senders beneath it. + if (Parent.Tenants.Any() && TenancyBehavior == TenancyBehavior.TenantAware) + { + var defaultSender = new MqttTopicSender(this, Parent.Client); + var tenantedSender = new TenantedSender(Uri, Parent.TenantedIdBehavior, defaultSender); + + foreach (var tenant in Parent.Tenants) + { + tenantedSender.RegisterSender(tenant.TenantId, + new MqttTopicSender(this, Parent.GetTenantClient(tenant))); + } + + return tenantedSender; + } + return this; } diff --git a/src/Transports/MQTT/Wolverine.MQTT/MqttTransportExpression.cs b/src/Transports/MQTT/Wolverine.MQTT/MqttTransportExpression.cs index b8b559870..119414b0d 100644 --- a/src/Transports/MQTT/Wolverine.MQTT/MqttTransportExpression.cs +++ b/src/Transports/MQTT/Wolverine.MQTT/MqttTransportExpression.cs @@ -1,7 +1,9 @@ using JasperFx.Core.Reflection; +using MQTTnet.Extensions.ManagedClient; using Wolverine.Configuration; using Wolverine.MQTT.Internals; using Wolverine.Runtime; +using Wolverine.Transports.Sending; namespace Wolverine.MQTT; @@ -75,4 +77,73 @@ public MqttTransportExpression ConfigureSenders(Action(); } + + /// + /// Override the routing behavior for unknown or missing tenant ids when using broker-per-tenant MQTT + /// multi-tenancy (GH-3307). See . Default is + /// . + /// + /// + /// + public MqttTransportExpression TenantIdBehavior(TenantedIdBehavior behavior) + { + _transport.TenantedIdBehavior = behavior; + return this; + } + + /// + /// Register a tenant that is served by its own dedicated MQTT connection (its own broker) while sharing the + /// topic topology declared on this transport. Outbound messages carrying a matching + /// are routed to this tenant's connection; inbound messages consumed from it + /// are stamped with the tenant id. The tenant connection is always given a unique ClientId derived from the + /// tenant id, because MQTT brokers forcibly disconnect a second connection sharing a ClientId. + /// + /// + /// Configuration for the tenant's own broker connection + /// Optional OAUTH2-JWT authentication for the tenant's own connection. + /// + public MqttTransportExpression AddTenant(string tenantId, Action configure, + MqttJwtAuthenticationOptions? jwt = null) + { + if (string.IsNullOrEmpty(tenantId)) + { + throw new ArgumentOutOfRangeException(nameof(tenantId), "Empty or null tenantId"); + } + + ArgumentNullException.ThrowIfNull(configure); + + var builder = new ManagedMqttClientOptionsBuilder(); + configure(builder); + + var tenant = _transport.Tenants[tenantId]; + tenant.Options = builder.Build(); + tenant.Jwt = jwt; + + return this; + } + + /// + /// Register a tenant that is served by its own dedicated MQTT connection (its own broker), bypassing the + /// fluent builder. See . + /// + /// + /// The connection options for the tenant's own broker + /// Optional OAUTH2-JWT authentication for the tenant's own connection. + /// + public MqttTransportExpression AddTenant(string tenantId, ManagedMqttClientOptions mqttOptions, + MqttJwtAuthenticationOptions? jwt = null) + { + if (string.IsNullOrEmpty(tenantId)) + { + throw new ArgumentOutOfRangeException(nameof(tenantId), "Empty or null tenantId"); + } + + ArgumentNullException.ThrowIfNull(mqttOptions); + + var tenant = _transport.Tenants[tenantId]; + tenant.Options = mqttOptions; + tenant.Jwt = jwt; + + return this; + } } \ No newline at end of file diff --git a/src/Transports/MQTT/Wolverine.MQTT/MqttTransportExtensions.cs b/src/Transports/MQTT/Wolverine.MQTT/MqttTransportExtensions.cs index c7620e9af..0b0009adc 100644 --- a/src/Transports/MQTT/Wolverine.MQTT/MqttTransportExtensions.cs +++ b/src/Transports/MQTT/Wolverine.MQTT/MqttTransportExtensions.cs @@ -14,11 +14,11 @@ public static class MqttTransportExtensions /// /// /// - internal static MqttTransport MqttTransport(this WolverineOptions endpoints) + internal static MqttTransport MqttTransport(this WolverineOptions endpoints, BrokerName? name = null) { var transports = endpoints.As().Transports; - return transports.GetOrCreate(); + return transports.GetOrCreate(name); } /// @@ -77,6 +77,52 @@ public static MqttTransportExpression UseMqttWithLocalBroker(this WolverineOptio }); } + /// + /// Configure a connection to a secondary, named MQTT broker used by this application. Only use this overload if + /// your Wolverine application needs to talk to two or more MQTT brokers. The doubles as + /// the URI scheme for the additional broker's endpoints, so pin publishing/listening to it with + /// / . + /// + /// + /// Identity of the additional MQTT broker + /// Configuration for the additional broker's connection + /// Optional OAUTH2-JWT authentication for the additional broker. + public static MqttTransportExpression AddNamedMqttBroker(this WolverineOptions options, + BrokerName name, + Action configure, + MqttJwtAuthenticationOptions? jwtAuthenticationOptions = null) + { + var transport = options.MqttTransport(name); + var builder = new ManagedMqttClientOptionsBuilder(); + configure(builder); + + transport.Options = builder.Build(); + transport.JwtAuthenticationOptions = jwtAuthenticationOptions; + + return new MqttTransportExpression(transport, options); + } + + /// + /// Configure a connection to a secondary, named MQTT broker used by this application, bypassing the fluent + /// builder. Only use this overload if your Wolverine application needs to talk to two or more MQTT brokers. + /// + /// + /// Identity of the additional MQTT broker + /// The connection options for the additional broker + /// Optional OAUTH2-JWT authentication for the additional broker. + public static MqttTransportExpression AddNamedMqttBroker(this WolverineOptions options, + BrokerName name, + ManagedMqttClientOptions mqttOptions, + MqttJwtAuthenticationOptions? jwtAuthenticationOptions = null) + { + var transport = options.MqttTransport(name); + + transport.Options = mqttOptions; + transport.JwtAuthenticationOptions = jwtAuthenticationOptions; + + return new MqttTransportExpression(transport, options); + } + /// /// Listen for incoming messages at the designated MQTT topic name /// @@ -118,6 +164,25 @@ public static MqttListenerConfiguration ListenToMqttTopic(this WolverineOptions return new MqttListenerConfiguration(endpoint); } + /// + /// Listen for incoming messages at the designated MQTT topic name on an additional, named broker registered via + /// . + /// + /// + /// Identity of the additional MQTT broker + /// The MQTT topic name to listen to + public static MqttListenerConfiguration ListenToMqttTopicOnNamedBroker(this WolverineOptions endpoints, + BrokerName name, string topicName) + { + var transport = endpoints.MqttTransport(name); + + var endpoint = transport.Topics[topicName]; + endpoint.EndpointName = topicName; + endpoint.IsListener = true; + + return new MqttListenerConfiguration(endpoint); + } + /// /// Publish messages to an MQTT topic /// @@ -137,6 +202,27 @@ public static MqttSubscriberConfiguration ToMqttTopic(this IPublishToExpression return new MqttSubscriberConfiguration(topic); } + /// + /// Publish messages to an MQTT topic on an additional, named broker registered via + /// . + /// + /// + /// Identity of the additional MQTT broker + /// The MQTT topic name + public static MqttSubscriberConfiguration ToMqttTopicOnNamedBroker(this IPublishToExpression publishing, + BrokerName name, string topicName) + { + var transports = publishing.As().Parent.Transports; + var transport = transports.GetOrCreate(name); + + var topic = transport.Topics[topicName]; + + // This is necessary unfortunately to hook up the subscription rules + publishing.To(topic.Uri); + + return new MqttSubscriberConfiguration(topic); + } + /// /// Publish messages to MQTT topics based on Wolverine's rules for deriving topic /// names from a message type diff --git a/src/Wolverine/AssemblyAttributes.cs b/src/Wolverine/AssemblyAttributes.cs index e19b6819c..f11ffad77 100644 --- a/src/Wolverine/AssemblyAttributes.cs +++ b/src/Wolverine/AssemblyAttributes.cs @@ -49,6 +49,7 @@ [assembly: InternalsVisibleTo("Wolverine.Kafka.Tests")] [assembly: InternalsVisibleTo("Wolverine.Kafka")] [assembly: InternalsVisibleTo("Wolverine.Nats")] +[assembly: InternalsVisibleTo("Wolverine.MQTT")] [assembly: InternalsVisibleTo("Wolverine.Redis")] [assembly: InternalsVisibleTo("Wolverine.Pubsub")] [assembly: InternalsVisibleTo("MetricsTests")]