feat: wire MQTT end-to-end through NATS SubList for cross-protocol messaging

- MqttListener accepts IMessageRouter + delegates for client ID allocation and account resolution (Phase 1-2) - MqttConnection creates MqttNatsClientAdapter on CONNECT, registers with SubList for cross-protocol delivery (Phase 2) - PUBLISH routes through ProcessMessage() when router available, falls back to MQTT-only fan-out for test compatibility (Phase 3) - SUBSCRIBE creates real SubList entries via adapter, enabling NATS→MQTT delivery with topic↔subject translation (Phase 4) - PUBREL now delivers stored QoS 2 messages before ack (Phase 5) - ConnzHandler includes MQTT adapters in /connz output (Phase 6) - MQTTnet E2E tests: MQTT pub/sub, MQTT→NATS, NATS→MQTT, QoS 1 (Phase 7)
2026-03-13 11:38:52 -04:00
parent a1fc600d84
commit a5592ed533
5 changed files with 383 additions and 15 deletions
--- a/src/NATS.Server/Monitoring/ConnzHandler.cs
+++ b/src/NATS.Server/Monitoring/ConnzHandler.cs
@@ -29,6 +29,9 @@ public sealed class ConnzHandler(NatsServer server)
        {
            var clients = server.GetClients().ToArray();
            connInfos.AddRange(clients.Select(c => BuildConnInfo(c, now, opts)));
+
+            // Include MQTT adapter connections
+            connInfos.AddRange(server.GetMqttAdapters().Select(a => BuildMqttConnInfo(a, now)));
        }

        // Collect closed connections from the ring buffer
@@ -254,6 +257,21 @@ public sealed class ConnzHandler(NatsServer server)
        };
    }

+    private static ConnInfo BuildMqttConnInfo(Mqtt.MqttNatsClientAdapter adapter, DateTime now)
+    {
+        return new ConnInfo
+        {
+            Cid = adapter.Id,
+            Kind = "Client",
+            Type = "mqtt",
+            Start = now, // MQTT adapters don't track start time yet
+            LastActivity = now,
+            NumSubs = (uint)adapter.Subscriptions.Count,
+            Account = adapter.Account?.Name ?? "",
+            MqttClient = adapter.MqttClientId,
+        };
+    }
+
    private static ConnzOptions ParseQueryParams(HttpContext ctx)
    {
        var q = ctx.Request.Query;
--- a/src/NATS.Server/Mqtt/MqttConnection.cs
+++ b/src/NATS.Server/Mqtt/MqttConnection.cs
@@ -23,10 +23,18 @@ public sealed class MqttConnection : IAsyncDisposable
    private bool _connected;
    private bool _willCleared;
    private MqttConnectInfo _connectInfo;
+    private readonly Dictionary<string, string> _topicToSid = new(StringComparer.Ordinal);
+    private int _nextSid;

    /// <summary>Auth result after successful CONNECT (populated for AuthService path).</summary>
    public AuthResult? AuthResult { get; private set; }

+    /// <summary>
+    /// The adapter that bridges this MQTT connection to the NATS SubList routing.
+    /// Created on CONNECT when running with a NatsServer router; null in test-only mode.
+    /// </summary>
+    public MqttNatsClientAdapter? Adapter { get; private set; }
+
    public string ClientId => _clientId;

    /// <summary>
@@ -142,11 +150,7 @@ public sealed class MqttConnection : IAsyncDisposable
            // Publish will message if not cleanly disconnected
            if (_connected && !_willCleared && _connectInfo.WillTopic != null)
            {
-                await _listener.PublishAsync(
-                    _connectInfo.WillTopic,
-                    Encoding.UTF8.GetString(_connectInfo.WillMessage ?? []),
-                    this,
-                    CancellationToken.None);
+                RoutePublish(_connectInfo.WillTopic, _connectInfo.WillMessage ?? []);
            }
        }
    }
