natsdotnet/tests/NATS.E2E.Tests/MqttE2ETests.cs

using System.Text;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using NATS.Client.Core;
using NATS.E2E.Tests.Infrastructure;

namespace NATS.E2E.Tests;

/// <summary>
/// End-to-end tests for MQTT 3.1.1 interop using MQTTnet client library.
/// Verifies binary MQTT protocol, cross-protocol MQTT↔NATS messaging, and QoS 1.
/// </summary>
[Collection("E2E-Mqtt")]
public class MqttE2ETests(MqttServerFixture fixture)
{
    [Fact]
    public async Task MqttE2E_ConnectPublishSubscribe()
    {
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        var factory = new MqttFactory();
        using var subscriber = factory.CreateMqttClient();
        using var publisher = factory.CreateMqttClient();

        var subOpts = new MqttClientOptionsBuilder()
            .WithTcpServer("127.0.0.1", fixture.MqttPort)
            .WithClientId("e2e-mqttnet-sub")
            .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
            .Build();

        var pubOpts = new MqttClientOptionsBuilder()
            .WithTcpServer("127.0.0.1", fixture.MqttPort)
            .WithClientId("e2e-mqttnet-pub")
            .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
            .Build();

        await subscriber.ConnectAsync(subOpts, cts.Token);
        await publisher.ConnectAsync(pubOpts, cts.Token);

        var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        subscriber.ApplicationMessageReceivedAsync += e =>
        {
            var payload = Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment);
            received.TrySetResult(payload);
            return Task.CompletedTask;
        };

        await subscriber.SubscribeAsync(
            factory.CreateSubscribeOptionsBuilder()
                .WithTopicFilter("test/mqttnet/e2e")
                .Build(),
            cts.Token);

        // Small delay to let subscription propagate
        await Task.Delay(100, cts.Token);

        await publisher.PublishAsync(
            new MqttApplicationMessageBuilder()
                .WithTopic("test/mqttnet/e2e")
                .WithPayload("hello-mqttnet")
                .Build(),
            cts.Token);

        var msg = await received.Task.WaitAsync(cts.Token);
        msg.ShouldBe("hello-mqttnet");

        await subscriber.DisconnectAsync(cancellationToken: cts.Token);
        await publisher.DisconnectAsync(cancellationToken: cts.Token);
    }

    [Fact]
    public async Task MqttE2E_CrossProtocol_MqttPublish_NatsSubscribe()
    {
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        // NATS subscriber on "sensor.temp" (MQTT topic "sensor/temp" maps to NATS subject "sensor.temp")
        await using var natsConn = fixture.CreateNatsClient();
        await natsConn.ConnectAsync();

        var natsReceived = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        _ = Task.Run(async () =>
        {
            await foreach (var msg in natsConn.SubscribeAsync<string>("sensor.temp", cancellationToken: cts.Token))
            {
                natsReceived.TrySetResult(msg.Data ?? "");
                break;
            }
        }, cts.Token);

        // Small delay to let NATS subscription register
        await Task.Delay(200, cts.Token);

        // MQTT publisher on "sensor/temp"
        var factory = new MqttFactory();
        using var mqttPub = factory.CreateMqttClient();
        var pubOpts = new MqttClientOptionsBuilder()
            .WithTcpServer("127.0.0.1", fixture.MqttPort)
            .WithClientId("e2e-cross-mqtt-pub")
            .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
            .Build();

        await mqttPub.ConnectAsync(pubOpts, cts.Token);
        await mqttPub.PublishAsync(
            new MqttApplicationMessageBuilder()
                .WithTopic("sensor/temp")
                .WithPayload("22.5")
                .Build(),
            cts.Token);

        var result = await natsReceived.Task.WaitAsync(cts.Token);
        result.ShouldBe("22.5");

        await mqttPub.DisconnectAsync(cancellationToken: cts.Token);
    }

    [Fact]
    public async Task MqttE2E_CrossProtocol_NatsPublish_MqttSubscribe()
    {
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        // MQTT subscriber on "sensor/humidity"
        var factory = new MqttFactory();
        using var mqttSub = factory.CreateMqttClient();
        var subOpts = new MqttClientOptionsBuilder()
            .WithTcpServer("127.0.0.1", fixture.MqttPort)
            .WithClientId("e2e-cross-mqtt-sub")
            .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
            .Build();

        await mqttSub.ConnectAsync(subOpts, cts.Token);

        var mqttReceived = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        mqttSub.ApplicationMessageReceivedAsync += e =>
        {
            var payload = Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment);
            mqttReceived.TrySetResult(payload);
            return Task.CompletedTask;
        };

        await mqttSub.SubscribeAsync(
            factory.CreateSubscribeOptionsBuilder()
                .WithTopicFilter("sensor/humidity")
                .Build(),
            cts.Token);

        // Small delay to let subscription propagate through SubList
        await Task.Delay(200, cts.Token);

        // NATS publisher on "sensor.humidity"
        await using var natsConn = fixture.CreateNatsClient();
        await natsConn.ConnectAsync();
        await natsConn.PublishAsync("sensor.humidity", "65%", cancellationToken: cts.Token);

        var result = await mqttReceived.Task.WaitAsync(cts.Token);
        result.ShouldBe("65%");

        await mqttSub.DisconnectAsync(cancellationToken: cts.Token);
    }

    [Fact]
    public async Task MqttE2E_Qos1_PubAck()
    {
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        var factory = new MqttFactory();
        using var subscriber = factory.CreateMqttClient();
        using var publisher = factory.CreateMqttClient();

        var subOpts = new MqttClientOptionsBuilder()
            .WithTcpServer("127.0.0.1", fixture.MqttPort)
            .WithClientId("e2e-qos1-sub")
            .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
            .Build();

        var pubOpts = new MqttClientOptionsBuilder()
            .WithTcpServer("127.0.0.1", fixture.MqttPort)
            .WithClientId("e2e-qos1-pub")
            .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
            .Build();

        await subscriber.ConnectAsync(subOpts, cts.Token);
        await publisher.ConnectAsync(pubOpts, cts.Token);

        var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        subscriber.ApplicationMessageReceivedAsync += e =>
        {
            var payload = Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment);
            received.TrySetResult(payload);
            return Task.CompletedTask;
        };

        await subscriber.SubscribeAsync(
            factory.CreateSubscribeOptionsBuilder()
                .WithTopicFilter(f => f.WithTopic("test/qos1").WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
                .Build(),
            cts.Token);

        await Task.Delay(100, cts.Token);

        // Publish with QoS 1 — MQTTnet will expect PUBACK from server
        var pubResult = await publisher.PublishAsync(
            new MqttApplicationMessageBuilder()
                .WithTopic("test/qos1")
                .WithPayload("qos1-payload")
                .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
                .Build(),
            cts.Token);

        // MQTTnet throws if PUBACK not received, so reaching here means server sent PUBACK
        pubResult.ReasonCode.ShouldBe(MqttClientPublishReasonCode.Success);

        var msg = await received.Task.WaitAsync(cts.Token);
        msg.ShouldBe("qos1-payload");

        await subscriber.DisconnectAsync(cancellationToken: cts.Token);
        await publisher.DisconnectAsync(cancellationToken: cts.Token);
    }
}