test: add E2E MQTT bridge tests (pub/sub, different-topic isolation, QoS 1)
This commit is contained in:
Joseph Doherty
50
tests/NATS.E2E.Tests/Infrastructure/MqttServerFixture.cs
Normal file
50
tests/NATS.E2E.Tests/Infrastructure/MqttServerFixture.cs
Normal file
@@ -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<MqttServerFixture>;
|
||||||
158
tests/NATS.E2E.Tests/MqttTests.cs
Normal file
158
tests/NATS.E2E.Tests/MqttTests.cs
Normal file
@@ -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;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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}
|
||||||
|
/// </summary>
|
||||||
|
[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<string?> ReadLineAsync(NetworkStream stream, CancellationToken ct)
|
||||||
|
{
|
||||||
|
var bytes = new List<byte>(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<TcpClient> 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<string>(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<string>(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");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user