@@ -276,6 +280,15 @@ public sealed class MqttConnection : IAsyncDisposable

        AuthResult = authResult;

+        // Create MqttNatsClientAdapter for cross-protocol routing (when running with NatsServer)
+        if (_listener.AllocateClientId != null)
+        {
+            var adapterId = _listener.AllocateClientId();
+            Adapter = new MqttNatsClientAdapter(this, adapterId);
+            Adapter.Account = _listener.ResolveAccount?.Invoke(authResult.AccountName);
+            _listener.RegisterMqttAdapter(Adapter);
+        }
+
        // Duplicate client-id takeover
        _listener.TakeoverExistingConnection(_clientId, this);

@@ -313,20 +326,18 @@ public sealed class MqttConnection : IAsyncDisposable
        switch (publishInfo.QoS)
        {
            case 0:
-                await _listener.PublishAsync(publishInfo.Topic,
-                    Encoding.UTF8.GetString(publishInfo.Payload.Span), this, ct);
+                RoutePublish(publishInfo.Topic, publishInfo.Payload);
                break;

            case 1:
                _listener.RecordPendingPublish(_clientId, publishInfo.PacketId, publishInfo.Topic,
                    Encoding.UTF8.GetString(publishInfo.Payload.Span));
                await WriteBinaryAsync(MqttPacketWriter.WritePubAck(publishInfo.PacketId), ct);
-                await _listener.PublishAsync(publishInfo.Topic,
-                    Encoding.UTF8.GetString(publishInfo.Payload.Span), this, ct);
+                RoutePublish(publishInfo.Topic, publishInfo.Payload);
                break;

            case 2:
-                // QoS 2 step 1: store and send PUBREC
+                // QoS 2 step 1: store and send PUBREC (delivery deferred to PUBREL)
                _listener.RecordPendingPublish(_clientId, publishInfo.PacketId, publishInfo.Topic,
                    Encoding.UTF8.GetString(publishInfo.Payload.Span));
                await WriteBinaryAsync(MqttPacketWriter.WritePubRec(publishInfo.PacketId), ct);
@@ -343,6 +354,24 @@ public sealed class MqttConnection : IAsyncDisposable
        return true;
    }

+    /// <summary>
+    /// Routes a published message through the NATS SubList (cross-protocol) if a router
+    /// and adapter are available, otherwise falls back to MQTT-only fan-out.
+    /// </summary>
+    private void RoutePublish(string mqttTopic, ReadOnlyMemory<byte> payload)
+    {
+        if (_listener.Router != null && Adapter != null)
+        {
+            var natsSubject = MqttTopicMapper.MqttToNats(mqttTopic);
+            _listener.Router.ProcessMessage(natsSubject, null, default, payload, Adapter);
+        }
+        else
+        {
+            // Test-only fallback: MQTT-only fan-out
+            _ = _listener.PublishAsync(mqttTopic, Encoding.UTF8.GetString(payload.Span), this, CancellationToken.None);
+        }
+    }
+
    private void HandlePubAck(MqttControlPacket packet)
    {
        if (packet.Payload.Length < 2) return;
@@ -362,7 +391,13 @@ public sealed class MqttConnection : IAsyncDisposable
        if (packet.Payload.Length < 2) return;
        var packetId = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]);

-        // QoS 2 step 2: deliver the stored message and send PUBCOMP
+        // QoS 2 step 2: deliver the stored message, then ack and send PUBCOMP
+        var pending = _listener.GetPendingPublish(_clientId, packetId);
+        if (pending != null)
+        {
+            RoutePublish(pending.Topic, Encoding.UTF8.GetBytes(pending.Payload));
+        }
+
        _listener.AckPendingPublish(_clientId, packetId);
        await WriteBinaryAsync(MqttPacketWriter.WritePubComp(packetId), ct);
    }
