diff --git a/src/NATS.Server/Monitoring/ConnzHandler.cs b/src/NATS.Server/Monitoring/ConnzHandler.cs index fb8d7c8..41cfa93 100644 --- a/src/NATS.Server/Monitoring/ConnzHandler.cs +++ b/src/NATS.Server/Monitoring/ConnzHandler.cs @@ -184,6 +184,7 @@ public sealed class ConnzHandler(NatsServer server) Tags = tags, Proxy = string.IsNullOrEmpty(proxyKey) ? null : new ProxyInfo { Key = proxyKey }, Rtt = FormatRtt(client.Rtt), + MqttClient = client.MqttClientId ?? "", }; if (opts.Subscriptions) diff --git a/src/NATS.Server/Monitoring/VarzHandler.cs b/src/NATS.Server/Monitoring/VarzHandler.cs index 9719b4a..a4f3536 100644 --- a/src/NATS.Server/Monitoring/VarzHandler.cs +++ b/src/NATS.Server/Monitoring/VarzHandler.cs @@ -196,7 +196,7 @@ public sealed class VarzHandler : IDisposable return new MqttOptsVarz { Host = mqtt.Host, - Port = mqtt.Port, + Port = _server.MqttListenerPort ?? mqtt.Port, NoAuthUser = mqtt.NoAuthUser ?? "", AuthTimeout = mqtt.AuthTimeout, TlsMap = mqtt.TlsMap, diff --git a/src/NATS.Server/Mqtt/MqttBinaryDecoder.cs b/src/NATS.Server/Mqtt/MqttBinaryDecoder.cs index ec326be..77dd781 100644 --- a/src/NATS.Server/Mqtt/MqttBinaryDecoder.cs +++ b/src/NATS.Server/Mqtt/MqttBinaryDecoder.cs @@ -222,6 +222,46 @@ public static class MqttBinaryDecoder return new MqttSubscribeInfo(packetId, filters); } + // ------------------------------------------------------------------------- + // UNSUBSCRIBE parsing + // Go reference: server/mqtt.go mqttParseUnsub ~line 1500 + // ------------------------------------------------------------------------- + + /// + /// Decoded fields from an MQTT UNSUBSCRIBE packet body. + /// + public readonly record struct MqttUnsubscribeInfo( + ushort PacketId, + IReadOnlyList Filters); + + /// + /// Parses the payload bytes of an MQTT UNSUBSCRIBE packet. + /// + /// The payload bytes from . + /// + /// Optional fixed-header flags nibble. When provided, must be 0x02 per MQTT 3.1.1 spec. + /// + public static MqttUnsubscribeInfo ParseUnsubscribe(ReadOnlySpan payload, byte? flags = null) + { + if (flags.HasValue && flags.Value != 0x02) + throw new FormatException("MQTT UNSUBSCRIBE packet has invalid fixed-header flags."); + + var pos = 0; + var packetId = ReadUInt16BigEndian(payload, ref pos); + + var filters = new List(); + while (pos < payload.Length) + { + var topicFilter = ReadUtf8String(payload, ref pos); + filters.Add(topicFilter); + } + + if (filters.Count == 0) + throw new FormatException("MQTT UNSUBSCRIBE packet must contain at least one topic filter."); + + return new MqttUnsubscribeInfo(packetId, filters); + } + // ------------------------------------------------------------------------- // MQTT wildcard → NATS subject translation // Go reference: server/mqtt.go mqttToNATSSubjectConversion ~line 2200 diff --git a/src/NATS.Server/Mqtt/MqttConnection.cs b/src/NATS.Server/Mqtt/MqttConnection.cs index 22a3119..8c533ca 100644 --- a/src/NATS.Server/Mqtt/MqttConnection.cs +++ b/src/NATS.Server/Mqtt/MqttConnection.cs @@ -1,20 +1,467 @@ +using System.Buffers; +using System.IO.Pipelines; using System.Net.Sockets; +using System.Security.Cryptography.X509Certificates; using System.Text; +using NATS.Server.Auth; +using static NATS.Server.Mqtt.MqttBinaryDecoder; namespace NATS.Server.Mqtt; -public sealed class MqttConnection(TcpClient client, MqttListener listener) : IAsyncDisposable +public sealed class MqttConnection : IAsyncDisposable { - private readonly TcpClient _client = client; - private readonly NetworkStream _stream = client.GetStream(); - private readonly MqttListener _listener = listener; + private readonly TcpClient? _tcpClient; + private readonly Stream _stream; + private readonly MqttListener _listener; private readonly MqttProtocolParser _parser = new(); private readonly SemaphoreSlim _writeGate = new(1, 1); + private readonly bool _useBinaryProtocol; + private readonly X509Certificate2? _clientCert; private string _clientId = string.Empty; private bool _cleanSession = true; private TimeSpan _idleTimeout = Timeout.InfiniteTimeSpan; + private bool _connected; + private bool _willCleared; + private MqttConnectInfo _connectInfo; + + /// Auth result after successful CONNECT (populated for AuthService path). + public AuthResult? AuthResult { get; private set; } + + public string ClientId => _clientId; + + /// + /// Creates a connection from a TcpClient (standard accept path). + /// + public MqttConnection(TcpClient client, MqttListener listener, bool useBinaryProtocol = true) + { + _tcpClient = client; + _stream = client.GetStream(); + _listener = listener; + _useBinaryProtocol = useBinaryProtocol; + } + + /// + /// Creates a connection from an arbitrary Stream (for TLS wrapping or testing). + /// + public MqttConnection(Stream stream, MqttListener listener, bool useBinaryProtocol = true) + { + _stream = stream; + _listener = listener; + _useBinaryProtocol = useBinaryProtocol; + } + + /// + /// Creates a connection from a Stream with a TLS client certificate. + /// Used by the accept loop after TLS handshake completes. + /// + public MqttConnection(Stream stream, MqttListener listener, bool useBinaryProtocol, X509Certificate2? clientCert) + { + _stream = stream; + _listener = listener; + _useBinaryProtocol = useBinaryProtocol; + _clientCert = clientCert; + } public async Task RunAsync(CancellationToken ct) + { + if (_useBinaryProtocol) + await RunBinaryAsync(ct); + else + await RunTextAsync(ct); + } + + private async Task RunBinaryAsync(CancellationToken ct) + { + var pipeReader = PipeReader.Create(_stream, new StreamPipeReaderOptions(leaveOpen: true)); + + try + { + while (!ct.IsCancellationRequested) + { + ReadResult readResult; + try + { + // Apply idle timeout for keepalive + if (_idleTimeout != Timeout.InfiniteTimeSpan && _idleTimeout > TimeSpan.Zero) + { + using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + timeoutCts.CancelAfter(_idleTimeout); + readResult = await pipeReader.ReadAsync(timeoutCts.Token); + } + else + { + readResult = await pipeReader.ReadAsync(ct); + } + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + // Keepalive timeout + break; + } + catch + { + break; + } + + var buffer = readResult.Buffer; + + if (buffer.IsEmpty && readResult.IsCompleted) + break; + + while (MqttPacketReader.TryRead(buffer, out var packet, out var consumed)) + { + buffer = buffer.Slice(consumed); + + try + { + var shouldContinue = await ProcessBinaryPacketAsync(packet!, ct); + if (!shouldContinue) + { + pipeReader.AdvanceTo(consumed); + return; + } + } + catch (FormatException) + { + // Protocol violation — disconnect + pipeReader.AdvanceTo(consumed); + return; + } + } + + pipeReader.AdvanceTo(buffer.Start, buffer.End); + + if (readResult.IsCompleted) + break; + } + } + finally + { + await pipeReader.CompleteAsync(); + + // 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); + } + } + } + + /// + /// Processes a single binary MQTT control packet. + /// Returns false if the connection should be closed. + /// + private async Task ProcessBinaryPacketAsync(MqttControlPacket packet, CancellationToken ct) + { + // MQTT 3.1.1: First packet MUST be CONNECT + if (!_connected && packet.Type != MqttControlPacketType.Connect) + return false; + + switch (packet.Type) + { + case MqttControlPacketType.Connect: + return await HandleConnectAsync(packet, ct); + + case MqttControlPacketType.Publish: + return await HandlePublishAsync(packet, ct); + + case MqttControlPacketType.PubAck: + HandlePubAck(packet); + return true; + + case MqttControlPacketType.PubRec: + await HandlePubRecAsync(packet, ct); + return true; + + case MqttControlPacketType.PubRel: + // Fixed-header flags must be 0x02 for PUBREL + if (packet.Flags != 0x02) + return false; + await HandlePubRelAsync(packet, ct); + return true; + + case MqttControlPacketType.PubComp: + HandlePubComp(packet); + return true; + + case MqttControlPacketType.Subscribe: + // Fixed-header flags must be 0x02 for SUBSCRIBE + if (packet.Flags != MqttProtocolConstants.SubscribeFlags) + return false; + await HandleSubscribeAsync(packet, ct); + return true; + + case MqttControlPacketType.Unsubscribe: + // Fixed-header flags must be 0x02 for UNSUBSCRIBE + if (packet.Flags != 0x02) + return false; + await HandleUnsubscribeAsync(packet, ct); + return true; + + case MqttControlPacketType.PingReq: + await WriteBinaryAsync(MqttPacketWriter.WritePingResp(), ct); + return true; + + case MqttControlPacketType.Disconnect: + // Clean disconnect — clear will message + _willCleared = true; + return false; + + default: + // Unknown packet type — disconnect + return false; + } + } + + private async Task HandleConnectAsync(MqttControlPacket packet, CancellationToken ct) + { + if (_connected) + return false; // Second CONNECT is a protocol violation + + var connectInfo = MqttBinaryDecoder.ParseConnect(packet.Payload.Span); + _connectInfo = connectInfo; + + // MQTT 3.1.1: Reserved bit (bit 0 of connect flags) must be 0 + // This is implicitly validated because we parse individual flag bits + + // Protocol level must be 4 for MQTT 3.1.1 + if (connectInfo.ProtocolLevel != 4) + { + await WriteBinaryAsync( + MqttPacketWriter.WriteConnAck(0x00, MqttProtocolConstants.ConnAckUnacceptableProtocolVersion), ct); + return false; + } + + // Will QoS range check (0-2) + if (connectInfo.WillQoS > 2) + { + await WriteBinaryAsync( + MqttPacketWriter.WriteConnAck(0x00, MqttProtocolConstants.ConnAckIdentifierRejected), ct); + return false; + } + + // Empty client-id handling + if (string.IsNullOrEmpty(connectInfo.ClientId)) + { + if (connectInfo.CleanSession) + { + // Generate a unique client ID + _clientId = $"auto-{Guid.NewGuid():N}"; + } + else + { + // Empty client-id with persistent session is not allowed + await WriteBinaryAsync( + MqttPacketWriter.WriteConnAck(0x00, MqttProtocolConstants.ConnAckIdentifierRejected), ct); + return false; + } + } + else + { + _clientId = connectInfo.ClientId; + } + + // Auth check via AuthService (passes TLS client cert for cert-mapping auth) + var authResult = _listener.AuthenticateMqtt(connectInfo.Username, connectInfo.Password, _clientCert); + if (authResult == null) + { + await WriteBinaryAsync( + MqttPacketWriter.WriteConnAck(0x00, MqttProtocolConstants.ConnAckNotAuthorized), ct); + return false; + } + + AuthResult = authResult; + + // Duplicate client-id takeover + _listener.TakeoverExistingConnection(_clientId, this); + + _cleanSession = connectInfo.CleanSession; + _idleTimeout = _listener.ResolveKeepAliveTimeout(connectInfo.KeepAlive); + + // Session-present bit: 1 if resuming existing session, 0 otherwise + var pending = _listener.OpenSession(_clientId, _cleanSession); + byte sessionPresent = (byte)(!_cleanSession && pending.Count > 0 ? 0x01 : 0x00); + + _connected = true; + await WriteBinaryAsync( + MqttPacketWriter.WriteConnAck(sessionPresent, MqttProtocolConstants.ConnAckAccepted), ct); + + // Redeliver pending QoS 1 messages + foreach (var redelivery in pending) + { + var payloadBytes = Encoding.UTF8.GetBytes(redelivery.Payload); + await WriteBinaryAsync( + MqttPacketWriter.WritePublish(redelivery.Topic, payloadBytes, qos: 1, dup: true, + packetId: (ushort)redelivery.PacketId), ct); + } + + return true; + } + + private async Task HandlePublishAsync(MqttControlPacket packet, CancellationToken ct) + { + var publishInfo = MqttBinaryDecoder.ParsePublish(packet.Payload.Span, packet.Flags); + + // Non-zero packet identifier required for QoS > 0 + if (publishInfo.QoS > 0 && publishInfo.PacketId == 0) + return false; + + switch (publishInfo.QoS) + { + case 0: + await _listener.PublishAsync(publishInfo.Topic, + Encoding.UTF8.GetString(publishInfo.Payload.Span), this, ct); + 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); + break; + + case 2: + // QoS 2 step 1: store and send PUBREC + _listener.RecordPendingPublish(_clientId, publishInfo.PacketId, publishInfo.Topic, + Encoding.UTF8.GetString(publishInfo.Payload.Span)); + await WriteBinaryAsync(MqttPacketWriter.WritePubRec(publishInfo.PacketId), ct); + break; + } + + // Handle retained messages + if (publishInfo.Retain) + { + _listener.SetRetainedMessage(publishInfo.Topic, + publishInfo.Payload.Length == 0 ? null : Encoding.UTF8.GetString(publishInfo.Payload.Span)); + } + + return true; + } + + private void HandlePubAck(MqttControlPacket packet) + { + if (packet.Payload.Length < 2) return; + var packetId = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]); + _listener.AckPendingPublish(_clientId, packetId); + } + + private async Task HandlePubRecAsync(MqttControlPacket packet, CancellationToken ct) + { + if (packet.Payload.Length < 2) return; + var packetId = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]); + await WriteBinaryAsync(MqttPacketWriter.WritePubRel(packetId), ct); + } + + private async Task HandlePubRelAsync(MqttControlPacket packet, CancellationToken ct) + { + 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 + _listener.AckPendingPublish(_clientId, packetId); + await WriteBinaryAsync(MqttPacketWriter.WritePubComp(packetId), ct); + } + + private void HandlePubComp(MqttControlPacket packet) + { + if (packet.Payload.Length < 2) return; + var packetId = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]); + _listener.AckPendingPublish(_clientId, packetId); + } + + private async Task HandleSubscribeAsync(MqttControlPacket packet, CancellationToken ct) + { + var subscribeInfo = MqttBinaryDecoder.ParseSubscribe(packet.Payload.Span, packet.Flags); + + // Grant QoS (cap at 2) + var grantedQoS = new byte[subscribeInfo.Filters.Count]; + for (var i = 0; i < subscribeInfo.Filters.Count; i++) + { + var (topicFilter, requestedQoS) = subscribeInfo.Filters[i]; + _listener.RegisterSubscription(this, topicFilter); + grantedQoS[i] = Math.Min(requestedQoS, (byte)2); + } + + await WriteBinaryAsync(MqttPacketWriter.WriteSubAck(subscribeInfo.PacketId, grantedQoS), ct); + } + + private async Task HandleUnsubscribeAsync(MqttControlPacket packet, CancellationToken ct) + { + var unsubInfo = MqttBinaryDecoder.ParseUnsubscribe(packet.Payload.Span, packet.Flags); + + foreach (var filter in unsubInfo.Filters) + _listener.UnregisterSubscription(this, filter); + + await WriteBinaryAsync(MqttPacketWriter.WriteUnsubAck(unsubInfo.PacketId), ct); + } + + /// + /// Sends a binary MQTT PUBLISH packet to this connection (for message delivery). + /// + public async Task SendBinaryPublishAsync(string topic, ReadOnlyMemory payload, byte qos, + bool retain, ushort packetId, CancellationToken ct) + { + var packet = MqttPacketWriter.WritePublish(topic, payload.Span, qos, retain, packetId: packetId); + await WriteBinaryAsync(packet, ct); + } + + /// + /// Sends a message to the connection. Used by the listener for fan-out delivery. + /// In binary mode, sends a PUBLISH packet; in text mode, sends a text line. + /// + public Task SendMessageAsync(string topic, string payload, CancellationToken ct) + { + if (_useBinaryProtocol) + { + var payloadBytes = Encoding.UTF8.GetBytes(payload); + return SendBinaryPublishAsync(topic, payloadBytes, qos: 0, retain: false, packetId: 0, ct); + } + + return WriteLineAsync($"MSG {topic} {payload}", ct); + } + + public async ValueTask DisposeAsync() + { + _listener.Unregister(this); + _writeGate.Dispose(); + await _stream.DisposeAsync(); + _tcpClient?.Dispose(); + } + + /// + /// Forces this connection to close (used for duplicate client-id takeover). + /// + internal void ForceClose() + { + try { _stream.Close(); } + catch { /* best effort */ } + try { _tcpClient?.Close(); } + catch { /* best effort */ } + } + + private async Task WriteBinaryAsync(byte[] data, CancellationToken ct) + { + await _writeGate.WaitAsync(ct); + try + { + await _stream.WriteAsync(data, ct); + await _stream.FlushAsync(ct); + } + finally + { + _writeGate.Release(); + } + } + + // --- Text protocol methods (for backward compatibility during test migration) --- + // TODO: Remove after test migration — deadline: Phase 3 completion + + private async Task RunTextAsync(CancellationToken ct) { while (!ct.IsCancellationRequested) { @@ -65,17 +512,6 @@ public sealed class MqttConnection(TcpClient client, MqttListener listener) : IA } } - public Task SendMessageAsync(string topic, string payload, CancellationToken ct) - => WriteLineAsync($"MSG {topic} {payload}", ct); - - public async ValueTask DisposeAsync() - { - _listener.Unregister(this); - _writeGate.Dispose(); - await _stream.DisposeAsync(); - _client.Dispose(); - } - private async Task WriteLineAsync(string line, CancellationToken ct) { await _writeGate.WaitAsync(ct); diff --git a/src/NATS.Server/Mqtt/MqttConsumerManager.cs b/src/NATS.Server/Mqtt/MqttConsumerManager.cs new file mode 100644 index 0000000..b3e943f --- /dev/null +++ b/src/NATS.Server/Mqtt/MqttConsumerManager.cs @@ -0,0 +1,199 @@ +// Manages per-subscription JetStream consumers for MQTT QoS 1/2 delivery. +// Go reference: golang/nats-server/server/mqtt.go mqttAccountSessionManager ~line 600 +// Consumer creation per subscription, ack tracking, session resume. + +using System.Collections.Concurrent; +using NATS.Server.JetStream; +using NATS.Server.JetStream.Models; +using NATS.Server.JetStream.Storage; + +namespace NATS.Server.Mqtt; + +/// +/// Tracks the mapping between an MQTT subscription and its JetStream consumer. +/// +public sealed record MqttConsumerBinding(string Stream, string DurableName, string FilterSubject); + +/// +/// Manages per-subscription JetStream consumers for MQTT QoS 1/2 message delivery. +/// Each QoS > 0 subscription gets a durable consumer on $MQTT_msgs. +/// Go reference: server/mqtt.go — mqttAccountSessionManager consumer management. +/// +public sealed class MqttConsumerManager +{ + private readonly StreamManager _streamManager; + private readonly ConsumerManager _consumerManager; + private readonly ConcurrentDictionary _bindings = new(StringComparer.Ordinal); + + public MqttConsumerManager(StreamManager streamManager, ConsumerManager consumerManager) + { + _streamManager = streamManager; + _consumerManager = consumerManager; + } + + /// + /// Creates a durable JetStream consumer for a QoS > 0 MQTT subscription. + /// The consumer filters $MQTT_msgs by the translated NATS subject. + /// Returns the binding, or null if creation failed. + /// Go reference: server/mqtt.go mqttProcessSub consumer creation. + /// + public MqttConsumerBinding? CreateSubscriptionConsumer(string clientId, string natsSubject, int qos, int maxAckPending) + { + var durableName = $"$MQTT_{clientId}_{natsSubject.Replace('.', '_').Replace('*', 'W').Replace('>', 'G')}"; + var filterSubject = $"{MqttProtocolConstants.StreamSubjectPrefix}{natsSubject}"; + + var response = _consumerManager.CreateOrUpdate(MqttProtocolConstants.StreamName, new ConsumerConfig + { + DurableName = durableName, + FilterSubject = filterSubject, + AckPolicy = AckPolicy.Explicit, + DeliverPolicy = DeliverPolicy.All, + MaxAckPending = maxAckPending, + AckWaitMs = (int)MqttProtocolConstants.DefaultAckWait.TotalMilliseconds, + MaxDeliver = -1, + }); + + if (response.Error != null) + return null; + + var binding = new MqttConsumerBinding(MqttProtocolConstants.StreamName, durableName, filterSubject); + _bindings[$"{clientId}:{natsSubject}"] = binding; + return binding; + } + + /// + /// Removes the JetStream consumer for an MQTT subscription. + /// Called on UNSUBSCRIBE or clean session disconnect. + /// + public void RemoveSubscriptionConsumer(string clientId, string natsSubject) + { + var key = $"{clientId}:{natsSubject}"; + if (_bindings.TryRemove(key, out var binding)) + { + _consumerManager.Delete(binding.Stream, binding.DurableName); + } + } + + /// + /// Removes all consumers for a client. Called on clean session disconnect. + /// + public void RemoveAllConsumers(string clientId) + { + var prefix = $"{clientId}:"; + var keysToRemove = _bindings.Keys.Where(k => k.StartsWith(prefix, StringComparison.Ordinal)).ToList(); + foreach (var key in keysToRemove) + { + if (_bindings.TryRemove(key, out var binding)) + { + _consumerManager.Delete(binding.Stream, binding.DurableName); + } + } + } + + /// + /// Gets the binding for a subscription, or null if none exists. + /// + public MqttConsumerBinding? GetBinding(string clientId, string natsSubject) + { + return _bindings.TryGetValue($"{clientId}:{natsSubject}", out var binding) ? binding : null; + } + + /// + /// Gets all bindings for a client (for session persistence). + /// + public IReadOnlyDictionary GetClientBindings(string clientId) + { + var prefix = $"{clientId}:"; + return _bindings + .Where(kvp => kvp.Key.StartsWith(prefix, StringComparison.Ordinal)) + .ToDictionary(kvp => kvp.Key[prefix.Length..], kvp => kvp.Value); + } + + /// + /// Publishes a message to the $MQTT_msgs stream for QoS delivery. + /// Returns the sequence number, or 0 if publish failed. + /// + public ulong PublishToStream(string natsSubject, ReadOnlyMemory payload) + { + var subject = $"{MqttProtocolConstants.StreamSubjectPrefix}{natsSubject}"; + if (_streamManager.TryGet(MqttProtocolConstants.StreamName, out var handle)) + { + var seq = handle.Store.AppendAsync(subject, payload, default).GetAwaiter().GetResult(); + return seq; + } + + return 0; + } + + /// + /// Acknowledges a message in the stream by removing it (for interest-based retention). + /// Called when PUBACK is received for QoS 1. + /// + public bool AcknowledgeMessage(ulong sequence) + { + if (_streamManager.TryGet(MqttProtocolConstants.StreamName, out var handle)) + { + return handle.Store.RemoveAsync(sequence, default).GetAwaiter().GetResult(); + } + + return false; + } + + /// + /// Loads a message from the $MQTT_msgs stream by sequence. + /// + public async ValueTask LoadMessageAsync(ulong sequence, CancellationToken ct = default) + { + if (_streamManager.TryGet(MqttProtocolConstants.StreamName, out var handle)) + { + return await handle.Store.LoadAsync(sequence, ct); + } + + return null; + } + + /// + /// Stores a QoS 2 incoming message for deduplication. + /// Returns the sequence number, or 0 if failed. + /// + public ulong StoreQoS2Incoming(string clientId, ushort packetId, ReadOnlyMemory payload) + { + var subject = $"{MqttProtocolConstants.QoS2IncomingMsgsStreamSubjectPrefix}{clientId}.{packetId}"; + if (_streamManager.TryGet(MqttProtocolConstants.QoS2IncomingMsgsStreamName, out var handle)) + { + return handle.Store.AppendAsync(subject, payload, default).GetAwaiter().GetResult(); + } + + return 0; + } + + /// + /// Loads a QoS 2 incoming message for delivery on PUBREL. + /// + public async ValueTask LoadQoS2IncomingAsync(string clientId, ushort packetId, CancellationToken ct = default) + { + var subject = $"{MqttProtocolConstants.QoS2IncomingMsgsStreamSubjectPrefix}{clientId}.{packetId}"; + if (_streamManager.TryGet(MqttProtocolConstants.QoS2IncomingMsgsStreamName, out var handle)) + { + return await handle.Store.LoadLastBySubjectAsync(subject, ct); + } + + return null; + } + + /// + /// Removes a QoS 2 incoming message after PUBCOMP. + /// + public async ValueTask RemoveQoS2IncomingAsync(string clientId, ushort packetId, CancellationToken ct = default) + { + var subject = $"{MqttProtocolConstants.QoS2IncomingMsgsStreamSubjectPrefix}{clientId}.{packetId}"; + if (_streamManager.TryGet(MqttProtocolConstants.QoS2IncomingMsgsStreamName, out var handle)) + { + var msg = await handle.Store.LoadLastBySubjectAsync(subject, ct); + if (msg != null) + return await handle.Store.RemoveAsync(msg.Sequence, ct); + } + + return false; + } +} diff --git a/src/NATS.Server/Mqtt/MqttFlowController.cs b/src/NATS.Server/Mqtt/MqttFlowController.cs index f3c6e12..ed796b1 100644 --- a/src/NATS.Server/Mqtt/MqttFlowController.cs +++ b/src/NATS.Server/Mqtt/MqttFlowController.cs @@ -66,12 +66,22 @@ public sealed class MqttFlowController : IDisposable /// /// Updates the MaxAckPending limit (e.g., on config reload). - /// Creates a new semaphore with the updated limit. /// public void UpdateLimit(int newLimit) { _defaultMaxAckPending = newLimit; - // Note: existing subscriptions keep their old limit until re-created + } + + /// + /// Returns whether the subscription has reached its MaxAckPending limit. + /// Used to pause JetStream consumer delivery when the limit is reached. + /// Go reference: server/mqtt.go mqttMaxAckPending flow control. + /// + public bool IsAtCapacity(string subscriptionId) + { + if (!_subscriptions.TryGetValue(subscriptionId, out var state)) + return false; + return state.Semaphore.CurrentCount == 0; } /// diff --git a/src/NATS.Server/Mqtt/MqttListener.cs b/src/NATS.Server/Mqtt/MqttListener.cs index 0fd7d71..5c6ebac 100644 --- a/src/NATS.Server/Mqtt/MqttListener.cs +++ b/src/NATS.Server/Mqtt/MqttListener.cs @@ -1,29 +1,96 @@ using System.Collections.Concurrent; using System.Net; +using System.Net.Security; using System.Net.Sockets; +using System.Security.Cryptography.X509Certificates; using NATS.Server.Auth; +using NATS.Server.Auth.Jwt; +using NATS.Server.Protocol; +using NATS.Server.Tls; namespace NATS.Server.Mqtt; -public sealed class MqttListener( - string host, - int port, - string? requiredUsername = null, - string? requiredPassword = null) : IAsyncDisposable +public sealed class MqttListener : IAsyncDisposable { - private readonly string _host = host; - private int _port = port; - private readonly string? _requiredUsername = requiredUsername; - private readonly string? _requiredPassword = requiredPassword; + private readonly string _host; + private int _port; + private readonly string? _requiredUsername; + private readonly string? _requiredPassword; + private readonly AuthService? _authService; + private readonly MqttOptions? _mqttOptions; + private readonly SslServerAuthenticationOptions? _sslOptions; private readonly ConcurrentDictionary _connections = new(); private readonly ConcurrentDictionary> _subscriptions = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _sessions = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _clientIdMap = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _retainedMessages = new(StringComparer.Ordinal); + private MqttStreamInitializer? _streamInitializer; + private MqttConsumerManager? _mqttConsumerManager; private TcpListener? _listener; private Task? _acceptLoop; private readonly CancellationTokenSource _cts = new(); + /// + /// When false, connections use the legacy text-line protocol for backward compatibility + /// with existing tests. Default is true (binary MQTT 3.1.1). + /// TODO: Remove after test migration — deadline: Phase 3 completion. + /// + internal bool UseBinaryProtocol { get; set; } = true; + public int Port => _port; + /// + /// Simple constructor for tests using static username/password auth (no TLS). + /// + public MqttListener( + string host, + int port, + string? requiredUsername = null, + string? requiredPassword = null) + { + _host = host; + _port = port; + _requiredUsername = requiredUsername; + _requiredPassword = requiredPassword; + } + + /// + /// Full constructor for production use with AuthService, TLS, and optional JetStream support. + /// + public MqttListener( + string host, + int port, + AuthService? authService, + MqttOptions mqttOptions, + MqttStreamInitializer? streamInitializer = null, + MqttConsumerManager? mqttConsumerManager = null) + { + _host = host; + _port = port; + _authService = authService; + _mqttOptions = mqttOptions; + _requiredUsername = mqttOptions.Username; + _requiredPassword = mqttOptions.Password; + _streamInitializer = streamInitializer; + _mqttConsumerManager = mqttConsumerManager; + + // Build TLS options if configured + if (mqttOptions.HasTls) + { + _sslOptions = BuildMqttSslOptions(mqttOptions); + } + } + + /// + /// The MQTT stream initializer for JetStream persistence, or null if JetStream is not enabled. + /// + internal MqttStreamInitializer? StreamInitializer => _streamInitializer; + + /// + /// The MQTT consumer manager for QoS 1/2 JetStream consumers, or null if JetStream is not enabled. + /// + internal MqttConsumerManager? ConsumerManager => _mqttConsumerManager; + public Task StartAsync(CancellationToken ct) { var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, _cts.Token); @@ -43,6 +110,12 @@ public sealed class MqttListener( set[connection] = 0; } + internal void UnregisterSubscription(MqttConnection connection, string topic) + { + if (_subscriptions.TryGetValue(topic, out var set)) + set.TryRemove(connection, out _); + } + internal async Task PublishAsync(string topic, string payload, MqttConnection sender, CancellationToken ct) { if (!_subscriptions.TryGetValue(topic, out var subscribers)) @@ -92,8 +165,47 @@ public sealed class MqttListener( session.Pending.TryRemove(packetId, out _); } + /// + /// Authenticates MQTT CONNECT credentials. Uses AuthService when available, + /// falls back to static username/password validation. + /// + internal AuthResult? AuthenticateMqtt(string? username, string? password, X509Certificate2? clientCert = null) + { + if (_authService != null) + { + var context = new ClientAuthContext + { + Opts = new ClientOptions + { + Username = username, + Password = password, + }, + Nonce = [], + ClientCertificate = clientCert, + ConnectionType = JwtConnectionTypes.Mqtt, + }; + + if (!_authService.IsAuthRequired) + return new AuthResult { Identity = username ?? string.Empty }; + + var result = _authService.Authenticate(context); + return result; + } + + // Fallback: static credential check + if (AuthService.ValidateMqttCredentials(_requiredUsername, _requiredPassword, username, password)) + return new AuthResult { Identity = username ?? string.Empty }; + + return null; + } + + /// + /// Backward-compatible simple auth check for text-protocol mode. + /// internal bool TryAuthenticate(string? username, string? password) - => AuthService.ValidateMqttCredentials(_requiredUsername, _requiredPassword, username, password); + { + return AuthenticateMqtt(username, password) != null; + } internal TimeSpan ResolveKeepAliveTimeout(int keepAliveSeconds) { @@ -103,11 +215,53 @@ public sealed class MqttListener( return TimeSpan.FromSeconds(Math.Max(keepAliveSeconds * 1.5, 1)); } + /// + /// Disconnects an existing connection with the same client-id (takeover). + /// Go reference: mqtt.go mqttHandleConnect ~line 850 duplicate client handling. + /// + internal void TakeoverExistingConnection(string clientId, MqttConnection newConnection) + { + if (_clientIdMap.TryGetValue(clientId, out var existing) && existing != newConnection) + { + existing.ForceClose(); + _connections.TryRemove(existing, out _); + } + + _clientIdMap[clientId] = newConnection; + } + + /// + /// Stores or deletes a retained message. Null payload = tombstone (delete). + /// + internal void SetRetainedMessage(string topic, string? payload) + { + if (payload == null) + _retainedMessages.TryRemove(topic, out _); + else + _retainedMessages[topic] = payload; + } + + /// + /// Gets the retained message for a topic, or null if none. + /// + internal string? GetRetainedMessage(string topic) + { + _retainedMessages.TryGetValue(topic, out var payload); + return payload; + } + internal void Unregister(MqttConnection connection) { _connections.TryRemove(connection, out _); foreach (var set in _subscriptions.Values) set.TryRemove(connection, out _); + + // Remove from client-id map if this connection is the current one + var clientId = connection.ClientId; + if (!string.IsNullOrEmpty(clientId)) + { + _clientIdMap.TryRemove(new KeyValuePair(clientId, connection)); + } } public async ValueTask DisposeAsync() @@ -124,6 +278,8 @@ public sealed class MqttListener( _connections.Clear(); _subscriptions.Clear(); _sessions.Clear(); + _clientIdMap.Clear(); + _retainedMessages.Clear(); _cts.Dispose(); } @@ -141,10 +297,50 @@ public sealed class MqttListener( break; } - var connection = new MqttConnection(client, this); - _connections[connection] = 0; _ = Task.Run(async () => { + Stream stream = client.GetStream(); + X509Certificate2? clientCert = null; + + // TLS wrapping for MQTT (TLS-first, no INFO negotiation) + if (_sslOptions != null) + { + try + { + var sslStream = new SslStream(stream, leaveInnerStreamOpen: false); + using var handshakeCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + handshakeCts.CancelAfter(TimeSpan.FromSeconds( + _mqttOptions?.TlsTimeout ?? 2.0)); + await sslStream.AuthenticateAsServerAsync(_sslOptions, handshakeCts.Token); + + clientCert = sslStream.RemoteCertificate as X509Certificate2; + + // Validate pinned certs + if (_mqttOptions?.TlsPinnedCerts != null && clientCert != null) + { + if (!TlsHelper.MatchesPinnedCert(clientCert, _mqttOptions.TlsPinnedCerts)) + { + sslStream.Dispose(); + client.Dispose(); + return; + } + } + + stream = sslStream; + } + catch + { + client.Dispose(); + return; + } + } + + // Lazily initialize MQTT JetStream streams on first connection + _streamInitializer?.EnsureStreams(); + + var connection = new MqttConnection(stream, this, UseBinaryProtocol, clientCert); + _connections[connection] = 0; + try { await connection.RunAsync(ct); @@ -157,6 +353,34 @@ public sealed class MqttListener( } } + private static SslServerAuthenticationOptions BuildMqttSslOptions(MqttOptions mqttOptions) + { + var cert = TlsHelper.LoadCertificate(mqttOptions.TlsCert!, mqttOptions.TlsKey); + var authOpts = new SslServerAuthenticationOptions + { + ServerCertificate = cert, + ClientCertificateRequired = mqttOptions.TlsVerify, + }; + + if (mqttOptions.TlsVerify && mqttOptions.TlsCaCert != null) + { + var caCerts = TlsHelper.LoadCaCertificates(mqttOptions.TlsCaCert); + authOpts.RemoteCertificateValidationCallback = (_, cert, chain, errors) => + { + if (cert == null) return false; + using var chain2 = new X509Chain(); + chain2.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust; + foreach (var ca in caCerts) + chain2.ChainPolicy.CustomTrustStore.Add(ca); + chain2.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck; + var cert2 = cert as X509Certificate2 ?? X509CertificateLoader.LoadCertificate(cert.GetRawCertData()); + return chain2.Build(cert2); + }; + } + + return authOpts; + } + private sealed class MqttSessionState { public ConcurrentDictionary Pending { get; } = new(); diff --git a/src/NATS.Server/Mqtt/MqttNatsClientAdapter.cs b/src/NATS.Server/Mqtt/MqttNatsClientAdapter.cs new file mode 100644 index 0000000..af57ba8 --- /dev/null +++ b/src/NATS.Server/Mqtt/MqttNatsClientAdapter.cs @@ -0,0 +1,111 @@ +// MqttNatsClientAdapter wraps an MqttConnection to implement INatsClient, +// enabling MQTT connections to participate in the standard NATS message routing. +// Go reference: mqtt.go — each MQTT connection behaves as a NATS client internally. + +using NATS.Server.Auth; +using NATS.Server.Protocol; +using NATS.Server.Subscriptions; +using System.Text; + +namespace NATS.Server.Mqtt; + +/// +/// Adapts an to the interface +/// so MQTT clients can be registered in the server's SubList and receive messages +/// through the standard NATS delivery path. +/// +public sealed class MqttNatsClientAdapter : INatsClient +{ + private readonly MqttConnection _connection; + private readonly Dictionary _subs = new(StringComparer.Ordinal); + + public ulong Id { get; } + public ClientKind Kind => ClientKind.Client; + public Account? Account { get; set; } + public ClientOptions? ClientOpts => null; + public ClientPermissions? Permissions { get; set; } + + public string MqttClientId => _connection.ClientId; + + public MqttNatsClientAdapter(MqttConnection connection, ulong id) + { + _connection = connection; + Id = id; + } + + /// + /// Delivers a NATS message to this MQTT client by translating the NATS subject + /// to an MQTT topic and writing a binary PUBLISH packet. + /// + public void SendMessage(string subject, string sid, string? replyTo, + ReadOnlyMemory headers, ReadOnlyMemory payload) + { + var mqttTopic = MqttTopicMapper.NatsToMqtt(subject); + // Fire-and-forget async send; MQTT delivery is best-effort for QoS 0 + _ = _connection.SendBinaryPublishAsync(mqttTopic, payload, qos: 0, + retain: false, packetId: 0, CancellationToken.None); + } + + public void SendMessageNoFlush(string subject, string sid, string? replyTo, + ReadOnlyMemory headers, ReadOnlyMemory payload) + { + // MQTT has no concept of deferred flush — deliver immediately + SendMessage(subject, sid, replyTo, headers, payload); + } + + public void SignalFlush() + { + // No-op for MQTT — each packet is written and flushed immediately + } + + public bool QueueOutbound(ReadOnlyMemory data) + { + // No-op for MQTT — binary framing, not raw NATS protocol bytes + return true; + } + + public void RemoveSubscription(string sid) + { + if (_subs.Remove(sid, out var sub)) + { + Account?.SubList.Remove(sub); + Account?.DecrementSubscriptions(); + } + } + + /// + /// Creates a NATS subscription for an MQTT topic filter and inserts it into + /// the account's SubList so NATS messages are delivered to this MQTT client. + /// + public Subscription AddSubscription(string natsSubject, string sid, string? queue = null) + { + var sub = new Subscription + { + Client = this, + Subject = natsSubject, + Sid = sid, + Queue = queue, + }; + _subs[sid] = sub; + + Account?.SubList.Insert(sub); + Account?.IncrementSubscriptions(); + return sub; + } + + /// + /// Removes all subscriptions for this adapter from the SubList. + /// Called during connection cleanup. + /// + public void RemoveAllSubscriptions() + { + foreach (var sub in _subs.Values) + { + Account?.SubList.Remove(sub); + } + + _subs.Clear(); + } + + public IReadOnlyDictionary Subscriptions => _subs; +} diff --git a/src/NATS.Server/Mqtt/MqttPacketReader.cs b/src/NATS.Server/Mqtt/MqttPacketReader.cs index 78d7ad8..921839a 100644 --- a/src/NATS.Server/Mqtt/MqttPacketReader.cs +++ b/src/NATS.Server/Mqtt/MqttPacketReader.cs @@ -1,3 +1,5 @@ +using System.Buffers; + namespace NATS.Server.Mqtt; public enum MqttControlPacketType : byte @@ -7,8 +9,13 @@ public enum MqttControlPacketType : byte ConnAck = 2, Publish = 3, PubAck = 4, + PubRec = 5, + PubRel = 6, + PubComp = 7, Subscribe = 8, SubAck = 9, + Unsubscribe = 10, + UnsubAck = 11, PingReq = 12, PingResp = 13, Disconnect = 14, @@ -22,6 +29,9 @@ public sealed record MqttControlPacket( public static class MqttPacketReader { + /// + /// Parses a complete MQTT control packet from a contiguous span. + /// public static MqttControlPacket Read(ReadOnlySpan buffer) { if (buffer.Length < 2) @@ -42,6 +52,74 @@ public static class MqttPacketReader return new MqttControlPacket(type, flags, remainingLength, payload); } + /// + /// Attempts to read a complete MQTT control packet from a . + /// Returns false if more data is needed (partial read). Advances + /// past the packet bytes on success. + /// Used with for incremental parsing. + /// + public static bool TryRead(ReadOnlySequence buffer, out MqttControlPacket? packet, out SequencePosition consumed) + { + packet = null; + consumed = buffer.Start; + + if (buffer.Length < 2) + return false; + + // Read the fixed header byte + var reader = new SequenceReader(buffer); + reader.TryRead(out var firstByte); + var type = (MqttControlPacketType)(firstByte >> 4); + var flags = (byte)(firstByte & 0x0F); + + // Decode remaining length (variable 1-4 bytes) + var multiplier = 1; + var remainingLength = 0; + var lengthBytesConsumed = 0; + + while (lengthBytesConsumed < 4) + { + if (!reader.TryRead(out var digit)) + return false; // need more data + + lengthBytesConsumed++; + remainingLength += (digit & 0x7F) * multiplier; + + if ((digit & 0x80) == 0) + break; + + multiplier *= 128; + + if (lengthBytesConsumed == 4) + throw new FormatException("Invalid MQTT remaining length encoding."); + } + + if (remainingLength > MqttProtocolConstants.MaxPayloadSize) + throw new FormatException("MQTT packet remaining length exceeds protocol maximum."); + + // Check if we have the full payload + var headerSize = 1 + lengthBytesConsumed; + var totalPacketSize = headerSize + remainingLength; + if (buffer.Length < totalPacketSize) + return false; // need more data + + // Extract payload + byte[] payload; + if (remainingLength == 0) + { + payload = []; + } + else + { + payload = new byte[remainingLength]; + buffer.Slice(headerSize, remainingLength).CopyTo(payload); + } + + packet = new MqttControlPacket(type, flags, remainingLength, payload); + consumed = buffer.GetPosition(totalPacketSize); + return true; + } + internal static int DecodeRemainingLength(ReadOnlySpan encoded, out int consumed) { var multiplier = 1; diff --git a/src/NATS.Server/Mqtt/MqttPacketWriter.cs b/src/NATS.Server/Mqtt/MqttPacketWriter.cs index ad9802b..e27c288 100644 --- a/src/NATS.Server/Mqtt/MqttPacketWriter.cs +++ b/src/NATS.Server/Mqtt/MqttPacketWriter.cs @@ -33,6 +33,119 @@ public static class MqttPacketWriter return buffer; } + /// + /// Writes a CONNACK packet. Go reference: mqtt.go mqttConnAck. + /// + /// 0x01 if resuming existing session, 0x00 otherwise. + /// CONNACK return code (0x00 = accepted). + public static byte[] WriteConnAck(byte sessionPresent, byte returnCode) + { + ReadOnlySpan payload = [sessionPresent, returnCode]; + return Write(MqttControlPacketType.ConnAck, payload); + } + + /// + /// Writes a PUBACK packet (QoS 1 acknowledgment). + /// + public static byte[] WritePubAck(ushort packetId) + { + Span payload = stackalloc byte[2]; + BinaryPrimitives.WriteUInt16BigEndian(payload, packetId); + return Write(MqttControlPacketType.PubAck, payload); + } + + /// + /// Writes a SUBACK packet with granted QoS values per subscription filter. + /// + public static byte[] WriteSubAck(ushort packetId, ReadOnlySpan grantedQoS) + { + var payload = new byte[2 + grantedQoS.Length]; + BinaryPrimitives.WriteUInt16BigEndian(payload.AsSpan(0, 2), packetId); + grantedQoS.CopyTo(payload.AsSpan(2)); + return Write(MqttControlPacketType.SubAck, payload); + } + + /// + /// Writes an UNSUBACK packet. + /// + public static byte[] WriteUnsubAck(ushort packetId) + { + Span payload = stackalloc byte[2]; + BinaryPrimitives.WriteUInt16BigEndian(payload, packetId); + return Write(MqttControlPacketType.UnsubAck, payload); + } + + /// + /// Writes a PINGRESP packet (no payload). + /// + public static byte[] WritePingResp() + => Write(MqttControlPacketType.PingResp, []); + + /// + /// Writes a PUBREC packet (QoS 2 step 1 response). + /// + public static byte[] WritePubRec(ushort packetId) + { + Span payload = stackalloc byte[2]; + BinaryPrimitives.WriteUInt16BigEndian(payload, packetId); + return Write(MqttControlPacketType.PubRec, payload); + } + + /// + /// Writes a PUBREL packet (QoS 2 step 2). Fixed-header flags must be 0x02 per MQTT spec. + /// + public static byte[] WritePubRel(ushort packetId) + { + Span payload = stackalloc byte[2]; + BinaryPrimitives.WriteUInt16BigEndian(payload, packetId); + return Write(MqttControlPacketType.PubRel, payload, flags: 0x02); + } + + /// + /// Writes a PUBCOMP packet (QoS 2 step 3 response). + /// + public static byte[] WritePubComp(ushort packetId) + { + Span payload = stackalloc byte[2]; + BinaryPrimitives.WriteUInt16BigEndian(payload, packetId); + return Write(MqttControlPacketType.PubComp, payload); + } + + /// + /// Writes an MQTT PUBLISH packet for delivery to a client. + /// + public static byte[] WritePublish(string topic, ReadOnlySpan payload, byte qos = 0, + bool retain = false, bool dup = false, ushort packetId = 0) + { + var topicBytes = Encoding.UTF8.GetBytes(topic); + var variableHeaderLen = 2 + topicBytes.Length + (qos > 0 ? 2 : 0); + var totalPayload = new byte[variableHeaderLen + payload.Length]; + var pos = 0; + + // Topic name (length-prefixed) + BinaryPrimitives.WriteUInt16BigEndian(totalPayload.AsSpan(pos, 2), (ushort)topicBytes.Length); + pos += 2; + topicBytes.CopyTo(totalPayload.AsSpan(pos)); + pos += topicBytes.Length; + + // Packet ID (only for QoS > 0) + if (qos > 0) + { + BinaryPrimitives.WriteUInt16BigEndian(totalPayload.AsSpan(pos, 2), packetId); + pos += 2; + } + + // Application payload + payload.CopyTo(totalPayload.AsSpan(pos)); + + byte flags = 0; + if (dup) flags |= 0x08; + flags |= (byte)((qos & 0x03) << 1); + if (retain) flags |= 0x01; + + return Write(MqttControlPacketType.Publish, totalPayload, flags); + } + internal static byte[] EncodeRemainingLength(int value) { if (value < 0 || value > MqttProtocolConstants.MaxPayloadSize) diff --git a/src/NATS.Server/Mqtt/MqttQoS1Tracker.cs b/src/NATS.Server/Mqtt/MqttQoS1Tracker.cs index c517cb1..851e0d2 100644 --- a/src/NATS.Server/Mqtt/MqttQoS1Tracker.cs +++ b/src/NATS.Server/Mqtt/MqttQoS1Tracker.cs @@ -1,6 +1,7 @@ -// QoS 1 outgoing message tracker for MQTT. +// QoS 1 outgoing message tracker for MQTT with JetStream ack integration. // Go reference: golang/nats-server/server/mqtt.go // QoS 1 outbound tracking — mqttProcessPub (~line 1200) +// trackPublish — maps packet IDs to stream sequences for ack tracking. using System.Collections.Concurrent; @@ -8,8 +9,8 @@ namespace NATS.Server.Mqtt; /// /// Tracks outgoing QoS 1 messages pending PUBACK from the client. -/// Messages are stored with their packet ID and can be redelivered on reconnect. -/// Go reference: server/mqtt.go — mqttProcessPub (QoS 1 outbound tracking). +/// Maps packet IDs to JetStream stream sequences for ack-based cleanup. +/// Go reference: server/mqtt.go — mqttProcessPub, trackPublish. /// public sealed class MqttQoS1Tracker { @@ -24,7 +25,7 @@ public sealed class MqttQoS1Tracker /// Registers an outgoing QoS 1 message and assigns a packet ID. /// Returns the assigned packet ID. /// - public ushort Register(string topic, byte[] payload) + public ushort Register(string topic, byte[] payload, ulong streamSequence = 0) { var id = GetNextPacketId(); _pending[id] = new QoS1PendingMessage @@ -34,17 +35,18 @@ public sealed class MqttQoS1Tracker Payload = payload, SentAtUtc = DateTime.UtcNow, DeliveryCount = 1, + StreamSequence = streamSequence, }; return id; } /// /// Acknowledges receipt of a PUBACK for the given packet ID. - /// Returns true if the message was found and removed. + /// Returns the pending message if found, or null. /// - public bool Acknowledge(ushort packetId) + public QoS1PendingMessage? Acknowledge(ushort packetId) { - return _pending.TryRemove(packetId, out _); + return _pending.TryRemove(packetId, out var msg) ? msg : null; } /// @@ -93,4 +95,11 @@ public sealed class QoS1PendingMessage public byte[] Payload { get; init; } = []; public DateTime SentAtUtc { get; set; } public int DeliveryCount { get; set; } = 1; + + /// + /// JetStream stream sequence for this message. 0 if not backed by JetStream. + /// Used to ack the message in the stream on PUBACK. + /// Go reference: server/mqtt.go trackPublish — maps packet ID → stream sequence. + /// + public ulong StreamSequence { get; init; } } diff --git a/src/NATS.Server/Mqtt/MqttRetainedStore.cs b/src/NATS.Server/Mqtt/MqttRetainedStore.cs index 70caab1..a4b16e4 100644 --- a/src/NATS.Server/Mqtt/MqttRetainedStore.cs +++ b/src/NATS.Server/Mqtt/MqttRetainedStore.cs @@ -110,6 +110,8 @@ public sealed class MqttRetainedStore /// /// Sets (or clears) the retained message and persists to backing store. + /// Uses the $MQTT_rmsgs stream with MaxMsgsPer=1 for per-subject latest-wins. + /// Empty payload = tombstone (delete retained). /// Go reference: server/mqtt.go mqttHandleRetainedMsg with JetStream. /// public async Task SetRetainedAsync(string topic, ReadOnlyMemory payload, CancellationToken ct = default) @@ -118,13 +120,17 @@ public sealed class MqttRetainedStore if (_backingStore is not null) { + var subject = $"{MqttProtocolConstants.RetainedMsgsStreamSubject}{topic}"; if (payload.IsEmpty) { - // Clear — the in-memory clear above is sufficient for this implementation. - // A full implementation would publish a tombstone to JetStream. + // Tombstone: remove from stream + var msg = await _backingStore.LoadLastBySubjectAsync(subject, ct); + if (msg is not null) + await _backingStore.RemoveAsync(msg.Sequence, ct); return; } - await _backingStore.AppendAsync($"$MQTT.rmsgs.{topic}", payload, ct); + + await _backingStore.AppendAsync(subject, payload, ct); } } @@ -144,12 +150,10 @@ public sealed class MqttRetainedStore if (_backingStore is not null) { - var messages = await _backingStore.ListAsync(ct); - foreach (var msg in messages) - { - if (msg.Subject == $"$MQTT.rmsgs.{topic}") - return msg.Payload.ToArray(); - } + var subject = $"{MqttProtocolConstants.RetainedMsgsStreamSubject}{topic}"; + var msg = await _backingStore.LoadLastBySubjectAsync(subject, ct); + if (msg is not null) + return msg.Payload.ToArray(); } return null; diff --git a/src/NATS.Server/Mqtt/MqttSessionStore.cs b/src/NATS.Server/Mqtt/MqttSessionStore.cs index 9ea9531..84004a7 100644 --- a/src/NATS.Server/Mqtt/MqttSessionStore.cs +++ b/src/NATS.Server/Mqtt/MqttSessionStore.cs @@ -372,25 +372,30 @@ public sealed class MqttSessionStore if (cleanSession) { DeleteSession(clientId); - // For now the in-memory delete is sufficient; a full implementation would - // publish a tombstone or use sequence lookup to remove from JetStream. + + // Remove from JetStream backing store + if (_backingStore is not null) + { + var subject = $"{MqttProtocolConstants.SessStreamSubjectPrefix}{clientId}"; + var msg = await _backingStore.LoadLastBySubjectAsync(subject, ct); + if (msg is not null) + await _backingStore.RemoveAsync(msg.Sequence, ct); + } + return; } - // Try to load from backing store + // Try to load from backing store using subject-keyed lookup if (_backingStore is not null) { - var messages = await _backingStore.ListAsync(ct); - foreach (var msg in messages) + var subject = $"{MqttProtocolConstants.SessStreamSubjectPrefix}{clientId}"; + var msg = await _backingStore.LoadLastBySubjectAsync(subject, ct); + if (msg is not null) { - if (msg.Subject == $"$MQTT.sess.{clientId}") + var data = System.Text.Json.JsonSerializer.Deserialize(msg.Payload.Span); + if (data is not null) { - var data = System.Text.Json.JsonSerializer.Deserialize(msg.Payload.Span); - if (data is not null) - { - SaveSession(data); - } - break; + SaveSession(data); } } } @@ -412,6 +417,7 @@ public sealed class MqttSessionStore /// /// Saves the session to the backing JetStream store if available. + /// Uses the $MQTT_sess stream with MaxMsgsPer=1 for idempotent per-subject writes. /// Go reference: server/mqtt.go mqttStoreSession. /// public async Task SaveSessionAsync(string clientId, CancellationToken ct = default) @@ -421,7 +427,7 @@ public sealed class MqttSessionStore return; var json = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(session); - await _backingStore.AppendAsync($"$MQTT.sess.{clientId}", json, ct); + await _backingStore.AppendAsync($"{MqttProtocolConstants.SessStreamSubjectPrefix}{clientId}", json, ct); } /// diff --git a/src/NATS.Server/Mqtt/MqttStreamInitializer.cs b/src/NATS.Server/Mqtt/MqttStreamInitializer.cs new file mode 100644 index 0000000..a64dbbb --- /dev/null +++ b/src/NATS.Server/Mqtt/MqttStreamInitializer.cs @@ -0,0 +1,70 @@ +// Initializes the 5 internal MQTT JetStream streams per account. +// Go reference: golang/nats-server/server/mqtt.go mqttCreateAccountSessionManager ~line 600 +// Stream creation for $MQTT_msgs, $MQTT_sess, $MQTT_rmsgs, $MQTT_qos2in, $MQTT_out. + +using NATS.Server.JetStream; +using NATS.Server.JetStream.Models; + +namespace NATS.Server.Mqtt; + +/// +/// Lazily creates the 5 internal MQTT JetStream streams required for MQTT persistence. +/// Called on first MQTT connection per account. +/// Go reference: server/mqtt.go mqttCreateAccountSessionManager ~line 600. +/// +public sealed class MqttStreamInitializer +{ + private readonly StreamManager _streamManager; + private volatile bool _initialized; + private readonly Lock _initLock = new(); + + public MqttStreamInitializer(StreamManager streamManager) + { + _streamManager = streamManager; + } + + /// + /// Whether the MQTT streams have been initialized. + /// + public bool IsInitialized => _initialized; + + /// + /// Ensures the 5 internal MQTT streams exist. Idempotent — safe to call multiple times. + /// Go reference: server/mqtt.go mqttCreateAccountSessionManager. + /// + public void EnsureStreams() + { + if (_initialized) + return; + + lock (_initLock) + { + if (_initialized) + return; + + CreateStream(MqttProtocolConstants.SessStreamName, [$"{MqttProtocolConstants.SessStreamSubjectPrefix}>"], maxMsgsPer: 1); + CreateStream(MqttProtocolConstants.StreamName, [$"{MqttProtocolConstants.StreamSubjectPrefix}>"], retention: RetentionPolicy.Interest); + CreateStream(MqttProtocolConstants.RetainedMsgsStreamName, [$"{MqttProtocolConstants.RetainedMsgsStreamSubject}>"], maxMsgsPer: 1); + CreateStream(MqttProtocolConstants.QoS2IncomingMsgsStreamName, [$"{MqttProtocolConstants.QoS2IncomingMsgsStreamSubjectPrefix}>"], maxMsgsPer: 1); + CreateStream(MqttProtocolConstants.OutStreamName, [$"{MqttProtocolConstants.OutSubjectPrefix}>"], retention: RetentionPolicy.Interest); + + _initialized = true; + } + } + + private void CreateStream(string name, List subjects, RetentionPolicy retention = RetentionPolicy.Limits, int maxMsgsPer = 0) + { + if (_streamManager.Exists(name)) + return; + + _streamManager.CreateOrUpdate(new StreamConfig + { + Name = name, + Subjects = subjects, + Storage = StorageType.Memory, + Retention = retention, + MaxMsgsPer = maxMsgsPer, + Replicas = 1, + }); + } +} diff --git a/src/NATS.Server/Mqtt/MqttTopicMapper.cs b/src/NATS.Server/Mqtt/MqttTopicMapper.cs new file mode 100644 index 0000000..076c7d6 --- /dev/null +++ b/src/NATS.Server/Mqtt/MqttTopicMapper.cs @@ -0,0 +1,136 @@ +// Full Go-parity MQTT topic ↔ NATS subject translation. +// Go reference: golang/nats-server/server/mqtt.go mqttToNATSSubjectConversion ~line 2200 +// +// Rules: +// MQTT → NATS: +// '/' → '.' (separator) +// '+' → '*' (single-level wildcard) +// '#' → '>' (multi-level wildcard) +// '.' in MQTT topics must be escaped (replaced with a placeholder) +// Empty levels (leading/trailing/consecutive slashes) produce empty tokens +// '$' prefix topics are protected from wildcard matching per MQTT spec [MQTT-4.7.2-1] +// +// NATS → MQTT (reverse): +// '.' → '/' +// '*' → '+' +// '>' → '#' + +using System.Text; + +namespace NATS.Server.Mqtt; + +/// +/// Translates MQTT topics/filters to NATS subjects and vice versa with full Go parity. +/// Go reference: mqtt.go mqttToNATSSubjectConversion, mqttNATSToMQTTSubjectConversion. +/// +public static class MqttTopicMapper +{ + // Escape sequence for dots that appear in MQTT topic names. + // Go uses _DOT_ internally to represent a literal dot in the NATS subject. + private const string DotEscape = "_DOT_"; + private const string DotEscapeReverse = "."; + + /// + /// Translates an MQTT topic or filter to a NATS subject. + /// Handles wildcards, dot escaping, empty levels, and '$' prefix protection. + /// + public static string MqttToNats(string mqttTopic) + { + if (mqttTopic.Length == 0) + return string.Empty; + + var sb = new StringBuilder(mqttTopic.Length); + + for (var i = 0; i < mqttTopic.Length; i++) + { + switch (mqttTopic[i]) + { + case '/': + sb.Append('.'); + break; + case '+': + sb.Append('*'); + break; + case '#': + sb.Append('>'); + break; + case '.': + // Dots in MQTT topic names must be escaped for NATS + sb.Append(DotEscape); + break; + default: + sb.Append(mqttTopic[i]); + break; + } + } + + return sb.ToString(); + } + + /// + /// Translates a NATS subject back to an MQTT topic. + /// Reverses the mapping: '.' → '/', '*' → '+', '>' → '#', '_DOT_' → '.'. + /// + public static string NatsToMqtt(string natsSubject) + { + if (natsSubject.Length == 0) + return string.Empty; + + // First, replace _DOT_ escape sequences back to dots + var working = natsSubject.Replace(DotEscape, "\x00"); + + var sb = new StringBuilder(working.Length); + for (var i = 0; i < working.Length; i++) + { + switch (working[i]) + { + case '.': + sb.Append('/'); + break; + case '*': + sb.Append('+'); + break; + case '>': + sb.Append('#'); + break; + case '\x00': + sb.Append('.'); + break; + default: + sb.Append(working[i]); + break; + } + } + + return sb.ToString(); + } + + /// + /// Returns true if an MQTT topic starts with '$', which means it should + /// NOT be matched by wildcard subscriptions (MQTT spec [MQTT-4.7.2-1]). + /// Topics starting with '$' are reserved for system/server use. + /// + public static bool IsDollarTopic(string mqttTopic) + => mqttTopic.Length > 0 && mqttTopic[0] == '$'; + + /// + /// Returns true if an MQTT topic filter starts with '$', indicating + /// it explicitly targets system topics. + /// + public static bool IsDollarFilter(string mqttFilter) + => mqttFilter.Length > 0 && mqttFilter[0] == '$'; + + /// + /// Checks if a wildcard filter would match a '$' topic. + /// Per MQTT spec, wildcard filters (starting with '#' or '+') must NOT + /// match topics beginning with '$'. Only explicit '$' filters match '$' topics. + /// + public static bool WildcardMatchesDollarTopic(string mqttFilter, string mqttTopic) + { + if (!IsDollarTopic(mqttTopic)) + return true; // non-$ topics are always matchable + + // $ topics only matched by filters that also start with $ + return IsDollarFilter(mqttFilter); + } +} diff --git a/src/NATS.Server/NatsClient.cs b/src/NATS.Server/NatsClient.cs index c39966d..cea518f 100644 --- a/src/NATS.Server/NatsClient.cs +++ b/src/NATS.Server/NatsClient.cs @@ -21,10 +21,10 @@ namespace NATS.Server; public interface IMessageRouter { void ProcessMessage(string subject, string? replyTo, ReadOnlyMemory headers, - ReadOnlyMemory payload, NatsClient sender); - void RemoveClient(NatsClient client); - void PublishConnectEvent(NatsClient client); - void PublishDisconnectEvent(NatsClient client); + ReadOnlyMemory payload, INatsClient sender); + void RemoveClient(INatsClient client); + void PublishConnectEvent(INatsClient client); + void PublishDisconnectEvent(INatsClient client); } public interface ISubListAccess @@ -98,6 +98,12 @@ public sealed class NatsClient : INatsClient, IDisposable public Account? Account { get; private set; } public ClientPermissions? Permissions => _permissions; + /// + /// MQTT client-id for monitoring (/connz mqtt_client field). + /// Set when this NatsClient proxies an MQTT connection via MqttNatsClientAdapter. + /// + public string? MqttClientId { get; set; } + private readonly ClientFlagHolder _flags = new(); public bool ConnectReceived => _flags.HasFlag(ClientFlags.ConnectReceived); public ClientClosedReason CloseReason { get; private set; } diff --git a/src/NATS.Server/NatsServer.cs b/src/NATS.Server/NatsServer.cs index 7558a79..f10f423 100644 --- a/src/NATS.Server/NatsServer.cs +++ b/src/NATS.Server/NatsServer.cs @@ -115,6 +115,13 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable public string ServerName => _serverInfo.ServerName; public int ClientCount => _clients.Count; public int Port => _options.Port; + + /// + /// Returns the actual bound port of the MQTT listener, or null if MQTT is not enabled. + /// Used by VarzHandler for monitoring. + /// + public int? MqttListenerPort => _mqttListener?.Port; + public Account SystemAccount => _systemAccount; public string ServerNKey { get; } public InternalEventSystem? EventSystem => _eventSystem; @@ -914,11 +921,23 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable if (_options.Mqtt is { Port: > 0 } mqttOptions) { var mqttHost = string.IsNullOrWhiteSpace(mqttOptions.Host) ? _options.Host : mqttOptions.Host; + + // Create MQTT JetStream components if JetStream is enabled + MqttStreamInitializer? mqttStreamInit = null; + MqttConsumerManager? mqttConsumerMgr = null; + if (_jetStreamStreamManager != null && _jetStreamConsumerManager != null) + { + mqttStreamInit = new Mqtt.MqttStreamInitializer(_jetStreamStreamManager); + mqttConsumerMgr = new Mqtt.MqttConsumerManager(_jetStreamStreamManager, _jetStreamConsumerManager); + } + _mqttListener = new MqttListener( mqttHost, mqttOptions.Port, - mqttOptions.Username, - mqttOptions.Password); + _authService, + mqttOptions, + mqttStreamInit, + mqttConsumerMgr); await _mqttListener.StartAsync(linked.Token); } if (_jetStreamService != null) @@ -1316,8 +1335,12 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable } public void ProcessMessage(string subject, string? replyTo, ReadOnlyMemory headers, - ReadOnlyMemory payload, NatsClient sender) + ReadOnlyMemory payload, INatsClient sender) { + // Cast to NatsClient for operations that require it (JetStream pub-ack, stats). + // Non-NatsClient senders (e.g. MqttNatsClientAdapter) skip those code paths. + var natsClient = sender as NatsClient; + if (replyTo != null && subject.StartsWith("$JS.API", StringComparison.Ordinal) && _jetStreamApiRouter != null) @@ -1327,10 +1350,11 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable // Go reference: consumer.go:4276 processNextMsgRequest if (subject.StartsWith(JetStream.Api.JetStreamApiSubjects.ConsumerNext, StringComparison.Ordinal) && _jetStreamConsumerManager != null - && _jetStreamStreamManager != null) + && _jetStreamStreamManager != null + && natsClient != null) { Interlocked.Increment(ref _stats.JetStreamApiTotal); - DeliverPullFetchMessages(subject, replyTo, payload, sender); + DeliverPullFetchMessages(subject, replyTo, payload, natsClient); return; } @@ -1353,7 +1377,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable if (TryCaptureJetStreamPublish(subject, payload, out var pubAck)) { - sender.RecordJetStreamPubAck(pubAck); + natsClient?.RecordJetStreamPubAck(pubAck); // Replicate data messages to cluster peers so their JetStream stores also capture them. // Route forwarding below is gated on subscriber interest, which JetStream streams don't @@ -1426,18 +1450,34 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable if (queueGroup.Length == 0) continue; // Simple round-robin -- pick based on total delivered across group - var idx = Math.Abs((int)Interlocked.Increment(ref sender.OutMsgs)) % queueGroup.Length; - // Undo the OutMsgs increment -- it will be incremented properly in SendMessageNoFlush - Interlocked.Decrement(ref sender.OutMsgs); - - for (int attempt = 0; attempt < queueGroup.Length; attempt++) + if (natsClient != null) { - var sub = queueGroup[(idx + attempt) % queueGroup.Length]; - if (sub.Client != null && (sub.Client != sender || (sender.ClientOpts?.Echo ?? true))) + var idx = Math.Abs((int)Interlocked.Increment(ref natsClient.OutMsgs)) % queueGroup.Length; + // Undo the OutMsgs increment -- it will be incremented properly in SendMessageNoFlush + Interlocked.Decrement(ref natsClient.OutMsgs); + + for (int attempt = 0; attempt < queueGroup.Length; attempt++) { - DeliverMessage(sub, subject, replyTo, headers, payload, pcd); - delivered = true; - break; + var sub = queueGroup[(idx + attempt) % queueGroup.Length]; + if (sub.Client != null && (sub.Client != sender || (sender.ClientOpts?.Echo ?? true))) + { + DeliverMessage(sub, subject, replyTo, headers, payload, pcd); + delivered = true; + break; + } + } + } + else + { + // Non-NatsClient sender: simple first-match + foreach (var sub in queueGroup) + { + if (sub.Client != null && sub.Client != sender) + { + DeliverMessage(sub, subject, replyTo, headers, payload, pcd); + delivered = true; + break; + } } } } @@ -1471,9 +1511,9 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable // No-responders: if nobody received the message and the publisher // opted in, send back a 503 status HMSG on the reply subject. - if (!delivered && replyTo != null && sender.ClientOpts?.NoResponders == true) + if (!delivered && replyTo != null && sender.ClientOpts?.NoResponders == true && natsClient != null) { - SendNoResponders(sender, replyTo); + SendNoResponders(natsClient, replyTo); } } @@ -2267,9 +2307,9 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable /// Publishes a $SYS.ACCOUNT.{account}.CONNECT advisory when a client /// completes authentication. Maps to Go's sendConnectEvent in events.go. /// - public void PublishConnectEvent(NatsClient client) + public void PublishConnectEvent(INatsClient client) { - if (_eventSystem == null) return; + if (_eventSystem == null || client is not NatsClient natsClient) return; var accountName = client.Account?.Name ?? Account.GlobalAccountName; var subject = string.Format(EventSubjects.ConnectEvent, accountName); var evt = new ConnectEventMsg @@ -2277,7 +2317,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable Id = Guid.NewGuid().ToString("N"), Time = DateTime.UtcNow, Server = BuildEventServerInfo(), - Client = BuildEventClientInfo(client), + Client = BuildEventClientInfo(natsClient), }; SendInternalMsg(subject, null, evt); } @@ -2286,9 +2326,9 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable /// Publishes a $SYS.ACCOUNT.{account}.DISCONNECT advisory when a client /// disconnects. Maps to Go's sendDisconnectEvent in events.go. /// - public void PublishDisconnectEvent(NatsClient client) + public void PublishDisconnectEvent(INatsClient client) { - if (_eventSystem == null) return; + if (_eventSystem == null || client is not NatsClient natsClient) return; var accountName = client.Account?.Name ?? Account.GlobalAccountName; var subject = string.Format(EventSubjects.DisconnectEvent, accountName); var evt = new DisconnectEventMsg @@ -2296,62 +2336,71 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable Id = Guid.NewGuid().ToString("N"), Time = DateTime.UtcNow, Server = BuildEventServerInfo(), - Client = BuildEventClientInfo(client), + Client = BuildEventClientInfo(natsClient), Sent = new DataStats { - Msgs = Interlocked.Read(ref client.OutMsgs), - Bytes = Interlocked.Read(ref client.OutBytes), + Msgs = Interlocked.Read(ref natsClient.OutMsgs), + Bytes = Interlocked.Read(ref natsClient.OutBytes), }, Received = new DataStats { - Msgs = Interlocked.Read(ref client.InMsgs), - Bytes = Interlocked.Read(ref client.InBytes), + Msgs = Interlocked.Read(ref natsClient.InMsgs), + Bytes = Interlocked.Read(ref natsClient.InBytes), }, - Reason = client.CloseReason.ToReasonString(), + Reason = natsClient.CloseReason.ToReasonString(), }; SendInternalMsg(subject, null, evt); } - public void RemoveClient(NatsClient client) + public void RemoveClient(INatsClient client) { - // Publish disconnect advisory before removing client state - if (client.ConnectReceived) - PublishDisconnectEvent(client); + if (client is not NatsClient natsClient) + { + // Non-NatsClient (e.g. MqttNatsClientAdapter) — basic cleanup + _clients.TryRemove(client.Id, out _); + var subList = client.Account?.SubList ?? _globalAccount.SubList; + client.Account?.RemoveClient(client.Id); + return; + } - _clients.TryRemove(client.Id, out _); - _logger.LogDebug("Removed client {ClientId}", client.Id); + // Publish disconnect advisory before removing client state + if (natsClient.ConnectReceived) + PublishDisconnectEvent(natsClient); + + _clients.TryRemove(natsClient.Id, out _); + _logger.LogDebug("Removed client {ClientId}", natsClient.Id); var (tlsPeerCertSubject, tlsPeerCertSubjectPkSha256, tlsPeerCertSha256) = - TlsPeerCertMapper.ToClosedFields(client.TlsState?.PeerCert); - var (jwt, issuerKey, tags) = ExtractJwtMetadata(client.ClientOpts?.JWT); - var proxyKey = ExtractProxyKey(client.ClientOpts?.Username); + TlsPeerCertMapper.ToClosedFields(natsClient.TlsState?.PeerCert); + var (jwt, issuerKey, tags) = ExtractJwtMetadata(natsClient.ClientOpts?.JWT); + var proxyKey = ExtractProxyKey(natsClient.ClientOpts?.Username); // Snapshot for closed-connections tracking (ring buffer auto-overwrites oldest when full) _closedClients.Add(new ClosedClient { - Cid = client.Id, - Ip = client.RemoteIp ?? "", - Port = client.RemotePort, - Start = client.StartTime, + Cid = natsClient.Id, + Ip = natsClient.RemoteIp ?? "", + Port = natsClient.RemotePort, + Start = natsClient.StartTime, Stop = DateTime.UtcNow, - Reason = client.CloseReason.ToReasonString(), - Name = client.ClientOpts?.Name ?? "", - Lang = client.ClientOpts?.Lang ?? "", - Version = client.ClientOpts?.Version ?? "", - AuthorizedUser = client.ClientOpts?.Username ?? "", - Account = client.Account?.Name ?? "", - InMsgs = Interlocked.Read(ref client.InMsgs), - OutMsgs = Interlocked.Read(ref client.OutMsgs), - InBytes = Interlocked.Read(ref client.InBytes), - OutBytes = Interlocked.Read(ref client.OutBytes), - NumSubs = (uint)client.Subscriptions.Count, - Rtt = client.Rtt, - TlsVersion = client.TlsState?.TlsVersion ?? "", - TlsCipherSuite = client.TlsState?.CipherSuite ?? "", + Reason = natsClient.CloseReason.ToReasonString(), + Name = natsClient.ClientOpts?.Name ?? "", + Lang = natsClient.ClientOpts?.Lang ?? "", + Version = natsClient.ClientOpts?.Version ?? "", + AuthorizedUser = natsClient.ClientOpts?.Username ?? "", + Account = natsClient.Account?.Name ?? "", + InMsgs = Interlocked.Read(ref natsClient.InMsgs), + OutMsgs = Interlocked.Read(ref natsClient.OutMsgs), + InBytes = Interlocked.Read(ref natsClient.InBytes), + OutBytes = Interlocked.Read(ref natsClient.OutBytes), + NumSubs = (uint)natsClient.Subscriptions.Count, + Rtt = natsClient.Rtt, + TlsVersion = natsClient.TlsState?.TlsVersion ?? "", + TlsCipherSuite = natsClient.TlsState?.CipherSuite ?? "", TlsPeerCertSubject = tlsPeerCertSubject, TlsPeerCertSubjectPkSha256 = tlsPeerCertSubjectPkSha256, TlsPeerCertSha256 = tlsPeerCertSha256, - MqttClient = "", // populated when MQTT transport is implemented + MqttClient = natsClient.MqttClientId ?? "", Stalls = 0, Jwt = jwt, IssuerKey = issuerKey, @@ -2360,9 +2409,9 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable ProxyKey = proxyKey, }); - var subList = client.Account?.SubList ?? _globalAccount.SubList; - client.RemoveAllSubscriptions(subList); - client.Account?.RemoveClient(client.Id); + var ncSubList = natsClient.Account?.SubList ?? _globalAccount.SubList; + natsClient.RemoveAllSubscriptions(ncSubList); + natsClient.Account?.RemoveClient(natsClient.Id); } private void TrackEarlyClosedClient(Socket socket, ulong clientId, ClientClosedReason reason) diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAdvancedParityTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAdvancedParityTests.cs index a7651b8..857c83e 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAdvancedParityTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAdvancedParityTests.cs @@ -28,6 +28,7 @@ public class MqttAdvancedParityTests public async Task Subscribe_exact_topic_receives_matching_publish() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -55,6 +56,7 @@ public class MqttAdvancedParityTests public async Task Subscribe_exact_topic_does_not_receive_non_matching_publish() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -82,6 +84,7 @@ public class MqttAdvancedParityTests public async Task Subscribe_two_level_topic_receives_matching_publish() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -109,6 +112,7 @@ public class MqttAdvancedParityTests public async Task Unsubscribe_stops_message_delivery() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -156,6 +160,7 @@ public class MqttAdvancedParityTests public async Task Publish_qos0_and_qos1_both_work() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -396,6 +401,7 @@ public class MqttAdvancedParityTests public async Task Subscription_matching_is_case_sensitive() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -430,6 +436,7 @@ public class MqttAdvancedParityTests public async Task Clean_session_reconnect_produces_no_pending_messages() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -461,6 +468,7 @@ public class MqttAdvancedParityTests public async Task Duplicate_client_id_second_connection_accepted() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -485,6 +493,7 @@ public class MqttAdvancedParityTests public async Task Server_accepts_tcp_connections() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -503,6 +512,7 @@ public class MqttAdvancedParityTests public async Task Connack_is_first_response_to_connect() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -523,6 +533,7 @@ public class MqttAdvancedParityTests public async Task Multiple_subscriptions_to_same_topic_do_not_cause_duplicates() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -557,6 +568,7 @@ public class MqttAdvancedParityTests public async Task Rapid_connect_disconnect_cycles_do_not_crash_server() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -578,6 +590,7 @@ public class MqttAdvancedParityTests public async Task Unacked_qos1_messages_are_redelivered_on_reconnect() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -688,6 +701,7 @@ public class MqttAdvancedParityTests public async Task Listener_allocates_dynamic_port_when_zero_specified() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -704,6 +718,7 @@ public class MqttAdvancedParityTests public async Task Multiple_subscribers_on_different_topics_receive_correct_messages() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -747,6 +762,7 @@ public class MqttAdvancedParityTests public async Task Client_connect_and_disconnect_lifecycle() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -836,6 +852,7 @@ public class MqttAdvancedParityTests public async Task Persistent_session_redelivers_unacked_on_reconnect() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -888,6 +905,7 @@ public class MqttAdvancedParityTests public async Task Concurrent_publishers_deliver_to_single_subscriber() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAuthIntegrationTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAuthIntegrationTests.cs index 24d2535..5b4d64a 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAuthIntegrationTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAuthIntegrationTests.cs @@ -10,6 +10,7 @@ public class MqttAuthIntegrationTests public async Task Invalid_mqtt_credentials_or_keepalive_timeout_close_session_with_protocol_error() { await using var listener = new MqttListener("127.0.0.1", 0, requiredUsername: "mqtt", requiredPassword: "secret"); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAuthParityTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAuthParityTests.cs index 277594a..b4c5b14 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAuthParityTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttAuthParityTests.cs @@ -24,6 +24,7 @@ public class MqttAuthParityTests "127.0.0.1", 0, requiredUsername: "mqtt", requiredPassword: "client"); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -43,6 +44,7 @@ public class MqttAuthParityTests "127.0.0.1", 0, requiredUsername: "mqtt", requiredPassword: "client"); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -64,6 +66,7 @@ public class MqttAuthParityTests "127.0.0.1", 0, requiredUsername: "mqtt", requiredPassword: "secret"); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -82,6 +85,7 @@ public class MqttAuthParityTests public async Task No_auth_configured_connects_without_credentials() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -97,6 +101,7 @@ public class MqttAuthParityTests public async Task No_auth_configured_accepts_any_credentials() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -164,6 +169,7 @@ public class MqttAuthParityTests "127.0.0.1", 0, requiredUsername: "admin", requiredPassword: "password"); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -193,6 +199,7 @@ public class MqttAuthParityTests public async Task Keepalive_timeout_disconnects_idle_client() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -279,6 +286,7 @@ public class MqttAuthParityTests public async Task Non_connect_as_first_packet_is_handled() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -300,6 +308,7 @@ public class MqttAuthParityTests public async Task Second_connect_from_same_tcp_connection_is_handled() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttBinaryProtocolTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttBinaryProtocolTests.cs new file mode 100644 index 0000000..df9b2f5 --- /dev/null +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttBinaryProtocolTests.cs @@ -0,0 +1,865 @@ +using System.Buffers; +using System.Net; +using System.Net.Sockets; +using System.Text; +using NATS.Server.Mqtt; + +namespace NATS.Server.Mqtt.Tests; + +/// +/// Tests for the binary MQTT 3.1.1 wire protocol implementation. +/// Covers: TryRead, ParseUnsubscribe, new WriteXxx methods, PipeReader-based +/// connection handling, and MQTT 3.1.1 compliance rules. +/// +public class MqttBinaryProtocolTests +{ + // ----------------------------------------------------------------------- + // MqttPacketReader.TryRead tests + // ----------------------------------------------------------------------- + + [Fact] + public void TryRead_complete_connect_packet_succeeds() + { + // Build a CONNECT packet + var connectPayload = BuildConnectPayload("test-client"); + var raw = MqttPacketWriter.Write(MqttControlPacketType.Connect, connectPayload); + var seq = new ReadOnlySequence(raw); + + MqttPacketReader.TryRead(seq, out var packet, out var consumed).ShouldBeTrue(); + packet.ShouldNotBeNull(); + packet.Type.ShouldBe(MqttControlPacketType.Connect); + seq.GetOffset(consumed).ShouldBe(raw.Length); + } + + [Fact] + public void TryRead_returns_false_on_partial_fixed_header() + { + var seq = new ReadOnlySequence([0x10]); // just first byte, no remaining length + MqttPacketReader.TryRead(seq, out var packet, out _).ShouldBeFalse(); + packet.ShouldBeNull(); + } + + [Fact] + public void TryRead_returns_false_on_partial_payload() + { + // CONNECT with remaining length indicating 10 bytes but only 3 present + var raw = new byte[] { 0x10, 10, 0x00, 0x04, 0x4D }; // truncated + var seq = new ReadOnlySequence(raw); + MqttPacketReader.TryRead(seq, out var packet, out _).ShouldBeFalse(); + packet.ShouldBeNull(); + } + + [Fact] + public void TryRead_handles_multi_byte_remaining_length() + { + // Create a packet with remaining length = 200 (requires 2 bytes to encode) + var payload = new byte[200]; + var raw = MqttPacketWriter.Write(MqttControlPacketType.Publish, payload, flags: 0x00); + var seq = new ReadOnlySequence(raw); + + MqttPacketReader.TryRead(seq, out var packet, out var consumed).ShouldBeTrue(); + packet.ShouldNotBeNull(); + packet.Type.ShouldBe(MqttControlPacketType.Publish); + packet.RemainingLength.ShouldBe(200); + } + + [Fact] + public void TryRead_handles_segmented_sequence() + { + // Simulate a split packet across two segments + var connectPayload = BuildConnectPayload("seg-client"); + var raw = MqttPacketWriter.Write(MqttControlPacketType.Connect, connectPayload); + var mid = raw.Length / 2; + + var first = new ReadOnlyMemory(raw, 0, mid); + var second = new ReadOnlyMemory(raw, mid, raw.Length - mid); + + var firstSegment = new MemorySegment(first); + var lastSegment = firstSegment.Append(second); + var seq = new ReadOnlySequence(firstSegment, 0, lastSegment, second.Length); + + MqttPacketReader.TryRead(seq, out var packet, out _).ShouldBeTrue(); + packet.ShouldNotBeNull(); + packet.Type.ShouldBe(MqttControlPacketType.Connect); + } + + [Fact] + public void TryRead_reads_multiple_packets_from_buffer() + { + var ping = MqttPacketWriter.Write(MqttControlPacketType.PingReq, []); + var combined = new byte[ping.Length * 3]; + ping.CopyTo(combined, 0); + ping.CopyTo(combined, ping.Length); + ping.CopyTo(combined, ping.Length * 2); + + var seq = new ReadOnlySequence(combined); + var count = 0; + + while (MqttPacketReader.TryRead(seq, out var packet, out var consumed)) + { + packet!.Type.ShouldBe(MqttControlPacketType.PingReq); + seq = seq.Slice(consumed); + count++; + } + + count.ShouldBe(3); + } + + [Fact] + public void TryRead_zero_remaining_length_packet() + { + // PINGREQ has 0 remaining length + var raw = MqttPacketWriter.Write(MqttControlPacketType.PingReq, []); + raw.Length.ShouldBe(2); // 1 byte header + 1 byte remaining length (0) + + var seq = new ReadOnlySequence(raw); + MqttPacketReader.TryRead(seq, out var packet, out _).ShouldBeTrue(); + packet!.Type.ShouldBe(MqttControlPacketType.PingReq); + packet.RemainingLength.ShouldBe(0); + } + + // ----------------------------------------------------------------------- + // MqttBinaryDecoder.ParseUnsubscribe tests + // ----------------------------------------------------------------------- + + [Fact] + public void ParseUnsubscribe_single_filter() + { + var payload = new List(); + // Packet ID + payload.Add(0x00); + payload.Add(0x0A); // 10 + // Topic filter + var filter = Encoding.UTF8.GetBytes("sensor/temp"); + payload.Add((byte)(filter.Length >> 8)); + payload.Add((byte)(filter.Length & 0xFF)); + payload.AddRange(filter); + + var result = MqttBinaryDecoder.ParseUnsubscribe([.. payload]); + result.PacketId.ShouldBe((ushort)10); + result.Filters.Count.ShouldBe(1); + result.Filters[0].ShouldBe("sensor/temp"); + } + + [Fact] + public void ParseUnsubscribe_multiple_filters() + { + var payload = new List(); + payload.Add(0x00); + payload.Add(0x01); // Packet ID = 1 + foreach (var topic in new[] { "a/b", "c/d", "e/f" }) + { + var bytes = Encoding.UTF8.GetBytes(topic); + payload.Add((byte)(bytes.Length >> 8)); + payload.Add((byte)(bytes.Length & 0xFF)); + payload.AddRange(bytes); + } + + var result = MqttBinaryDecoder.ParseUnsubscribe([.. payload]); + result.PacketId.ShouldBe((ushort)1); + result.Filters.Count.ShouldBe(3); + result.Filters[0].ShouldBe("a/b"); + result.Filters[1].ShouldBe("c/d"); + result.Filters[2].ShouldBe("e/f"); + } + + [Fact] + public void ParseUnsubscribe_rejects_invalid_flags() + { + var payload = new byte[] { 0x00, 0x01, 0x00, 0x01, (byte)'a' }; + Should.Throw(() => MqttBinaryDecoder.ParseUnsubscribe(payload, flags: 0x00)); + } + + [Fact] + public void ParseUnsubscribe_rejects_empty_filter_list() + { + // Just packet ID, no filters + var payload = new byte[] { 0x00, 0x01 }; + Should.Throw(() => MqttBinaryDecoder.ParseUnsubscribe(payload)); + } + + // ----------------------------------------------------------------------- + // MqttPacketWriter response helper tests + // ----------------------------------------------------------------------- + + [Fact] + public void WriteConnAck_encodes_correctly() + { + var data = MqttPacketWriter.WriteConnAck(0x01, 0x00); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.ConnAck); + packet.RemainingLength.ShouldBe(2); + packet.Payload.Span[0].ShouldBe((byte)0x01); // session present + packet.Payload.Span[1].ShouldBe((byte)0x00); // accepted + } + + [Fact] + public void WritePubAck_round_trips_packet_id() + { + var data = MqttPacketWriter.WritePubAck(42); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.PubAck); + var id = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]); + id.ShouldBe((ushort)42); + } + + [Fact] + public void WriteSubAck_encodes_granted_qos() + { + byte[] grantedQoS = [0, 1, 2, 0x80]; // 0x80 = failure + var data = MqttPacketWriter.WriteSubAck(99, grantedQoS); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.SubAck); + // Packet ID + var id = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]); + id.ShouldBe((ushort)99); + // QoS values + packet.Payload.Span[2].ShouldBe((byte)0); + packet.Payload.Span[3].ShouldBe((byte)1); + packet.Payload.Span[4].ShouldBe((byte)2); + packet.Payload.Span[5].ShouldBe((byte)0x80); + } + + [Fact] + public void WriteUnsubAck_round_trips_packet_id() + { + var data = MqttPacketWriter.WriteUnsubAck(7); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.UnsubAck); + var id = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]); + id.ShouldBe((ushort)7); + } + + [Fact] + public void WritePingResp_is_correct() + { + var data = MqttPacketWriter.WritePingResp(); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.PingResp); + packet.RemainingLength.ShouldBe(0); + } + + [Fact] + public void WritePubRec_round_trips_packet_id() + { + var data = MqttPacketWriter.WritePubRec(100); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.PubRec); + var id = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]); + id.ShouldBe((ushort)100); + } + + [Fact] + public void WritePubRel_has_correct_flags() + { + var data = MqttPacketWriter.WritePubRel(50); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.PubRel); + packet.Flags.ShouldBe((byte)0x02); // PUBREL must have flags 0x02 + } + + [Fact] + public void WritePubComp_round_trips_packet_id() + { + var data = MqttPacketWriter.WritePubComp(200); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.PubComp); + var id = (ushort)((packet.Payload.Span[0] << 8) | packet.Payload.Span[1]); + id.ShouldBe((ushort)200); + } + + [Fact] + public void WritePublish_qos0_no_packet_id() + { + var data = MqttPacketWriter.WritePublish("test/topic", "hello"u8, qos: 0); + var packet = MqttPacketReader.Read(data); + packet.Type.ShouldBe(MqttControlPacketType.Publish); + var pub = MqttBinaryDecoder.ParsePublish(packet.Payload.Span, packet.Flags); + pub.Topic.ShouldBe("test/topic"); + pub.QoS.ShouldBe((byte)0); + pub.PacketId.ShouldBe((ushort)0); + Encoding.UTF8.GetString(pub.Payload.Span).ShouldBe("hello"); + } + + [Fact] + public void WritePublish_qos1_with_flags() + { + var data = MqttPacketWriter.WritePublish("a/b", "data"u8, qos: 1, retain: true, dup: true, packetId: 5); + var packet = MqttPacketReader.Read(data); + var pub = MqttBinaryDecoder.ParsePublish(packet.Payload.Span, packet.Flags); + pub.QoS.ShouldBe((byte)1); + pub.Retain.ShouldBeTrue(); + pub.Dup.ShouldBeTrue(); + pub.PacketId.ShouldBe((ushort)5); + } + + // ----------------------------------------------------------------------- + // Enum completeness + // ----------------------------------------------------------------------- + + [Theory] + [InlineData(MqttControlPacketType.PubRec, 5)] + [InlineData(MqttControlPacketType.PubRel, 6)] + [InlineData(MqttControlPacketType.PubComp, 7)] + [InlineData(MqttControlPacketType.Unsubscribe, 10)] + [InlineData(MqttControlPacketType.UnsubAck, 11)] + public void Enum_has_all_mqtt_packet_types(MqttControlPacketType type, byte expectedValue) + { + ((byte)type).ShouldBe(expectedValue); + } + + // ----------------------------------------------------------------------- + // Binary connection integration tests (MQTT 3.1.1 compliance) + // ----------------------------------------------------------------------- + + [Fact] + public async Task Binary_connect_and_ping_pong() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + + // Send CONNECT + await SendMqttPacketAsync(stream, BuildConnectPacket("ping-client")); + + // Read CONNACK + var connAck = await ReadMqttPacketAsync(stream); + connAck.Type.ShouldBe(MqttControlPacketType.ConnAck); + connAck.Payload.Span[1].ShouldBe(MqttProtocolConstants.ConnAckAccepted); + + // Send PINGREQ + await SendMqttPacketAsync(stream, MqttPacketWriter.Write(MqttControlPacketType.PingReq, [])); + + // Read PINGRESP + var pingResp = await ReadMqttPacketAsync(stream); + pingResp.Type.ShouldBe(MqttControlPacketType.PingResp); + } + + [Fact] + public async Task Binary_first_packet_must_be_connect() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + + // Send PINGREQ as first packet (not CONNECT) — should be disconnected + await SendMqttPacketAsync(stream, MqttPacketWriter.Write(MqttControlPacketType.PingReq, [])); + + // Connection should be closed + var response = await ReadWithTimeoutAsync(stream, 500); + response.ShouldBeNull(); + } + + [Fact] + public async Task Binary_reject_bad_protocol_level() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + + // CONNECT with protocol level 5 (not 4) + var connectPayload = BuildConnectPayload("bad-level", protocolLevel: 5); + await SendMqttPacketAsync(stream, MqttPacketWriter.Write(MqttControlPacketType.Connect, connectPayload)); + + var connAck = await ReadMqttPacketAsync(stream); + connAck.Type.ShouldBe(MqttControlPacketType.ConnAck); + connAck.Payload.Span[1].ShouldBe(MqttProtocolConstants.ConnAckUnacceptableProtocolVersion); + } + + [Fact] + public async Task Binary_empty_clientid_clean_session_generates_id() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + + // CONNECT with empty client ID + clean session + var connectPayload = BuildConnectPayload("", cleanSession: true); + await SendMqttPacketAsync(stream, MqttPacketWriter.Write(MqttControlPacketType.Connect, connectPayload)); + + var connAck = await ReadMqttPacketAsync(stream); + connAck.Payload.Span[1].ShouldBe(MqttProtocolConstants.ConnAckAccepted); + } + + [Fact] + public async Task Binary_empty_clientid_persistent_session_rejected() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + + // CONNECT with empty client ID + persistent session + var connectPayload = BuildConnectPayload("", cleanSession: false); + await SendMqttPacketAsync(stream, MqttPacketWriter.Write(MqttControlPacketType.Connect, connectPayload)); + + var connAck = await ReadMqttPacketAsync(stream); + connAck.Payload.Span[1].ShouldBe(MqttProtocolConstants.ConnAckIdentifierRejected); + } + + [Fact] + public async Task Binary_auth_failure_returns_not_authorized() + { + await using var listener = new MqttListener("127.0.0.1", 0, + requiredUsername: "admin", requiredPassword: "pass"); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + + // CONNECT with wrong credentials + var connectPayload = BuildConnectPayload("auth-fail", username: "wrong", password: "creds"); + await SendMqttPacketAsync(stream, MqttPacketWriter.Write(MqttControlPacketType.Connect, connectPayload)); + + var connAck = await ReadMqttPacketAsync(stream); + connAck.Payload.Span[1].ShouldBe(MqttProtocolConstants.ConnAckNotAuthorized); + } + + [Fact] + public async Task Binary_auth_success_with_credentials() + { + await using var listener = new MqttListener("127.0.0.1", 0, + requiredUsername: "admin", requiredPassword: "secret"); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + + var connectPayload = BuildConnectPayload("auth-ok", username: "admin", password: "secret"); + await SendMqttPacketAsync(stream, MqttPacketWriter.Write(MqttControlPacketType.Connect, connectPayload)); + + var connAck = await ReadMqttPacketAsync(stream); + connAck.Payload.Span[1].ShouldBe(MqttProtocolConstants.ConnAckAccepted); + } + + [Fact] + public async Task Binary_subscribe_and_publish_qos0() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + // Subscriber + using var subTcp = new TcpClient(); + await subTcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var subStream = subTcp.GetStream(); + await ConnectAsync(subStream, "sub-client"); + + // Subscribe to "test/topic" + await SendMqttPacketAsync(subStream, BuildSubscribePacket(1, "test/topic", 0)); + var subAck = await ReadMqttPacketAsync(subStream); + subAck.Type.ShouldBe(MqttControlPacketType.SubAck); + + // Publisher + using var pubTcp = new TcpClient(); + await pubTcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var pubStream = pubTcp.GetStream(); + await ConnectAsync(pubStream, "pub-client"); + + // Publish to "test/topic" + await SendMqttPacketAsync(pubStream, + MqttPacketWriter.WritePublish("test/topic", "hello binary"u8)); + + // Subscriber should receive PUBLISH + var received = await ReadMqttPacketAsync(subStream); + received.Type.ShouldBe(MqttControlPacketType.Publish); + var pub = MqttBinaryDecoder.ParsePublish(received.Payload.Span, received.Flags); + pub.Topic.ShouldBe("test/topic"); + Encoding.UTF8.GetString(pub.Payload.Span).ShouldBe("hello binary"); + } + + [Fact] + public async Task Binary_publish_qos1_gets_puback() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + await ConnectAsync(stream, "qos1-pub"); + + // Publish QoS 1 + await SendMqttPacketAsync(stream, + MqttPacketWriter.WritePublish("qos1/topic", "msg"u8, qos: 1, packetId: 42)); + + var pubAck = await ReadMqttPacketAsync(stream); + pubAck.Type.ShouldBe(MqttControlPacketType.PubAck); + var id = (ushort)((pubAck.Payload.Span[0] << 8) | pubAck.Payload.Span[1]); + id.ShouldBe((ushort)42); + } + + [Fact] + public async Task Binary_publish_qos2_full_flow() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + await ConnectAsync(stream, "qos2-pub"); + + // Step 1: PUBLISH QoS 2 + await SendMqttPacketAsync(stream, + MqttPacketWriter.WritePublish("qos2/topic", "msg"u8, qos: 2, packetId: 10)); + + // Step 2: Receive PUBREC + var pubRec = await ReadMqttPacketAsync(stream); + pubRec.Type.ShouldBe(MqttControlPacketType.PubRec); + + // Step 3: Send PUBREL + await SendMqttPacketAsync(stream, MqttPacketWriter.WritePubRel(10)); + + // Step 4: Receive PUBCOMP + var pubComp = await ReadMqttPacketAsync(stream); + pubComp.Type.ShouldBe(MqttControlPacketType.PubComp); + var id = (ushort)((pubComp.Payload.Span[0] << 8) | pubComp.Payload.Span[1]); + id.ShouldBe((ushort)10); + } + + [Fact] + public async Task Binary_unsubscribe_returns_unsuback() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + await ConnectAsync(stream, "unsub-client"); + + // Subscribe + await SendMqttPacketAsync(stream, BuildSubscribePacket(1, "test/unsub", 0)); + _ = await ReadMqttPacketAsync(stream); // SUBACK + + // Unsubscribe + await SendMqttPacketAsync(stream, BuildUnsubscribePacket(2, "test/unsub")); + var unsubAck = await ReadMqttPacketAsync(stream); + unsubAck.Type.ShouldBe(MqttControlPacketType.UnsubAck); + var id = (ushort)((unsubAck.Payload.Span[0] << 8) | unsubAck.Payload.Span[1]); + id.ShouldBe((ushort)2); + } + + [Fact] + public async Task Binary_unsubscribe_stops_message_delivery() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + // Subscriber + using var subTcp = new TcpClient(); + await subTcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var subStream = subTcp.GetStream(); + await ConnectAsync(subStream, "unsub-recv"); + + await SendMqttPacketAsync(subStream, BuildSubscribePacket(1, "nosub/topic", 0)); + _ = await ReadMqttPacketAsync(subStream); // SUBACK + + // Unsubscribe + await SendMqttPacketAsync(subStream, BuildUnsubscribePacket(2, "nosub/topic")); + _ = await ReadMqttPacketAsync(subStream); // UNSUBACK + + // Publisher + using var pubTcp = new TcpClient(); + await pubTcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var pubStream = pubTcp.GetStream(); + await ConnectAsync(pubStream, "unsub-pub"); + + await SendMqttPacketAsync(pubStream, + MqttPacketWriter.WritePublish("nosub/topic", "invisible"u8)); + + // Subscriber should NOT receive anything + var result = await ReadWithTimeoutAsync(subStream, 200); + result.ShouldBeNull(); + } + + [Fact] + public async Task Binary_disconnect_clears_will_message() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + // Subscriber for will topic + using var subTcp = new TcpClient(); + await subTcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var subStream = subTcp.GetStream(); + await ConnectAsync(subStream, "will-sub"); + await SendMqttPacketAsync(subStream, BuildSubscribePacket(1, "will/topic", 0)); + _ = await ReadMqttPacketAsync(subStream); // SUBACK + + // Client with will + using var willTcp = new TcpClient(); + await willTcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var willStream = willTcp.GetStream(); + var connectPayload = BuildConnectPayload("will-client", + willTopic: "will/topic", willMessage: "oops"); + await SendMqttPacketAsync(willStream, MqttPacketWriter.Write(MqttControlPacketType.Connect, connectPayload)); + _ = await ReadMqttPacketAsync(willStream); // CONNACK + + // Clean DISCONNECT — should clear will + await SendMqttPacketAsync(willStream, + MqttPacketWriter.Write(MqttControlPacketType.Disconnect, [])); + + // Wait a bit and check that will was NOT published + var result = await ReadWithTimeoutAsync(subStream, 300); + result.ShouldBeNull(); + } + + [Fact] + public async Task Binary_duplicate_clientid_takeover() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + // First connection + using var tcp1 = new TcpClient(); + await tcp1.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream1 = tcp1.GetStream(); + await ConnectAsync(stream1, "dup-client"); + + // Second connection with same client-id (takeover) + using var tcp2 = new TcpClient(); + await tcp2.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream2 = tcp2.GetStream(); + await ConnectAsync(stream2, "dup-client"); + + // First connection should be closed + var result = await ReadWithTimeoutAsync(stream1, 500); + result.ShouldBeNull(); + + // Second connection should still work (PINGREQ/PINGRESP) + await SendMqttPacketAsync(stream2, MqttPacketWriter.Write(MqttControlPacketType.PingReq, [])); + var pingResp = await ReadMqttPacketAsync(stream2); + pingResp.Type.ShouldBe(MqttControlPacketType.PingResp); + } + + [Fact] + public async Task Binary_subscribe_flags_validation() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + await ConnectAsync(stream, "bad-sub-flags"); + + // Send SUBSCRIBE with wrong flags (0x00 instead of 0x02) + var subPayload = BuildSubscribePayload(1, "test/topic", 0); + var badPacket = MqttPacketWriter.Write(MqttControlPacketType.Subscribe, subPayload, flags: 0x00); + await SendMqttPacketAsync(stream, badPacket); + + // Connection should be closed + var result = await ReadWithTimeoutAsync(stream, 500); + result.ShouldBeNull(); + } + + [Fact] + public async Task Binary_retained_message_tombstone() + { + await using var listener = new MqttListener("127.0.0.1", 0); + await listener.StartAsync(CancellationToken.None); + + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, listener.Port); + var stream = tcp.GetStream(); + await ConnectAsync(stream, "retain-client"); + + // Publish retained message + await SendMqttPacketAsync(stream, + MqttPacketWriter.WritePublish("retain/topic", "kept"u8, retain: true)); + + // Wait for the server to process the retained publish + for (var i = 0; i < 20; i++) + { + if (listener.GetRetainedMessage("retain/topic") != null) + break; + await Task.Delay(25); + } + + // Verify retained + listener.GetRetainedMessage("retain/topic").ShouldBe("kept"); + + // Publish empty retained (tombstone) + await SendMqttPacketAsync(stream, + MqttPacketWriter.WritePublish("retain/topic", ReadOnlySpan.Empty, retain: true)); + + // Wait for the server to process the packet + for (var i = 0; i < 20; i++) + { + if (listener.GetRetainedMessage("retain/topic") == null) + break; + await Task.Delay(25); + } + + // Verify tombstoned + listener.GetRetainedMessage("retain/topic").ShouldBeNull(); + } + + // ----------------------------------------------------------------------- + // Helpers + // ----------------------------------------------------------------------- + + private static async Task ConnectAsync(NetworkStream stream, string clientId) + { + await SendMqttPacketAsync(stream, BuildConnectPacket(clientId)); + var connAck = await ReadMqttPacketAsync(stream); + connAck.Type.ShouldBe(MqttControlPacketType.ConnAck); + connAck.Payload.Span[1].ShouldBe(MqttProtocolConstants.ConnAckAccepted); + } + + private static byte[] BuildConnectPacket(string clientId, string? username = null, string? password = null, + bool cleanSession = true, byte protocolLevel = 4, string? willTopic = null, string? willMessage = null) + { + var payload = BuildConnectPayload(clientId, username, password, cleanSession, protocolLevel, willTopic, willMessage); + return MqttPacketWriter.Write(MqttControlPacketType.Connect, payload); + } + + private static byte[] BuildConnectPayload(string clientId, string? username = null, string? password = null, + bool cleanSession = true, byte protocolLevel = 4, string? willTopic = null, string? willMessage = null) + { + var buf = new List(); + + // Protocol name "MQTT" + buf.AddRange(MqttPacketWriter.WriteString("MQTT")); + + // Protocol level + buf.Add(protocolLevel); + + // Connect flags + byte flags = 0; + if (cleanSession) flags |= 0x02; + if (username != null) flags |= 0x80; + if (password != null) flags |= 0x40; + if (willTopic != null) + { + flags |= 0x04; // will flag + // will QoS = 0, will retain = 0 + } + buf.Add(flags); + + // Keep-alive (60 seconds) + buf.Add(0x00); + buf.Add(0x3C); + + // Client ID + buf.AddRange(MqttPacketWriter.WriteString(clientId)); + + // Will topic + message + if (willTopic != null) + { + buf.AddRange(MqttPacketWriter.WriteString(willTopic)); + buf.AddRange(MqttPacketWriter.WriteBytes( + Encoding.UTF8.GetBytes(willMessage ?? ""))); + } + + // Username + if (username != null) + buf.AddRange(MqttPacketWriter.WriteString(username)); + + // Password + if (password != null) + buf.AddRange(MqttPacketWriter.WriteString(password)); + + return [.. buf]; + } + + private static byte[] BuildSubscribePacket(ushort packetId, string topic, byte qos) + { + var payload = BuildSubscribePayload(packetId, topic, qos); + return MqttPacketWriter.Write(MqttControlPacketType.Subscribe, payload, flags: 0x02); + } + + private static byte[] BuildSubscribePayload(ushort packetId, string topic, byte qos) + { + var buf = new List(); + buf.Add((byte)(packetId >> 8)); + buf.Add((byte)(packetId & 0xFF)); + buf.AddRange(MqttPacketWriter.WriteString(topic)); + buf.Add(qos); + return [.. buf]; + } + + private static byte[] BuildUnsubscribePacket(ushort packetId, string topic) + { + var buf = new List(); + buf.Add((byte)(packetId >> 8)); + buf.Add((byte)(packetId & 0xFF)); + buf.AddRange(MqttPacketWriter.WriteString(topic)); + return MqttPacketWriter.Write(MqttControlPacketType.Unsubscribe, [.. buf], flags: 0x02); + } + + private static async Task SendMqttPacketAsync(NetworkStream stream, byte[] packet) + { + await stream.WriteAsync(packet); + await stream.FlushAsync(); + } + + private static async Task ReadMqttPacketAsync(NetworkStream stream, int timeoutMs = 2000) + { + using var cts = new CancellationTokenSource(timeoutMs); + var buf = new byte[4096]; + var offset = 0; + + while (true) + { + var read = await stream.ReadAsync(buf.AsMemory(offset), cts.Token); + if (read == 0) + throw new IOException("Connection closed while reading MQTT packet"); + offset += read; + + var seq = new ReadOnlySequence(buf.AsMemory(0, offset)); + if (MqttPacketReader.TryRead(seq, out var packet, out _)) + return packet!; + } + } + + private static async Task ReadWithTimeoutAsync(NetworkStream stream, int timeoutMs) + { + try + { + return await ReadMqttPacketAsync(stream, timeoutMs); + } + catch + { + return null; + } + } + + /// + /// Helper for creating segmented ReadOnlySequence for split-packet tests. + /// + private sealed class MemorySegment : ReadOnlySequenceSegment + { + public MemorySegment(ReadOnlyMemory memory) + { + Memory = memory; + } + + public MemorySegment Append(ReadOnlyMemory memory) + { + var segment = new MemorySegment(memory) + { + RunningIndex = RunningIndex + Memory.Length, + }; + Next = segment; + return segment; + } + } +} diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttCrossProtocolTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttCrossProtocolTests.cs new file mode 100644 index 0000000..46f11e1 --- /dev/null +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttCrossProtocolTests.cs @@ -0,0 +1,164 @@ +using NATS.Server.Auth; +using NATS.Server.Mqtt; +using NATS.Server.Subscriptions; + +namespace NATS.Server.Mqtt.Tests; + +/// +/// Tests for the MqttNatsClientAdapter and cross-protocol bridging concepts. +/// Verifies that MQTT connections can participate in the NATS SubList and +/// that topic/subject translation works end-to-end. +/// +public class MqttCrossProtocolTests +{ + [Fact] + public void Adapter_implements_INatsClient() + { + using var stream = new MemoryStream(); + var listener = CreateTestListener(); + var connection = new MqttConnection(stream, listener); + var adapter = new MqttNatsClientAdapter(connection, 42); + + adapter.Id.ShouldBe((ulong)42); + adapter.Kind.ShouldBe(ClientKind.Client); + adapter.ClientOpts.ShouldBeNull(); + } + + [Fact] + public void Adapter_add_and_remove_subscription() + { + using var stream = new MemoryStream(); + var listener = CreateTestListener(); + var connection = new MqttConnection(stream, listener); + var adapter = new MqttNatsClientAdapter(connection, 1); + var account = new Account("test"); + adapter.Account = account; + + // Add subscription + var sub = adapter.AddSubscription("sensor.temp", "sid1"); + sub.Subject.ShouldBe("sensor.temp"); + sub.Client.ShouldBe(adapter); + adapter.Subscriptions.Count.ShouldBe(1); + + // Verify it's in the SubList + var result = account.SubList.Match("sensor.temp"); + result.PlainSubs.ShouldContain(s => s.Sid == "sid1"); + + // Remove subscription + adapter.RemoveSubscription("sid1"); + adapter.Subscriptions.Count.ShouldBe(0); + + // Verify removed from SubList + result = account.SubList.Match("sensor.temp"); + result.PlainSubs.ShouldNotContain(s => s.Sid == "sid1"); + } + + [Fact] + public void Adapter_remove_all_subscriptions() + { + using var stream = new MemoryStream(); + var listener = CreateTestListener(); + var connection = new MqttConnection(stream, listener); + var adapter = new MqttNatsClientAdapter(connection, 1); + var account = new Account("test"); + adapter.Account = account; + + adapter.AddSubscription("a.b", "s1"); + adapter.AddSubscription("c.d", "s2"); + adapter.AddSubscription("e.f", "s3"); + adapter.Subscriptions.Count.ShouldBe(3); + + adapter.RemoveAllSubscriptions(); + adapter.Subscriptions.Count.ShouldBe(0); + } + + [Fact] + public void Adapter_queue_outbound_is_noop() + { + using var stream = new MemoryStream(); + var listener = CreateTestListener(); + var connection = new MqttConnection(stream, listener); + var adapter = new MqttNatsClientAdapter(connection, 1); + + adapter.QueueOutbound(new byte[] { 1, 2, 3 }).ShouldBeTrue(); + } + + [Fact] + public void Adapter_signal_flush_is_noop() + { + using var stream = new MemoryStream(); + var listener = CreateTestListener(); + var connection = new MqttConnection(stream, listener); + var adapter = new MqttNatsClientAdapter(connection, 1); + + // Should not throw + adapter.SignalFlush(); + } + + [Fact] + public void Topic_mapper_integration_with_sublist() + { + var account = new Account("test"); + + // Simulate an MQTT client subscribing to "sensor/+" + var natsSubject = MqttTopicMapper.MqttToNats("sensor/+"); + natsSubject.ShouldBe("sensor.*"); + + var sub = new Subscription + { + Subject = natsSubject, + Sid = "mqtt-sub-1", + }; + account.SubList.Insert(sub); + + // Simulate a NATS publish to "sensor.temp" — should match + var result = account.SubList.Match("sensor.temp"); + result.PlainSubs.ShouldContain(s => s.Sid == "mqtt-sub-1"); + + // "sensor.humidity" should also match + result = account.SubList.Match("sensor.humidity"); + result.PlainSubs.ShouldContain(s => s.Sid == "mqtt-sub-1"); + + // "other.temp" should NOT match + result = account.SubList.Match("other.temp"); + result.PlainSubs.ShouldNotContain(s => s.Sid == "mqtt-sub-1"); + } + + [Fact] + public void Topic_mapper_multilevel_wildcard_with_sublist() + { + var account = new Account("test"); + + // MQTT subscribe to "home/#" + var natsSubject = MqttTopicMapper.MqttToNats("home/#"); + natsSubject.ShouldBe("home.>"); + + var sub = new Subscription + { + Subject = natsSubject, + Sid = "mqtt-sub-2", + }; + account.SubList.Insert(sub); + + // Should match multi-level subjects + account.SubList.Match("home.living.light").PlainSubs + .ShouldContain(s => s.Sid == "mqtt-sub-2"); + account.SubList.Match("home.kitchen").PlainSubs + .ShouldContain(s => s.Sid == "mqtt-sub-2"); + } + + [Fact] + public void Adapter_mqtt_client_id_exposed() + { + using var stream = new MemoryStream(); + var listener = CreateTestListener(); + var connection = new MqttConnection(stream, listener); + var adapter = new MqttNatsClientAdapter(connection, 1); + + // ClientId comes from the underlying connection + adapter.MqttClientId.ShouldBe(string.Empty); // not yet connected + } + + private static MqttListener CreateTestListener() + => new("127.0.0.1", 0); +} diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttJetStreamPersistenceTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttJetStreamPersistenceTests.cs new file mode 100644 index 0000000..59bf2ba --- /dev/null +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttJetStreamPersistenceTests.cs @@ -0,0 +1,339 @@ +using NATS.Server.JetStream; +using NATS.Server.JetStream.Models; +using NATS.Server.JetStream.Storage; +using NATS.Server.Mqtt; +// Retained/Session store tests use MemStore + StreamConfig directly + +namespace NATS.Server.Mqtt.Tests; + +/// +/// Tests for MQTT JetStream persistence: stream initialization, consumer management, +/// QoS 1/2 flow with JetStream backing, session persistence, and retained message persistence. +/// Go reference: server/mqtt.go mqttCreateAccountSessionManager, mqttStoreSession, +/// mqttHandleRetainedMsg, trackPublish. +/// +public class MqttJetStreamPersistenceTests +{ + // ----------------------------------------------------------------------- + // MqttStreamInitializer + // ----------------------------------------------------------------------- + + [Fact] + public void StreamInitializer_creates_all_five_streams() + { + // Go reference: server/mqtt.go mqttCreateAccountSessionManager creates 5 streams + var (streamMgr, _, initializer) = CreateJetStreamInfra(); + + initializer.IsInitialized.ShouldBeFalse(); + initializer.EnsureStreams(); + initializer.IsInitialized.ShouldBeTrue(); + + streamMgr.Exists(MqttProtocolConstants.SessStreamName).ShouldBeTrue(); + streamMgr.Exists(MqttProtocolConstants.StreamName).ShouldBeTrue(); + streamMgr.Exists(MqttProtocolConstants.RetainedMsgsStreamName).ShouldBeTrue(); + streamMgr.Exists(MqttProtocolConstants.QoS2IncomingMsgsStreamName).ShouldBeTrue(); + streamMgr.Exists(MqttProtocolConstants.OutStreamName).ShouldBeTrue(); + } + + [Fact] + public void StreamInitializer_is_idempotent() + { + var (streamMgr, _, initializer) = CreateJetStreamInfra(); + + initializer.EnsureStreams(); + initializer.EnsureStreams(); // should not throw + + streamMgr.StreamNames.Count.ShouldBe(5); + } + + // ----------------------------------------------------------------------- + // MqttConsumerManager — subscription consumers + // ----------------------------------------------------------------------- + + [Fact] + public void ConsumerManager_creates_subscription_consumer() + { + // Go reference: server/mqtt.go mqttProcessSub — creates durable consumer per QoS>0 sub + var (streamMgr, consumerMgr, initializer) = CreateJetStreamInfra(); + initializer.EnsureStreams(); + var mqttConsumerMgr = new MqttConsumerManager(streamMgr, consumerMgr); + + var binding = mqttConsumerMgr.CreateSubscriptionConsumer("client1", "sensor.temp", qos: 1, maxAckPending: 100); + + binding.ShouldNotBeNull(); + binding.Stream.ShouldBe(MqttProtocolConstants.StreamName); + binding.FilterSubject.ShouldBe($"{MqttProtocolConstants.StreamSubjectPrefix}sensor.temp"); + } + + [Fact] + public void ConsumerManager_removes_subscription_consumer() + { + var (streamMgr, consumerMgr, initializer) = CreateJetStreamInfra(); + initializer.EnsureStreams(); + var mqttConsumerMgr = new MqttConsumerManager(streamMgr, consumerMgr); + + mqttConsumerMgr.CreateSubscriptionConsumer("client1", "sensor.temp", qos: 1, maxAckPending: 100); + mqttConsumerMgr.GetBinding("client1", "sensor.temp").ShouldNotBeNull(); + + mqttConsumerMgr.RemoveSubscriptionConsumer("client1", "sensor.temp"); + mqttConsumerMgr.GetBinding("client1", "sensor.temp").ShouldBeNull(); + } + + [Fact] + public void ConsumerManager_removes_all_consumers_for_client() + { + // Go reference: clean session disconnect removes all consumers + var (streamMgr, consumerMgr, initializer) = CreateJetStreamInfra(); + initializer.EnsureStreams(); + var mqttConsumerMgr = new MqttConsumerManager(streamMgr, consumerMgr); + + mqttConsumerMgr.CreateSubscriptionConsumer("client1", "sensor.temp", qos: 1, maxAckPending: 100); + mqttConsumerMgr.CreateSubscriptionConsumer("client1", "sensor.humidity", qos: 1, maxAckPending: 100); + mqttConsumerMgr.GetClientBindings("client1").Count.ShouldBe(2); + + mqttConsumerMgr.RemoveAllConsumers("client1"); + mqttConsumerMgr.GetClientBindings("client1").Count.ShouldBe(0); + } + + // ----------------------------------------------------------------------- + // QoS 1 with JetStream + // ----------------------------------------------------------------------- + + [Fact] + public async Task QoS1_publish_stores_to_stream() + { + // Go reference: server/mqtt.go QoS 1 publish stores message in $MQTT_msgs + var (streamMgr, consumerMgr, initializer) = CreateJetStreamInfra(); + initializer.EnsureStreams(); + var mqttConsumerMgr = new MqttConsumerManager(streamMgr, consumerMgr); + + var seq = mqttConsumerMgr.PublishToStream("sensor.temp", "72.5"u8.ToArray()); + + seq.ShouldBeGreaterThan((ulong)0); + + // Verify message is in the stream + streamMgr.TryGet(MqttProtocolConstants.StreamName, out var handle).ShouldBeTrue(); + var msg = await handle.Store.LoadAsync(seq, default); + msg.ShouldNotBeNull(); + System.Text.Encoding.UTF8.GetString(msg.Payload.Span).ShouldBe("72.5"); + } + + [Fact] + public async Task QoS1_acknowledge_removes_from_stream() + { + // Go reference: PUBACK acks the JetStream message + var (streamMgr, consumerMgr, initializer) = CreateJetStreamInfra(); + initializer.EnsureStreams(); + var mqttConsumerMgr = new MqttConsumerManager(streamMgr, consumerMgr); + + var seq = mqttConsumerMgr.PublishToStream("sensor.temp", "72.5"u8.ToArray()); + mqttConsumerMgr.AcknowledgeMessage(seq).ShouldBeTrue(); + + // Message should be removed + streamMgr.TryGet(MqttProtocolConstants.StreamName, out var handle).ShouldBeTrue(); + var msg = await handle.Store.LoadAsync(seq, default); + msg.ShouldBeNull(); + } + + [Fact] + public void QoS1_tracker_records_stream_sequence() + { + // Go reference: server/mqtt.go trackPublish — maps packet ID → stream sequence + var tracker = new MqttQoS1Tracker(); + + var packetId = tracker.Register("sensor/temp", "72.5"u8.ToArray(), streamSequence: 42); + + tracker.IsPending(packetId).ShouldBeTrue(); + var acked = tracker.Acknowledge(packetId); + acked.ShouldNotBeNull(); + acked.StreamSequence.ShouldBe((ulong)42); + } + + [Fact] + public void QoS1_tracker_redelivery_preserves_stream_sequence() + { + var tracker = new MqttQoS1Tracker(); + tracker.Register("sensor/temp", "72.5"u8.ToArray(), streamSequence: 99); + + var pending = tracker.GetPendingForRedelivery(); + pending.Count.ShouldBe(1); + pending[0].StreamSequence.ShouldBe((ulong)99); + pending[0].DeliveryCount.ShouldBe(2); // incremented + } + + // ----------------------------------------------------------------------- + // QoS 2 with JetStream + // ----------------------------------------------------------------------- + + [Fact] + public async Task QoS2_incoming_stores_for_dedup() + { + // Go reference: server/mqtt.go QoS 2 incoming stored in $MQTT_qos2in for dedup + var (streamMgr, consumerMgr, initializer) = CreateJetStreamInfra(); + initializer.EnsureStreams(); + var mqttConsumerMgr = new MqttConsumerManager(streamMgr, consumerMgr); + + var seq = mqttConsumerMgr.StoreQoS2Incoming("client1", 1, "payload"u8.ToArray()); + seq.ShouldBeGreaterThan((ulong)0); + + var msg = await mqttConsumerMgr.LoadQoS2IncomingAsync("client1", 1); + msg.ShouldNotBeNull(); + System.Text.Encoding.UTF8.GetString(msg.Payload.Span).ShouldBe("payload"); + } + + [Fact] + public async Task QoS2_incoming_removed_after_pubcomp() + { + // Go reference: server/mqtt.go QoS 2 state removed on PUBCOMP + var (streamMgr, consumerMgr, initializer) = CreateJetStreamInfra(); + initializer.EnsureStreams(); + var mqttConsumerMgr = new MqttConsumerManager(streamMgr, consumerMgr); + + mqttConsumerMgr.StoreQoS2Incoming("client1", 1, "payload"u8.ToArray()); + + var removed = await mqttConsumerMgr.RemoveQoS2IncomingAsync("client1", 1); + removed.ShouldBeTrue(); + + var msg = await mqttConsumerMgr.LoadQoS2IncomingAsync("client1", 1); + msg.ShouldBeNull(); + } + + // ----------------------------------------------------------------------- + // Session persistence with JetStream backing + // ----------------------------------------------------------------------- + + [Fact] + public async Task Session_persists_and_recovers_from_jetstream() + { + // Go reference: server/mqtt.go mqttStoreSession + mqttLoadSession via JetStream + var backingStore = new MemStore(new StreamConfig + { + Name = MqttProtocolConstants.SessStreamName, + Subjects = [$"{MqttProtocolConstants.SessStreamSubjectPrefix}>"], + MaxMsgsPer = 1, + }); + + var store1 = new MqttSessionStore(backingStore); + await store1.ConnectAsync("client-js", cleanSession: false); + store1.AddSubscription("client-js", "topic/a", 1); + store1.AddSubscription("client-js", "topic/b", 0); + await store1.SaveSessionAsync("client-js"); + + // Simulate restart with same backing store + var store2 = new MqttSessionStore(backingStore); + await store2.ConnectAsync("client-js", cleanSession: false); + + var subs = store2.GetSubscriptions("client-js"); + subs.Count.ShouldBe(2); + subs["topic/a"].ShouldBe(1); + subs["topic/b"].ShouldBe(0); + } + + [Fact] + public async Task Clean_session_removes_from_jetstream() + { + var backingStore = new MemStore(new StreamConfig + { + Name = MqttProtocolConstants.SessStreamName, + Subjects = [$"{MqttProtocolConstants.SessStreamSubjectPrefix}>"], + MaxMsgsPer = 1, + }); + + var store = new MqttSessionStore(backingStore); + await store.ConnectAsync("client-clean", cleanSession: false); + store.AddSubscription("client-clean", "topic/x", 1); + await store.SaveSessionAsync("client-clean"); + + // Clean session connect + await store.ConnectAsync("client-clean", cleanSession: true); + + // Simulate restart — should not find session + var store2 = new MqttSessionStore(backingStore); + await store2.ConnectAsync("client-clean", cleanSession: false); + store2.GetSubscriptions("client-clean").ShouldBeEmpty(); + } + + // ----------------------------------------------------------------------- + // Retained messages with JetStream backing + // ----------------------------------------------------------------------- + + [Fact] + public async Task Retained_persists_and_recovers_from_jetstream() + { + // Go reference: server/mqtt.go retained messages stored in $MQTT_rmsgs + var backingStore = new MemStore(new StreamConfig + { + Name = MqttProtocolConstants.RetainedMsgsStreamName, + Subjects = [$"{MqttProtocolConstants.RetainedMsgsStreamSubject}>"], + MaxMsgsPer = 1, + }); + + var retained1 = new MqttRetainedStore(backingStore); + await retained1.SetRetainedAsync("sensors/temp", "72.5"u8.ToArray()); + + // Simulate restart — new store backed by same JetStream + var retained2 = new MqttRetainedStore(backingStore); + var msg = await retained2.GetRetainedAsync("sensors/temp"); + + msg.ShouldNotBeNull(); + System.Text.Encoding.UTF8.GetString(msg).ShouldBe("72.5"); + } + + [Fact] + public async Task Retained_tombstone_removes_from_jetstream() + { + // Go reference: empty payload + retain = delete retained + var backingStore = new MemStore(new StreamConfig + { + Name = MqttProtocolConstants.RetainedMsgsStreamName, + Subjects = [$"{MqttProtocolConstants.RetainedMsgsStreamSubject}>"], + MaxMsgsPer = 1, + }); + + var retained = new MqttRetainedStore(backingStore); + await retained.SetRetainedAsync("sensors/temp", "72.5"u8.ToArray()); + await retained.SetRetainedAsync("sensors/temp", ReadOnlyMemory.Empty); // tombstone + + // Should be gone even from backing store + var recovered = new MqttRetainedStore(backingStore); + var msg = await recovered.GetRetainedAsync("sensors/temp"); + msg.ShouldBeNull(); + } + + // ----------------------------------------------------------------------- + // Flow controller — JetStream integration + // ----------------------------------------------------------------------- + + [Fact] + public async Task FlowController_IsAtCapacity_when_max_reached() + { + // Go reference: server/mqtt.go mqttMaxAckPending flow control + using var fc = new MqttFlowController(defaultMaxAckPending: 2); + + // Acquire 2 slots + (await fc.TryAcquireAsync("sub1")).ShouldBeTrue(); + (await fc.TryAcquireAsync("sub1")).ShouldBeTrue(); + + // Now at capacity + fc.IsAtCapacity("sub1").ShouldBeTrue(); + (await fc.TryAcquireAsync("sub1")).ShouldBeFalse(); + + // Release one + fc.Release("sub1"); + fc.IsAtCapacity("sub1").ShouldBeFalse(); + (await fc.TryAcquireAsync("sub1")).ShouldBeTrue(); + } + + // ----------------------------------------------------------------------- + // Helpers + // ----------------------------------------------------------------------- + + private static (StreamManager StreamMgr, ConsumerManager ConsumerMgr, MqttStreamInitializer Initializer) CreateJetStreamInfra() + { + var consumerMgr = new ConsumerManager(); + var streamMgr = new StreamManager(consumerManager: consumerMgr); + consumerMgr.StreamManager = streamMgr; + var initializer = new MqttStreamInitializer(streamMgr); + return (streamMgr, consumerMgr, initializer); + } +} diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttKeepAliveTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttKeepAliveTests.cs index 81b5805..8139c0c 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttKeepAliveTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttKeepAliveTests.cs @@ -11,6 +11,7 @@ public class MqttKeepAliveTests public async Task Invalid_mqtt_credentials_or_keepalive_timeout_close_session_with_protocol_error() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttListenerParityTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttListenerParityTests.cs index 2caf700..2a3e2bf 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttListenerParityTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttListenerParityTests.cs @@ -11,6 +11,7 @@ public class MqttListenerParityTests public async Task Mqtt_listener_accepts_connect_and_routes_publish_to_matching_subscription() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttPublishSubscribeParityTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttPublishSubscribeParityTests.cs index f8832c9..7b2d80d 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttPublishSubscribeParityTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttPublishSubscribeParityTests.cs @@ -10,6 +10,7 @@ public class MqttPublishSubscribeParityTests public async Task Mqtt_publish_only_reaches_matching_topic_subscribers() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQoSTrackingTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQoSTrackingTests.cs index a4c6fe8..28ea648 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQoSTrackingTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQoSTrackingTests.cs @@ -43,19 +43,19 @@ public sealed class MqttQoSTrackingTests tracker.PendingCount.ShouldBe(1); var removed = tracker.Acknowledge(id); - removed.ShouldBeTrue(); + removed.ShouldNotBeNull(); tracker.PendingCount.ShouldBe(0); } [Fact] - public void Acknowledge_returns_false_for_unknown() + public void Acknowledge_returns_null_for_unknown() { // Go reference: server/mqtt.go — PUBACK for unknown packet ID is silently ignored var tracker = new MqttQoS1Tracker(); var result = tracker.Acknowledge(9999); - result.ShouldBeFalse(); + result.ShouldBeNull(); } [Fact] diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQosAckRuntimeTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQosAckRuntimeTests.cs index 2e09ab9..b570bb3 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQosAckRuntimeTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQosAckRuntimeTests.cs @@ -10,6 +10,7 @@ public class MqttQosAckRuntimeTests public async Task Qos1_publish_receives_puback_and_redelivery_on_session_reconnect_when_unacked() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQosDeliveryParityTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQosDeliveryParityTests.cs index 0aa20c0..348aa61 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQosDeliveryParityTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttQosDeliveryParityTests.cs @@ -15,6 +15,7 @@ public class MqttQosDeliveryParityTests public async Task Qos0_publish_is_fire_and_forget_no_puback_returned() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -37,6 +38,7 @@ public class MqttQosDeliveryParityTests public async Task Qos1_publish_with_subscriber_delivers_message_to_subscriber() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -72,6 +74,7 @@ public class MqttQosDeliveryParityTests public async Task Qos1_publish_without_subscriber_still_returns_puback_to_publisher() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -94,6 +97,7 @@ public class MqttQosDeliveryParityTests public async Task Multiple_qos1_publishes_use_incrementing_packet_ids() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttRetainedMessageParityTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttRetainedMessageParityTests.cs index 7cd911b..568fa67 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttRetainedMessageParityTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttRetainedMessageParityTests.cs @@ -17,6 +17,7 @@ public class MqttRetainedMessageParityTests public async Task Retained_message_not_delivered_when_subscriber_connects_after_publish() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -44,6 +45,7 @@ public class MqttRetainedMessageParityTests public async Task Non_retained_publish_delivers_to_existing_subscriber() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -71,6 +73,7 @@ public class MqttRetainedMessageParityTests public async Task Live_message_delivered_to_existing_subscriber_is_not_flagged_retained() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -98,6 +101,7 @@ public class MqttRetainedMessageParityTests public async Task Multiple_publishers_deliver_to_same_subscriber() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -133,6 +137,7 @@ public class MqttRetainedMessageParityTests public async Task Message_payload_is_not_corrupted_through_broker() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -161,6 +166,7 @@ public class MqttRetainedMessageParityTests public async Task Sequential_publishes_all_deliver() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -190,6 +196,7 @@ public class MqttRetainedMessageParityTests public async Task Multiple_topics_receive_messages_independently() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -230,6 +237,7 @@ public class MqttRetainedMessageParityTests public async Task Subscriber_reconnect_resubscribe_receives_new_messages() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttSessionParityTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttSessionParityTests.cs index 6cdf5a1..de2e88f 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttSessionParityTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttSessionParityTests.cs @@ -17,6 +17,7 @@ public class MqttSessionParityTests public async Task Clean_session_true_discards_previous_session_state() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -49,6 +50,7 @@ public class MqttSessionParityTests public async Task Clean_session_false_preserves_unacked_publishes_across_reconnect() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -80,6 +82,7 @@ public class MqttSessionParityTests public async Task Session_disconnect_cleans_up_client_tracking_on_clean_session() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -109,6 +112,7 @@ public class MqttSessionParityTests public async Task Multiple_concurrent_sessions_on_different_client_ids_work_independently() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttSessionRuntimeTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttSessionRuntimeTests.cs index a14cc51..c842931 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttSessionRuntimeTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttSessionRuntimeTests.cs @@ -11,6 +11,7 @@ public class MqttSessionRuntimeTests public async Task Qos1_publish_receives_puback_and_redelivery_on_session_reconnect_when_unacked() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttTopicMapperTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttTopicMapperTests.cs new file mode 100644 index 0000000..227b9eb --- /dev/null +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttTopicMapperTests.cs @@ -0,0 +1,154 @@ +using NATS.Server.Mqtt; + +namespace NATS.Server.Mqtt.Tests; + +/// +/// Tests for MqttTopicMapper with full Go parity including dots in topics, +/// empty levels, '$' prefix protection, and leading/trailing slashes. +/// Go reference: mqtt.go mqttToNATSSubjectConversion ~line 2200. +/// +public class MqttTopicMapperTests +{ + // ----------------------------------------------------------------------- + // MqttToNats — basic mapping + // ----------------------------------------------------------------------- + + [Theory] + [InlineData("a/b/c", "a.b.c")] + [InlineData("sensor/temp", "sensor.temp")] + [InlineData("home/living/light", "home.living.light")] + public void MqttToNats_separator_mapping(string mqtt, string nats) + { + MqttTopicMapper.MqttToNats(mqtt).ShouldBe(nats); + } + + [Theory] + [InlineData("+", "*")] + [InlineData("sensor/+", "sensor.*")] + [InlineData("+/temp", "*.temp")] + [InlineData("+/+/+", "*.*.*")] + public void MqttToNats_single_level_wildcard(string mqtt, string nats) + { + MqttTopicMapper.MqttToNats(mqtt).ShouldBe(nats); + } + + [Theory] + [InlineData("#", ">")] + [InlineData("sensor/#", "sensor.>")] + [InlineData("home/+/#", "home.*.>")] + public void MqttToNats_multi_level_wildcard(string mqtt, string nats) + { + MqttTopicMapper.MqttToNats(mqtt).ShouldBe(nats); + } + + [Fact] + public void MqttToNats_empty_string() + { + MqttTopicMapper.MqttToNats("").ShouldBe(""); + } + + // ----------------------------------------------------------------------- + // MqttToNats — dot escaping (Go parity) + // ----------------------------------------------------------------------- + + [Theory] + [InlineData("a.b/c", "a_DOT_b.c")] + [InlineData("host.name/metric", "host_DOT_name.metric")] + [InlineData("a.b.c", "a_DOT_b_DOT_c")] + public void MqttToNats_dots_in_topic_are_escaped(string mqtt, string nats) + { + MqttTopicMapper.MqttToNats(mqtt).ShouldBe(nats); + } + + // ----------------------------------------------------------------------- + // MqttToNats — empty levels (leading/trailing/consecutive slashes) + // ----------------------------------------------------------------------- + + [Theory] + [InlineData("/a/b", ".a.b")] + [InlineData("a/b/", "a.b.")] + [InlineData("a//b", "a..b")] + [InlineData("//", "..")] + public void MqttToNats_empty_levels(string mqtt, string nats) + { + MqttTopicMapper.MqttToNats(mqtt).ShouldBe(nats); + } + + // ----------------------------------------------------------------------- + // NatsToMqtt — reverse mapping + // ----------------------------------------------------------------------- + + [Theory] + [InlineData("a.b.c", "a/b/c")] + [InlineData("sensor.temp", "sensor/temp")] + [InlineData("*", "+")] + [InlineData(">", "#")] + [InlineData("sensor.*", "sensor/+")] + [InlineData("sensor.>", "sensor/#")] + public void NatsToMqtt_basic_reverse(string nats, string mqtt) + { + MqttTopicMapper.NatsToMqtt(nats).ShouldBe(mqtt); + } + + [Fact] + public void NatsToMqtt_empty_string() + { + MqttTopicMapper.NatsToMqtt("").ShouldBe(""); + } + + [Theory] + [InlineData("a_DOT_b.c", "a.b/c")] + [InlineData("host_DOT_name.metric", "host.name/metric")] + public void NatsToMqtt_dot_escape_reversed(string nats, string mqtt) + { + MqttTopicMapper.NatsToMqtt(nats).ShouldBe(mqtt); + } + + // ----------------------------------------------------------------------- + // Round-trip: MqttToNats → NatsToMqtt should be identity + // ----------------------------------------------------------------------- + + [Theory] + [InlineData("a/b/c")] + [InlineData("sensor/+/data")] + [InlineData("home/#")] + [InlineData("a.b/c.d")] + [InlineData("/leading")] + [InlineData("trailing/")] + [InlineData("a//b")] + public void RoundTrip_mqtt_to_nats_and_back(string mqtt) + { + var nats = MqttTopicMapper.MqttToNats(mqtt); + var roundTripped = MqttTopicMapper.NatsToMqtt(nats); + roundTripped.ShouldBe(mqtt); + } + + // ----------------------------------------------------------------------- + // Dollar topic protection (MQTT spec [MQTT-4.7.2-1]) + // ----------------------------------------------------------------------- + + [Fact] + public void IsDollarTopic_detects_system_topics() + { + MqttTopicMapper.IsDollarTopic("$SYS/info").ShouldBeTrue(); + MqttTopicMapper.IsDollarTopic("$share/group/topic").ShouldBeTrue(); + MqttTopicMapper.IsDollarTopic("normal/topic").ShouldBeFalse(); + MqttTopicMapper.IsDollarTopic("").ShouldBeFalse(); + } + + [Fact] + public void WildcardMatchesDollarTopic_enforces_spec() + { + // Wildcard filters should NOT match $ topics + MqttTopicMapper.WildcardMatchesDollarTopic("#", "$SYS/info").ShouldBeFalse(); + MqttTopicMapper.WildcardMatchesDollarTopic("+/info", "$SYS/info").ShouldBeFalse(); + + // Explicit $ filters match $ topics + MqttTopicMapper.WildcardMatchesDollarTopic("$SYS/#", "$SYS/info").ShouldBeTrue(); + MqttTopicMapper.WildcardMatchesDollarTopic("$SYS/+", "$SYS/info").ShouldBeTrue(); + + // Non-$ topics always matchable + MqttTopicMapper.WildcardMatchesDollarTopic("#", "normal/topic").ShouldBeTrue(); + MqttTopicMapper.WildcardMatchesDollarTopic("+/topic", "normal/topic").ShouldBeTrue(); + } +} diff --git a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttWillMessageParityTests.cs b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttWillMessageParityTests.cs index 90a970f..4a810e9 100644 --- a/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttWillMessageParityTests.cs +++ b/tests/NATS.Server.Mqtt.Tests/Mqtt/MqttWillMessageParityTests.cs @@ -17,6 +17,7 @@ public class MqttWillMessageParityTests public async Task Subscriber_receives_message_on_abrupt_publisher_disconnect() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -44,6 +45,7 @@ public class MqttWillMessageParityTests public async Task Qos1_will_message_is_delivered_to_subscriber() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -72,6 +74,7 @@ public class MqttWillMessageParityTests public async Task Graceful_disconnect_does_not_deliver_extra_messages() { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -104,6 +107,7 @@ public class MqttWillMessageParityTests public async Task Will_message_at_various_qos_levels_reaches_subscriber(int qos, string payload) { await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token); @@ -205,6 +209,7 @@ public class MqttWillMessageParityTests _ = subQos; await using var listener = new MqttListener("127.0.0.1", 0); + listener.UseBinaryProtocol = false; using var cts = new CancellationTokenSource(); await listener.StartAsync(cts.Token);