diff --git a/tests/NATS.E2E.Tests/Infrastructure/MqttServerFixture.cs b/tests/NATS.E2E.Tests/Infrastructure/MqttServerFixture.cs new file mode 100644 index 0000000..38f4ed8 --- /dev/null +++ b/tests/NATS.E2E.Tests/Infrastructure/MqttServerFixture.cs @@ -0,0 +1,50 @@ +using NATS.Client.Core; + +namespace NATS.E2E.Tests.Infrastructure; + +public sealed class MqttServerFixture : IAsyncLifetime +{ + private NatsServerProcess _server = null!; + private string _storeDir = null!; + + public int Port => _server.Port; + public int MqttPort { get; private set; } + + public async Task InitializeAsync() + { + MqttPort = NatsServerProcess.AllocateFreePort(); + _storeDir = Path.Combine(Path.GetTempPath(), "nats-e2e-mqtt-" + Guid.NewGuid().ToString("N")[..8]); + Directory.CreateDirectory(_storeDir); + + var config = $$""" + jetstream { + store_dir: "{{_storeDir}}" + max_mem_store: 64mb + max_file_store: 256mb + } + mqtt { + listen: 127.0.0.1:{{MqttPort}} + } + """; + + _server = NatsServerProcess.WithConfig(config); + await _server.StartAsync(); + } + + public async Task DisposeAsync() + { + await _server.DisposeAsync(); + + if (_storeDir is not null && Directory.Exists(_storeDir)) + { + try { Directory.Delete(_storeDir, recursive: true); } + catch (IOException ex) { _ = ex; /* best-effort temp dir cleanup */ } + } + } + + public NatsConnection CreateNatsClient() + => new(new NatsOpts { Url = $"nats://127.0.0.1:{Port}" }); +} + +[CollectionDefinition("E2E-Mqtt")] +public class MqttCollection : ICollectionFixture; diff --git a/tests/NATS.E2E.Tests/MqttTests.cs b/tests/NATS.E2E.Tests/MqttTests.cs new file mode 100644 index 0000000..1cc74cc --- /dev/null +++ b/tests/NATS.E2E.Tests/MqttTests.cs @@ -0,0 +1,158 @@ +using System.Net; +using System.Net.Sockets; +using System.Text; +using NATS.Client.Core; +using NATS.E2E.Tests.Infrastructure; + +namespace NATS.E2E.Tests; + +/// +/// E2E tests for the server's MQTT bridge using the server's line-based MQTT wire protocol. +/// The server exposes a text-framed MQTT protocol (not binary MQTT 3.1.1) on the mqtt port. +/// Wire commands: CONNECT {clientId}, SUB {topic}, PUB {topic} {payload}, PUBQ1 {packetId} {topic} {payload}, ACK {packetId} +/// Wire responses: CONNACK, SUBACK {topic}, MSG {topic} {payload}, PUBACK {packetId}, REDLIVER {packetId} {topic} {payload} +/// +[Collection("E2E-Mqtt")] +public class MqttTests(MqttServerFixture fixture) +{ + // ---- wire helpers (duplicated per E2E convention, no shared TestHelpers) ---- + + private static async Task WriteLineAsync(NetworkStream stream, string line, CancellationToken ct) + { + var bytes = Encoding.UTF8.GetBytes(line + "\n"); + await stream.WriteAsync(bytes, ct); + await stream.FlushAsync(ct); + } + + private static async Task ReadLineAsync(NetworkStream stream, CancellationToken ct) + { + var bytes = new List(64); + var one = new byte[1]; + try + { + while (true) + { + var read = await stream.ReadAsync(one.AsMemory(0, 1), ct); + if (read == 0) + return null; + if (one[0] == (byte)'\n') + break; + if (one[0] != (byte)'\r') + bytes.Add(one[0]); + } + } + catch (OperationCanceledException) + { + return null; + } + + return Encoding.UTF8.GetString([.. bytes]); + } + + private static async Task ConnectMqttClientAsync(int port, string clientId, CancellationToken ct) + { + var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, port, ct); + var stream = tcp.GetStream(); + await WriteLineAsync(stream, $"CONNECT {clientId}", ct); + var connAck = await ReadLineAsync(stream, ct); + connAck.ShouldBe("CONNACK"); + return tcp; + } + + // ---- tests ---- + + [Fact] + public async Task Mqtt_PubSub_SameTopicDelivered() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + + // Subscriber connects and subscribes to topic + using var subTcp = await ConnectMqttClientAsync(fixture.MqttPort, "e2e-sub-1", cts.Token); + var subStream = subTcp.GetStream(); + await WriteLineAsync(subStream, "SUB e2e/mqtt/test1", cts.Token); + var subAck = await ReadLineAsync(subStream, cts.Token); + subAck.ShouldNotBeNull(); + subAck!.ShouldContain("SUBACK"); + + // Publisher connects and publishes + using var pubTcp = await ConnectMqttClientAsync(fixture.MqttPort, "e2e-pub-1", cts.Token); + var pubStream = pubTcp.GetStream(); + await WriteLineAsync(pubStream, "PUB e2e/mqtt/test1 hello-world", cts.Token); + + // Subscriber receives the message + var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _ = Task.Run(async () => + { + var line = await ReadLineAsync(subStream, cts.Token); + if (line is not null) + received.TrySetResult(line); + else + received.TrySetCanceled(); + }, cts.Token); + + var msg = await received.Task.WaitAsync(cts.Token); + msg.ShouldBe("MSG e2e/mqtt/test1 hello-world"); + } + + [Fact] + public async Task Mqtt_DifferentTopic_NotDelivered() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + + // Subscriber subscribes to topic A + using var subTcp = await ConnectMqttClientAsync(fixture.MqttPort, "e2e-sub-2", cts.Token); + var subStream = subTcp.GetStream(); + await WriteLineAsync(subStream, "SUB e2e/mqtt/topicA", cts.Token); + var subAck = await ReadLineAsync(subStream, cts.Token); + subAck.ShouldNotBeNull(); + subAck!.ShouldContain("SUBACK"); + + // Publisher publishes to topic B (different topic) + using var pubTcp = await ConnectMqttClientAsync(fixture.MqttPort, "e2e-pub-2", cts.Token); + var pubStream = pubTcp.GetStream(); + await WriteLineAsync(pubStream, "PUB e2e/mqtt/topicB unrelated-message", cts.Token); + + // Short-timeout read — subscriber should NOT receive anything + using var shortCts = new CancellationTokenSource(TimeSpan.FromMilliseconds(300)); + var unexpected = await ReadLineAsync(subStream, shortCts.Token); + unexpected.ShouldBeNull(); + } + + [Fact] + public async Task Mqtt_Qos1_PubAckReceived_AndDelivered() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + + // Subscriber connects and subscribes + using var subTcp = await ConnectMqttClientAsync(fixture.MqttPort, "e2e-qos1-sub", cts.Token); + var subStream = subTcp.GetStream(); + await WriteLineAsync(subStream, "SUB e2e/mqtt/qos1", cts.Token); + var subAck = await ReadLineAsync(subStream, cts.Token); + subAck.ShouldNotBeNull(); + subAck!.ShouldContain("SUBACK"); + + // Publisher connects and sends QoS 1 publish (PUBQ1 {packetId} {topic} {payload}) + using var pubTcp = await ConnectMqttClientAsync(fixture.MqttPort, "e2e-qos1-pub", cts.Token); + var pubStream = pubTcp.GetStream(); + await WriteLineAsync(pubStream, "PUBQ1 42 e2e/mqtt/qos1 qos1-message", cts.Token); + + // Publisher receives PUBACK from server + var pubAck = await ReadLineAsync(pubStream, cts.Token); + pubAck.ShouldBe("PUBACK 42"); + + // Subscriber receives the message + var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _ = Task.Run(async () => + { + var line = await ReadLineAsync(subStream, cts.Token); + if (line is not null) + received.TrySetResult(line); + else + received.TrySetCanceled(); + }, cts.Token); + + var msg = await received.Task.WaitAsync(cts.Token); + msg.ShouldBe("MSG e2e/mqtt/qos1 qos1-message"); + } +}