@@ -383,8 +418,31 @@ public sealed class MqttConnection : IAsyncDisposable
        for (var i = 0; i < subscribeInfo.Filters.Count; i++)
        {
            var (topicFilter, requestedQoS) = subscribeInfo.Filters[i];
-            _listener.RegisterSubscription(this, topicFilter);
+
+            if (Adapter != null)
+            {
+                // Route through SubList for cross-protocol delivery
+                var natsSubject = MqttTopicMapper.MqttToNats(topicFilter);
+                var sid = $"$MQTT_{Interlocked.Increment(ref _nextSid)}";
+                Adapter.AddSubscription(natsSubject, sid);
+                _topicToSid[topicFilter] = sid;
+            }
+            else
+            {
+                // Test-only fallback: MQTT-only subscription
+                _listener.RegisterSubscription(this, topicFilter);
+            }
+
            grantedQoS[i] = Math.Min(requestedQoS, (byte)2);
+
+            // Deliver retained messages for this topic filter
+            var retained = _listener.GetRetainedMessage(topicFilter);
+            if (retained != null)
+            {
+                var retainedPayload = Encoding.UTF8.GetBytes(retained);
+                await WriteBinaryAsync(
+                    MqttPacketWriter.WritePublish(topicFilter, retainedPayload, qos: 0, retain: true, packetId: 0), ct);
+            }
        }

        await WriteBinaryAsync(MqttPacketWriter.WriteSubAck(subscribeInfo.PacketId, grantedQoS), ct);
@@ -395,7 +453,16 @@ public sealed class MqttConnection : IAsyncDisposable
        var unsubInfo = MqttBinaryDecoder.ParseUnsubscribe(packet.Payload.Span, packet.Flags);

        foreach (var filter in unsubInfo.Filters)
-            _listener.UnregisterSubscription(this, filter);
+        {
+            if (Adapter != null && _topicToSid.Remove(filter, out var sid))
+            {
+                Adapter.RemoveSubscription(sid);
+            }
+            else
+            {
+                _listener.UnregisterSubscription(this, filter);
+            }
+        }

        await WriteBinaryAsync(MqttPacketWriter.WriteUnsubAck(unsubInfo.PacketId), ct);
    }
@@ -427,6 +494,13 @@ public sealed class MqttConnection : IAsyncDisposable

    public async ValueTask DisposeAsync()
    {
+        // Clean up adapter subscriptions and unregister from listener
+        if (Adapter != null)
+        {
+            Adapter.RemoveAllSubscriptions();
+            _listener.UnregisterMqttAdapter(Adapter);
+        }
+
        _listener.Unregister(this);
        _writeGate.Dispose();
        await _stream.DisposeAsync();
--- a/src/NATS.Server/Mqtt/MqttListener.cs
+++ b/src/NATS.Server/Mqtt/MqttListener.cs
@@ -24,6 +24,8 @@ public sealed class MqttListener : IAsyncDisposable
    private readonly ConcurrentDictionary<string, MqttSessionState> _sessions = new(StringComparer.Ordinal);
    private readonly ConcurrentDictionary<string, MqttConnection> _clientIdMap = new(StringComparer.Ordinal);
    private readonly ConcurrentDictionary<string, string> _retainedMessages = new(StringComparer.Ordinal);
+    private readonly IMessageRouter? _router;
+    private readonly ConcurrentDictionary<ulong, MqttNatsClientAdapter> _mqttAdapters = new();
    private MqttStreamInitializer? _streamInitializer;
    private MqttConsumerManager? _mqttConsumerManager;
    private TcpListener? _listener;
@@ -63,7 +65,8 @@ public sealed class MqttListener : IAsyncDisposable
        AuthService? authService,
        MqttOptions mqttOptions,
        MqttStreamInitializer? streamInitializer = null,
-        MqttConsumerManager? mqttConsumerManager = null)
+        MqttConsumerManager? mqttConsumerManager = null,
+        IMessageRouter? router = null)
    {
        _host = host;
        _port = port;
@@ -73,6 +76,7 @@ public sealed class MqttListener : IAsyncDisposable
        _requiredPassword = mqttOptions.Password;
        _streamInitializer = streamInitializer;
        _mqttConsumerManager = mqttConsumerManager;
+        _router = router;

        // Build TLS options if configured
        if (mqttOptions.HasTls)
@@ -91,6 +95,49 @@ public sealed class MqttListener : IAsyncDisposable
    /// </summary>
    internal MqttConsumerManager? ConsumerManager => _mqttConsumerManager;

+    /// <summary>
+    /// The message router for cross-protocol delivery (MQTT→NATS SubList routing).
+    /// Null when running in test-only mode without NatsServer.
+    /// </summary>
+    internal IMessageRouter? Router => _router;
+
+    /// <summary>
+    /// Delegate to allocate a server-unique client ID for MQTT adapters.
+    /// Set by NatsServer after construction.
+    /// </summary>
+    internal Func<ulong>? AllocateClientId { get; set; }
+
+    /// <summary>
+    /// Delegate to resolve an Account by name for MQTT adapters.
+    /// Set by NatsServer after construction.
+    /// </summary>
+    internal Func<string?, Auth.Account?>? ResolveAccount { get; set; }
+
+    internal void RegisterMqttAdapter(MqttNatsClientAdapter adapter)
+        => _mqttAdapters[adapter.Id] = adapter;
+
+    internal void UnregisterMqttAdapter(MqttNatsClientAdapter adapter)
+        => _mqttAdapters.TryRemove(adapter.Id, out _);
+
+    internal IEnumerable<MqttNatsClientAdapter> GetMqttAdapters()
+        => _mqttAdapters.Values;
+
+    /// <summary>
+    /// Looks up a specific pending publish by client ID and packet ID.
+    /// Used by QoS 2 PUBREL to retrieve the stored message for delivery.
+    /// </summary>
+    internal MqttPendingPublish? GetPendingPublish(string clientId, int packetId)
+    {
+        if (string.IsNullOrWhiteSpace(clientId) || packetId <= 0)
+            return null;
+
+        if (_sessions.TryGetValue(clientId, out var session)
+            && session.Pending.TryGetValue(packetId, out var pending))
+            return pending;
+
+        return null;
+    }
+
    public Task StartAsync(CancellationToken ct)
    {
        var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, _cts.Token);
@@ -280,6 +327,7 @@ public sealed class MqttListener : IAsyncDisposable
        _sessions.Clear();
        _clientIdMap.Clear();
        _retainedMessages.Clear();
+        _mqttAdapters.Clear();
        _cts.Dispose();
    }

--- a/src/NATS.Server/NatsServer.cs
+++ b/src/NATS.Server/NatsServer.cs
@@ -122,6 +122,12 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
    /// </summary>
    public int? MqttListenerPort => _mqttListener?.Port;

+    /// <summary>
+    /// Returns all active MQTT client adapters for monitoring (/connz).
+    /// </summary>
+    public IEnumerable<Mqtt.MqttNatsClientAdapter> GetMqttAdapters()
+        => _mqttListener?.GetMqttAdapters() ?? [];
+
    public Account SystemAccount => _systemAccount;
    public string ServerNKey { get; }
    public InternalEventSystem? EventSystem => _eventSystem;
@@ -937,7 +943,10 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
                _authService,
                mqttOptions,
                mqttStreamInit,
-                mqttConsumerMgr);
+                mqttConsumerMgr,
+                router: this);
+            _mqttListener.AllocateClientId = () => Interlocked.Increment(ref _nextClientId);
+            _mqttListener.ResolveAccount = name => GetOrCreateAccount(name ?? Auth.Account.GlobalAccountName);
            await _mqttListener.StartAsync(linked.Token);
        }
        if (_jetStreamService != null)