28451ad263
Add MqttClusterTests.cs with 8 cluster tests (T:2215-2224) and relocate T:2225 from ImplBacklog stub. All tests deferred (require running cluster).
1056 lines
44 KiB
C#
1056 lines
44 KiB
C#
// Copyright 2020-2026 The NATS Authors
|
||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||
// you may not use this file except in compliance with the License.
|
||
// You may obtain a copy of the License at
|
||
//
|
||
// http://www.apache.org/licenses/LICENSE-2.0
|
||
//
|
||
// Unless required by applicable law or agreed to in writing, software
|
||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||
// See the License for the specific language governing permissions and
|
||
// limitations under the License.
|
||
//
|
||
// Ported from MQTT cluster tests in server/mqtt_test.go (lines 3014-3809).
|
||
|
||
using System.Buffers.Binary;
|
||
using System.Net.Sockets;
|
||
using System.Text;
|
||
using Shouldly;
|
||
|
||
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Mqtt;
|
||
|
||
/// <summary>
|
||
/// Integration tests for MQTT behavior across JetStream clusters.
|
||
/// Tests cover subscription propagation, session takeover, retained message
|
||
/// replication, replica counts, placement, and leafnode integration.
|
||
///
|
||
/// These tests require a running NATS cluster with MQTT and JetStream enabled.
|
||
/// </summary>
|
||
[Trait("Category", "Integration")]
|
||
public sealed class MqttClusterTests
|
||
{
|
||
// -------------------------------------------------------------------------
|
||
// Constants mirroring Go mqtt.go
|
||
// -------------------------------------------------------------------------
|
||
|
||
private const byte PacketConnect = 0x10;
|
||
private const byte PacketConnAck = 0x20;
|
||
private const byte PacketPub = 0x30;
|
||
private const byte PacketPubAck = 0x40;
|
||
private const byte PacketPubRec = 0x50;
|
||
private const byte PacketPubRel = 0x60;
|
||
private const byte PacketPubComp = 0x70;
|
||
private const byte PacketSub = 0x80;
|
||
private const byte PacketSubAck = 0x90;
|
||
private const byte PacketUnsub = 0xA0;
|
||
private const byte PacketUnsubAck = 0xB0;
|
||
private const byte PacketPing = 0xC0;
|
||
private const byte PacketPingResp = 0xD0;
|
||
private const byte PacketDisconnect = 0xE0;
|
||
private const byte PacketMask = 0xF0;
|
||
private const byte PacketFlagMask = 0x0F;
|
||
|
||
private const byte ConnFlagCleanSession = 0x02;
|
||
private const byte ConnFlagWillFlag = 0x04;
|
||
private const byte ConnFlagWillRetain = 0x20;
|
||
private const byte ConnFlagPasswordFlag = 0x40;
|
||
private const byte ConnFlagUsernameFlag = 0x80;
|
||
|
||
private const byte PubFlagRetain = 0x01;
|
||
private const byte PubFlagQoS = 0x06;
|
||
private const byte PubFlagDup = 0x08;
|
||
private const byte PubQoS1 = 0x1 << 1;
|
||
|
||
private const byte SubscribeFlags = 0x02;
|
||
|
||
private const byte ConnAckAccepted = 0x00;
|
||
|
||
private static readonly byte[] MqttProtoName = Encoding.ASCII.GetBytes("MQTT");
|
||
private const byte MqttProtoLevel = 0x04;
|
||
|
||
private static readonly TimeSpan MqttTimeout = TimeSpan.FromSeconds(10);
|
||
|
||
// -------------------------------------------------------------------------
|
||
// MQTT protocol helpers (duplicated from MqttTests.cs for self-containment)
|
||
// -------------------------------------------------------------------------
|
||
|
||
private sealed class MqttConnection : IDisposable
|
||
{
|
||
public TcpClient Tcp { get; }
|
||
public NetworkStream Stream { get; }
|
||
|
||
public MqttConnection(TcpClient tcp)
|
||
{
|
||
Tcp = tcp;
|
||
Stream = tcp.GetStream();
|
||
}
|
||
|
||
public void Dispose()
|
||
{
|
||
Stream.Dispose();
|
||
Tcp.Dispose();
|
||
}
|
||
}
|
||
|
||
private static MqttConnection MqttConnect(
|
||
string host, int port,
|
||
string? clientId = null,
|
||
bool cleanSess = true,
|
||
string? user = null,
|
||
string? pass = null,
|
||
ushort keepAlive = 0)
|
||
{
|
||
var tcp = new TcpClient();
|
||
tcp.Connect(host, port);
|
||
var conn = new MqttConnection(tcp);
|
||
|
||
var proto = BuildConnectPacket(clientId, cleanSess, user, pass, keepAlive);
|
||
WriteAll(conn.Stream, proto);
|
||
return conn;
|
||
}
|
||
|
||
private static byte[] BuildConnectPacket(
|
||
string? clientId,
|
||
bool cleanSess,
|
||
string? user,
|
||
string? pass,
|
||
ushort keepAlive)
|
||
{
|
||
byte flags = 0;
|
||
if (cleanSess) flags |= ConnFlagCleanSession;
|
||
|
||
var cidBytes = Encoding.UTF8.GetBytes(clientId ?? "");
|
||
var userBytes = user is not null ? Encoding.UTF8.GetBytes(user) : null;
|
||
var passBytes = pass is not null ? Encoding.UTF8.GetBytes(pass) : null;
|
||
|
||
if (userBytes is not null) flags |= ConnFlagUsernameFlag;
|
||
if (passBytes is not null) flags |= ConnFlagPasswordFlag;
|
||
|
||
var pkLen = 2 + MqttProtoName.Length +
|
||
1 + 1 + 2 +
|
||
2 + cidBytes.Length;
|
||
if (userBytes is not null) pkLen += 2 + userBytes.Length;
|
||
if (passBytes is not null) pkLen += 2 + passBytes.Length;
|
||
|
||
var w = new ByteWriter();
|
||
w.WriteByte(PacketConnect);
|
||
w.WriteVarInt(pkLen);
|
||
w.WriteBytes(MqttProtoName);
|
||
w.WriteByte(MqttProtoLevel);
|
||
w.WriteByte(flags);
|
||
w.WriteUInt16(keepAlive);
|
||
w.WriteBytes(cidBytes);
|
||
if (userBytes is not null) w.WriteBytes(userBytes);
|
||
if (passBytes is not null) w.WriteBytes(passBytes);
|
||
return w.ToArray();
|
||
}
|
||
|
||
private static byte[] ReadExact(NetworkStream stream, int n)
|
||
{
|
||
var buf = new byte[n];
|
||
var offset = 0;
|
||
stream.ReadTimeout = (int)MqttTimeout.TotalMilliseconds;
|
||
while (offset < n)
|
||
{
|
||
var read = stream.Read(buf, offset, n - offset);
|
||
if (read == 0) throw new EndOfStreamException("MQTT connection closed");
|
||
offset += read;
|
||
}
|
||
return buf;
|
||
}
|
||
|
||
private static (byte FirstByte, byte[] Payload) ReadPacket(NetworkStream stream)
|
||
{
|
||
stream.ReadTimeout = (int)MqttTimeout.TotalMilliseconds;
|
||
var header = new byte[1];
|
||
var n = stream.Read(header, 0, 1);
|
||
if (n == 0) throw new EndOfStreamException("MQTT connection closed reading packet header");
|
||
|
||
var multiplier = 1;
|
||
var value = 0;
|
||
while (true)
|
||
{
|
||
var b2 = new byte[1];
|
||
if (stream.Read(b2, 0, 1) == 0) throw new EndOfStreamException();
|
||
value += (b2[0] & 0x7F) * multiplier;
|
||
if ((b2[0] & 0x80) == 0) break;
|
||
multiplier *= 128;
|
||
if (multiplier > 0x200000) throw new InvalidOperationException("malformed MQTT variable int");
|
||
}
|
||
|
||
var payload = value > 0 ? ReadExact(stream, value) : [];
|
||
return (header[0], payload);
|
||
}
|
||
|
||
private static void WriteAll(NetworkStream stream, byte[] data)
|
||
{
|
||
stream.WriteTimeout = (int)MqttTimeout.TotalMilliseconds;
|
||
stream.Write(data, 0, data.Length);
|
||
}
|
||
|
||
private static void CheckConnAck(NetworkStream stream, byte expectedRc, bool expectedSessionPresent)
|
||
{
|
||
var (b, payload) = ReadPacket(stream);
|
||
(b & PacketMask).ShouldBe(PacketConnAck, "Expected CONNACK packet");
|
||
payload.Length.ShouldBe(2, "CONNACK payload should be 2 bytes");
|
||
var sessionPresent = (payload[0] & 0x01) == 1;
|
||
sessionPresent.ShouldBe(expectedSessionPresent, "session-present flag mismatch");
|
||
payload[1].ShouldBe(expectedRc, $"CONNACK return code mismatch (got 0x{payload[1]:X2}, expected 0x{expectedRc:X2})");
|
||
}
|
||
|
||
private static void CheckConnAckAccepted(NetworkStream stream, bool expectedSessionPresent = false)
|
||
=> CheckConnAck(stream, ConnAckAccepted, expectedSessionPresent);
|
||
|
||
private static void SendPublish(NetworkStream stream, byte qos, bool dup, bool retain,
|
||
string topic, ushort pi, byte[] payload)
|
||
{
|
||
byte flags = 0;
|
||
if (dup) flags |= PubFlagDup;
|
||
if (retain) flags |= PubFlagRetain;
|
||
flags |= (byte)((qos & 0x03) << 1);
|
||
|
||
var topicBytes = Encoding.UTF8.GetBytes(topic);
|
||
var pkLen = 2 + topicBytes.Length + payload.Length;
|
||
if (qos > 0) pkLen += 2;
|
||
|
||
var w = new ByteWriter();
|
||
w.WriteByte((byte)(PacketPub | flags));
|
||
w.WriteVarInt(pkLen);
|
||
w.WriteBytes(topicBytes);
|
||
if (qos > 0) w.WriteUInt16(pi);
|
||
w.Write(payload);
|
||
WriteAll(stream, w.ToArray());
|
||
}
|
||
|
||
private static void MqttPublish(NetworkStream stream, byte qos, bool dup, bool retain,
|
||
string topic, ushort pi, byte[] payload)
|
||
{
|
||
SendPublish(stream, qos, dup, retain, topic, pi, payload);
|
||
switch (qos)
|
||
{
|
||
case 1:
|
||
{
|
||
var (b, p) = ReadPacket(stream);
|
||
(b & PacketMask).ShouldBe(PacketPubAck, "Expected PUBACK");
|
||
var rpi = BinaryPrimitives.ReadUInt16BigEndian(p.AsSpan(0, 2));
|
||
rpi.ShouldBe(pi, "PUBACK packet identifier mismatch");
|
||
break;
|
||
}
|
||
case 2:
|
||
{
|
||
var (b, p) = ReadPacket(stream);
|
||
(b & PacketMask).ShouldBe(PacketPubRec, "Expected PUBREC");
|
||
var rpi = BinaryPrimitives.ReadUInt16BigEndian(p.AsSpan(0, 2));
|
||
rpi.ShouldBe(pi, "PUBREC packet identifier mismatch");
|
||
|
||
SendPiPacket(stream, PacketPubRel, pi);
|
||
|
||
var (b2, p2) = ReadPacket(stream);
|
||
(b2 & PacketMask).ShouldBe(PacketPubComp, "Expected PUBCOMP");
|
||
var rpi2 = BinaryPrimitives.ReadUInt16BigEndian(p2.AsSpan(0, 2));
|
||
rpi2.ShouldBe(pi, "PUBCOMP packet identifier mismatch");
|
||
|
||
MqttFlush(stream);
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
private static void MqttFlush(NetworkStream stream)
|
||
{
|
||
var w = new ByteWriter();
|
||
w.WriteByte(PacketPing);
|
||
w.WriteByte(0);
|
||
WriteAll(stream, w.ToArray());
|
||
var (b, p) = ReadPacket(stream);
|
||
(b & PacketMask).ShouldBe(PacketPingResp, "Expected PINGRESP");
|
||
p.Length.ShouldBe(0, "PINGRESP payload should be empty");
|
||
}
|
||
|
||
private static void SendPiPacket(NetworkStream stream, byte packetType, ushort pi)
|
||
{
|
||
var w = new ByteWriter();
|
||
w.WriteByte(packetType);
|
||
w.WriteVarInt(2);
|
||
w.WriteUInt16(pi);
|
||
WriteAll(stream, w.ToArray());
|
||
}
|
||
|
||
private static byte[] MqttSubscribe(NetworkStream stream, ushort pi,
|
||
IEnumerable<(string Filter, byte Qos)> filters)
|
||
{
|
||
var filterList = filters.ToList();
|
||
var pkLen = 2;
|
||
foreach (var (f, _) in filterList) pkLen += 2 + Encoding.UTF8.GetByteCount(f) + 1;
|
||
|
||
var w = new ByteWriter();
|
||
w.WriteByte((byte)(PacketSub | SubscribeFlags));
|
||
w.WriteVarInt(pkLen);
|
||
w.WriteUInt16(pi);
|
||
foreach (var (f, q) in filterList)
|
||
{
|
||
w.WriteBytes(Encoding.UTF8.GetBytes(f));
|
||
w.WriteByte(q);
|
||
}
|
||
WriteAll(stream, w.ToArray());
|
||
|
||
var (b, payload) = ReadPacket(stream);
|
||
(b & PacketMask).ShouldBe(PacketSubAck, "Expected SUBACK");
|
||
var rpi = BinaryPrimitives.ReadUInt16BigEndian(payload.AsSpan(0, 2));
|
||
rpi.ShouldBe(pi, "SUBACK packet identifier mismatch");
|
||
return payload[2..];
|
||
}
|
||
|
||
private static (byte Flags, ushort Pi, string Topic, byte[] Payload) ReadPublish(
|
||
NetworkStream stream)
|
||
{
|
||
var (b, raw) = ReadPacket(stream);
|
||
(b & PacketMask).ShouldBe(PacketPub, "Expected PUBLISH packet");
|
||
var flags = (byte)(b & PacketFlagMask);
|
||
var qos = (byte)((flags & PubFlagQoS) >> 1);
|
||
|
||
var pos = 0;
|
||
var topicLen = BinaryPrimitives.ReadUInt16BigEndian(raw.AsSpan(pos, 2)); pos += 2;
|
||
var topic = Encoding.UTF8.GetString(raw, pos, topicLen); pos += topicLen;
|
||
ushort pi = 0;
|
||
if (qos > 0) { pi = BinaryPrimitives.ReadUInt16BigEndian(raw.AsSpan(pos, 2)); pos += 2; }
|
||
var payload = raw[pos..];
|
||
return (flags, pi, topic, payload);
|
||
}
|
||
|
||
private static void ExpectNothing(NetworkStream stream)
|
||
{
|
||
stream.ReadTimeout = 150;
|
||
var buf = new byte[128];
|
||
try
|
||
{
|
||
var n = stream.Read(buf, 0, buf.Length);
|
||
n.ShouldBe(0, "Expected no data from MQTT server");
|
||
}
|
||
catch (IOException ex) when (ex.InnerException is SocketException se &&
|
||
se.SocketErrorCode == SocketError.TimedOut)
|
||
{
|
||
// expected
|
||
}
|
||
catch (IOException) { }
|
||
finally
|
||
{
|
||
stream.ReadTimeout = (int)MqttTimeout.TotalMilliseconds;
|
||
}
|
||
}
|
||
|
||
private static void ExpectDisconnect(NetworkStream stream)
|
||
{
|
||
stream.ReadTimeout = (int)MqttTimeout.TotalMilliseconds;
|
||
var buf = new byte[64];
|
||
try
|
||
{
|
||
var n = stream.Read(buf, 0, buf.Length);
|
||
n.ShouldBe(0, $"Expected server to close the connection, got {n} bytes");
|
||
}
|
||
catch (EndOfStreamException) { }
|
||
catch (IOException) { }
|
||
}
|
||
|
||
private static void SendDisconnect(NetworkStream stream)
|
||
{
|
||
var w = new ByteWriter();
|
||
w.WriteByte(PacketDisconnect);
|
||
w.WriteByte(0);
|
||
WriteAll(stream, w.ToArray());
|
||
}
|
||
|
||
private sealed class ByteWriter
|
||
{
|
||
private readonly List<byte> _buf = [];
|
||
|
||
public void WriteByte(byte b) => _buf.Add(b);
|
||
public void Write(byte[] data) { foreach (var b in data) _buf.Add(b); }
|
||
|
||
public void WriteUInt16(ushort v)
|
||
{
|
||
_buf.Add((byte)(v >> 8));
|
||
_buf.Add((byte)v);
|
||
}
|
||
|
||
public void WriteBytes(byte[] data)
|
||
{
|
||
WriteUInt16((ushort)data.Length);
|
||
Write(data);
|
||
}
|
||
|
||
public void WriteVarInt(int v)
|
||
{
|
||
while (true)
|
||
{
|
||
var b = (byte)(v & 0x7F);
|
||
v >>= 7;
|
||
if (v > 0) b |= 0x80;
|
||
_buf.Add(b);
|
||
if (v == 0) break;
|
||
}
|
||
}
|
||
|
||
public byte[] ToArray() => [.. _buf];
|
||
}
|
||
|
||
// -------------------------------------------------------------------------
|
||
// Helpers used by cluster tests
|
||
// -------------------------------------------------------------------------
|
||
|
||
/// <summary>
|
||
/// Checks a received PUBLISH message against expected topic, flags, and payload.
|
||
/// Mirrors Go's testMQTTCheckPubMsg.
|
||
/// </summary>
|
||
private static void CheckPubMsg(NetworkStream stream, string expectedTopic,
|
||
byte expectedFlags, byte[] expectedPayload)
|
||
{
|
||
var (flags, pi, topic, payload) = ReadPublish(stream);
|
||
topic.ShouldBe(expectedTopic, "PUBLISH topic mismatch");
|
||
flags.ShouldBe(expectedFlags, $"PUBLISH flags mismatch (got 0x{flags:X2}, expected 0x{expectedFlags:X2})");
|
||
payload.ShouldBe(expectedPayload, "PUBLISH payload mismatch");
|
||
|
||
// ACK QoS 1 messages
|
||
if (((flags & PubFlagQoS) >> 1) == 1)
|
||
{
|
||
SendPiPacket(stream, PacketPubAck, pi);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Connects, verifies CONNACK, then disconnects. Mirrors Go's testMQTTConnectDisconnect.
|
||
/// </summary>
|
||
private static void MqttConnectDisconnect(string host, int port,
|
||
string clientId, bool clean, bool expectedSessionPresent)
|
||
{
|
||
using var conn = MqttConnect(host, port, clientId: clientId, cleanSess: clean);
|
||
CheckConnAck(conn.Stream, ConnAckAccepted, expectedSessionPresent);
|
||
SendDisconnect(conn.Stream);
|
||
}
|
||
|
||
// -------------------------------------------------------------------------
|
||
// Tests — 8 MQTT cluster tests from server/mqtt_test.go
|
||
// -------------------------------------------------------------------------
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTSubPropagation (mqtt_test.go:3014).
|
||
/// Verifies that MQTT subscriptions on one cluster node cause interest
|
||
/// propagation to other nodes. Specifically checks that foo/# matches
|
||
/// foo.bar, foo./, and foo on the NATS side.
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2215
|
||
public void MQTTSubPropagation_ShouldSucceed()
|
||
{
|
||
// 2-server cluster; subscribe on S1 with foo/#, publish from NATS on S2.
|
||
// Verify propagation covers foo.bar, foo./, and foo.
|
||
//
|
||
// Go test:
|
||
// cl := createJetStreamClusterWithTemplate(t, ..., "MQTT", 2)
|
||
// mc, r := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
|
||
// testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: "foo/#", qos: 0}}, []byte{0})
|
||
// testMQTTFlush(t, mc, nil, r)
|
||
// checkSubInterest(t, s2, globalAccountName, "foo", time.Second)
|
||
//
|
||
// natsPub(t, nc, "foo.bar", []byte("hello"))
|
||
// testMQTTCheckPubMsg(t, mc, r, "foo/bar", 0, []byte("hello"))
|
||
//
|
||
// natsPub(t, nc, "foo./", []byte("from"))
|
||
// testMQTTCheckPubMsg(t, mc, r, "foo/", 0, []byte("from"))
|
||
//
|
||
// natsPub(t, nc, "foo", []byte("NATS"))
|
||
// testMQTTCheckPubMsg(t, mc, r, "foo", 0, []byte("NATS"))
|
||
|
||
// Cluster setup: 2 servers with MQTT + JetStream
|
||
var srv1Host = "127.0.0.1";
|
||
var srv1Port = 1883;
|
||
|
||
using var mc = MqttConnect(srv1Host, srv1Port, cleanSess: true);
|
||
CheckConnAckAccepted(mc.Stream);
|
||
|
||
// Subscribe to foo/# on server 1
|
||
MqttSubscribe(mc.Stream, 1, [("foo/#", 0)]);
|
||
MqttFlush(mc.Stream);
|
||
|
||
// Publish on NATS subjects from server 2; MQTT subscriber should receive
|
||
// foo.bar → foo/bar, foo./ → foo/, foo → foo
|
||
// NOTE: This requires a NATS client connected to S2 to publish.
|
||
// When running against a real cluster, the NATS publish would be done
|
||
// via natsConnect(s2.ClientURL()) / natsPub().
|
||
|
||
// Verify received messages match expected MQTT topics
|
||
// CheckPubMsg(mc.Stream, "foo/bar", 0, "hello"u8.ToArray());
|
||
// CheckPubMsg(mc.Stream, "foo/", 0, "from"u8.ToArray());
|
||
// CheckPubMsg(mc.Stream, "foo", 0, "NATS"u8.ToArray());
|
||
|
||
SendDisconnect(mc.Stream);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTCluster (mqtt_test.go:3044).
|
||
/// Tests cross-node pub/sub and session takeover across a 2-server cluster.
|
||
/// Runs subtests: {first_start, restart} × {qos_0, qos_1}.
|
||
/// After first_start, the cluster is restarted and the tests are repeated.
|
||
/// Verifies that:
|
||
/// - NATS-published messages reach MQTT subscribers across nodes
|
||
/// - MQTT-published messages (QoS 0 and 1) reach subscribers
|
||
/// - Session takeover from another node disconnects the first client
|
||
/// - QoS downgrade to subscriber QoS level works correctly
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2216
|
||
public void MQTTCluster_ShouldSucceed()
|
||
{
|
||
// 2-server cluster with nested subtests.
|
||
//
|
||
// Go test structure:
|
||
// for _, topTest := range [{first_start, true}, {restart, false}] {
|
||
// for _, test := range [{qos_0, 0}, {qos_1, 1}] {
|
||
// clientID := nuid.Next()
|
||
// mc, r := testMQTTConnectRetry(cleanSess: false)
|
||
// testMQTTSub(1, mc, r, "foo/#", test.subQos)
|
||
//
|
||
// check := func(mc, r, o, s) {
|
||
// // NATS pub → MQTT sub
|
||
// natsPub(nc, "foo.bar", "fromNats")
|
||
// testMQTTCheckPubMsg(mc, r, "foo/bar", 0, "fromNats")
|
||
// // MQTT pub QoS0 → MQTT sub
|
||
// testMQTTPublish(mpc, pr, 0, false, false, "foo/baz", 0, "mqtt_qos0")
|
||
// testMQTTCheckPubMsg(mc, r, "foo/baz", 0, "mqtt_qos0")
|
||
// // MQTT pub QoS1 → MQTT sub (downgraded to sub QoS)
|
||
// testMQTTPublish(mpc, pr, 1, false, false, "foo/bat", 1, "mqtt_qos1")
|
||
// expectedQoS := 0 or PubQoS1 depending on test.subQos
|
||
// testMQTTCheckPubMsg(mc, r, "foo/bat", expectedQoS, "mqtt_qos1")
|
||
// }
|
||
// check(mc, r, cl.opts[0], cl.servers[0])
|
||
// check(mc, r, cl.opts[1], cl.servers[1])
|
||
//
|
||
// // Session takeover from other node
|
||
// mc2, r2 := testMQTTConnect(clientID, cleanSess: false, server 2)
|
||
// testMQTTCheckConnAck(r2, accepted, sessionPresent: true)
|
||
// testMQTTExpectDisconnect(mc) // first connection kicked
|
||
// check(mc2, r2, both servers)
|
||
//
|
||
// // Clean up session
|
||
// testMQTTDisconnect(mc2)
|
||
// mc2 = testMQTTConnect(clientID, cleanSess: true)
|
||
// testMQTTDisconnect(mc2)
|
||
// }
|
||
// if topTest.restart { cl.stopAll(); cl.restartAll(); waitOnStreams }
|
||
// }
|
||
|
||
var host = "127.0.0.1";
|
||
var port = 1883;
|
||
|
||
foreach (var (topTestName, restart) in new[] { ("first_start", true), ("restart", false) })
|
||
{
|
||
foreach (var (testName, subQos) in new[] { ("qos_0", (byte)0), ("qos_1", (byte)1) })
|
||
{
|
||
var clientId = Guid.NewGuid().ToString("N");
|
||
|
||
using var mc = MqttConnect(host, port, clientId: clientId, cleanSess: false);
|
||
CheckConnAckAccepted(mc.Stream);
|
||
|
||
MqttSubscribe(mc.Stream, 1, [("foo/#", subQos)]);
|
||
MqttFlush(mc.Stream);
|
||
|
||
// Cross-node pub/sub checks would happen here with real cluster
|
||
// Session takeover from server 2 would disconnect mc
|
||
// Clean up session state
|
||
|
||
SendDisconnect(mc.Stream);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTClusterConnectDisconnectClean (mqtt_test.go:3176).
|
||
/// Stress test: 100 random connect/disconnect iterations with clean sessions
|
||
/// across a 3-server cluster. Skipped in Go source (flaky due to timing).
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2217
|
||
public void MQTTClusterConnectDisconnectClean_ShouldSucceed()
|
||
{
|
||
// Go test (skipped in Go source):
|
||
// t.Skip()
|
||
// nServers := 3
|
||
// cl := createJetStreamClusterWithTemplate(..., "MQTT", nServers)
|
||
// clientID := nuid.Next()
|
||
// N := 100
|
||
// for n := 0; n < N; n++ {
|
||
// testMQTTConnectDisconnect(t, cl.opts[rand.Intn(nServers)], clientID, true, false)
|
||
// }
|
||
|
||
var clientId = Guid.NewGuid().ToString("N");
|
||
var hosts = new[] { ("127.0.0.1", 1883), ("127.0.0.1", 1884), ("127.0.0.1", 1885) };
|
||
var rng = new Random();
|
||
|
||
for (var n = 0; n < 100; n++)
|
||
{
|
||
var (host, port) = hosts[rng.Next(hosts.Length)];
|
||
MqttConnectDisconnect(host, port, clientId, clean: true, expectedSessionPresent: false);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTClusterConnectDisconnectPersist (mqtt_test.go:3198).
|
||
/// 20 iterations of persistent session connect/disconnect across a 3-server cluster.
|
||
/// On each iteration, first cleans sessions on all servers, then performs persistent
|
||
/// connects verifying session-present flag progression. Skipped in Go (flaky).
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2218
|
||
public void MQTTClusterConnectDisconnectPersist_ShouldSucceed()
|
||
{
|
||
// Go test (skipped in Go source):
|
||
// t.Skip()
|
||
// nServers := 3
|
||
// cl := createJetStreamClusterWithTemplate(..., "MQTT", nServers)
|
||
// clientID := nuid.Next()
|
||
// N := 20
|
||
// for n := 0; n < N; n++ {
|
||
// // First clean sessions on all servers
|
||
// for i := 0; i < nServers; i++ {
|
||
// testMQTTConnectDisconnect(t, cl.opts[i], clientID, true, false)
|
||
// }
|
||
// testMQTTConnectDisconnect(t, cl.opts[0], clientID, false, false)
|
||
// testMQTTConnectDisconnect(t, cl.opts[1], clientID, false, true)
|
||
// testMQTTConnectDisconnect(t, cl.opts[2], clientID, false, true)
|
||
// testMQTTConnectDisconnect(t, cl.opts[0], clientID, false, true)
|
||
// }
|
||
|
||
var clientId = Guid.NewGuid().ToString("N");
|
||
var hosts = new[] { ("127.0.0.1", 1883), ("127.0.0.1", 1884), ("127.0.0.1", 1885) };
|
||
|
||
for (var n = 0; n < 20; n++)
|
||
{
|
||
// Clean sessions on all servers
|
||
foreach (var (host, port) in hosts)
|
||
{
|
||
MqttConnectDisconnect(host, port, clientId, clean: true, expectedSessionPresent: false);
|
||
}
|
||
|
||
// Persistent connects: first has no session, subsequent find existing session
|
||
MqttConnectDisconnect(hosts[0].Item1, hosts[0].Item2, clientId, clean: false, expectedSessionPresent: false);
|
||
MqttConnectDisconnect(hosts[1].Item1, hosts[1].Item2, clientId, clean: false, expectedSessionPresent: true);
|
||
MqttConnectDisconnect(hosts[2].Item1, hosts[2].Item2, clientId, clean: false, expectedSessionPresent: true);
|
||
MqttConnectDisconnect(hosts[0].Item1, hosts[0].Item2, clientId, clean: false, expectedSessionPresent: true);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTClusterRetainedMsg (mqtt_test.go:3228).
|
||
/// Tests retained message behavior across a 2-server cluster:
|
||
/// - Retained messages published on one server are received on the other
|
||
/// - New subscribers receive retained messages with retain flag
|
||
/// - Empty retained message deletes the retained message
|
||
/// - After deletion, reconnecting subscribers don't receive the retained message
|
||
/// - Network deletes (cross-node retained message removal) work correctly
|
||
/// - After network delete + new retained message, reconnecting subs get the new message
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2219
|
||
public void MQTTClusterRetainedMsg_ShouldSucceed()
|
||
{
|
||
// Go test:
|
||
// cl := createJetStreamClusterWithTemplate(..., "MQTT", 2)
|
||
// srv1Opts, srv2Opts := cl.opts[0], cl.opts[1]
|
||
//
|
||
// -- Subscribe on S1 --
|
||
// mc, rc := testMQTTConnectRetry(clientID: "sub", cleanSess: false, S1)
|
||
// testMQTTSub(1, mc, rc, "foo/#", qos: 1)
|
||
//
|
||
// -- Publish retained from S2 --
|
||
// mp, rp := testMQTTConnect(cleanSess: true, S2)
|
||
// testMQTTPublish(mp, rp, 1, false, true, "foo/bar", 1, "retained")
|
||
// testMQTTCheckPubMsg(mc, rc, "foo/bar", PubQoS1, "retained")
|
||
//
|
||
// -- New sub on S1 gets retained msg with retain flag --
|
||
// mc2, rc2 := testMQTTConnect(cleanSess: true, S1)
|
||
// testMQTTSub(1, mc2, rc2, "foo/#", qos: 1)
|
||
// testMQTTCheckPubMsg(mc2, rc2, "foo/bar", PubQoS1|PubFlagRetain, "retained")
|
||
//
|
||
// -- Empty retained deletes from storage but still delivered --
|
||
// testMQTTPublish(mp, rp, 1, false, true, "foo/bar", 1, "")
|
||
// testMQTTCheckPubMsg(mc, rc, "foo/bar", PubQoS1, "")
|
||
//
|
||
// -- Reconnect to S2: retained should NOT be delivered --
|
||
// mc, rc = testMQTTConnect(clientID: "sub", cleanSess: false, S2)
|
||
// testMQTTSub(1, mc, rc, "foo/#", qos: 1)
|
||
// testMQTTExpectNothing(rc)
|
||
//
|
||
// -- Reconnect to S1: retained should NOT be delivered --
|
||
// mc, rc = testMQTTConnect(clientID: "sub", cleanSess: false, S1)
|
||
// testMQTTSub(1, mc, rc, "foo/#", qos: 1)
|
||
// testMQTTExpectNothing(rc)
|
||
//
|
||
// -- Network delete test --
|
||
// Sub on S1 ("sub_one"), sub on S2 ("sub_two"), both subscribe to "bar"
|
||
// Publish retained "msg1" from S2 → both receive
|
||
// Publish empty retained → both receive (delete)
|
||
// Publish retained "msg2" from S2 → both receive
|
||
// Reconnect "sub_one" on S1 → gets "msg2" with retain flag
|
||
|
||
var srv1Host = "127.0.0.1";
|
||
var srv1Port = 1883;
|
||
var srv2Host = "127.0.0.1";
|
||
var srv2Port = 1884;
|
||
|
||
// Subscribe on server 1
|
||
using var mc = MqttConnect(srv1Host, srv1Port, clientId: "sub", cleanSess: false);
|
||
CheckConnAckAccepted(mc.Stream);
|
||
MqttSubscribe(mc.Stream, 1, [("foo/#", 1)]);
|
||
MqttFlush(mc.Stream);
|
||
|
||
// Publish retained from server 2
|
||
using var mp = MqttConnect(srv2Host, srv2Port, cleanSess: true);
|
||
CheckConnAckAccepted(mp.Stream);
|
||
|
||
MqttPublish(mp.Stream, 1, false, true, "foo/bar", 1, "retained"u8.ToArray());
|
||
CheckPubMsg(mc.Stream, "foo/bar", PubQoS1, "retained"u8.ToArray());
|
||
|
||
// New subscriber on S1 gets retained message with retain flag
|
||
using var mc2 = MqttConnect(srv1Host, srv1Port, cleanSess: true);
|
||
CheckConnAckAccepted(mc2.Stream);
|
||
MqttSubscribe(mc2.Stream, 1, [("foo/#", 1)]);
|
||
CheckPubMsg(mc2.Stream, "foo/bar", PubQoS1 | PubFlagRetain, "retained"u8.ToArray());
|
||
SendDisconnect(mc2.Stream);
|
||
|
||
// Empty retained message deletes it from storage but still delivered
|
||
MqttPublish(mp.Stream, 1, false, true, "foo/bar", 1, []); // empty payload
|
||
CheckPubMsg(mc.Stream, "foo/bar", PubQoS1, []);
|
||
|
||
SendDisconnect(mc.Stream);
|
||
|
||
// Reconnect to S2: retained should NOT be delivered (was deleted)
|
||
using var mc3 = MqttConnect(srv2Host, srv2Port, clientId: "sub", cleanSess: false);
|
||
CheckConnAck(mc3.Stream, ConnAckAccepted, expectedSessionPresent: true);
|
||
MqttSubscribe(mc3.Stream, 1, [("foo/#", 1)]);
|
||
ExpectNothing(mc3.Stream);
|
||
SendDisconnect(mc3.Stream);
|
||
|
||
// Reconnect to S1: retained should NOT be delivered
|
||
using var mc4 = MqttConnect(srv1Host, srv1Port, clientId: "sub", cleanSess: false);
|
||
CheckConnAck(mc4.Stream, ConnAckAccepted, expectedSessionPresent: true);
|
||
MqttSubscribe(mc4.Stream, 1, [("foo/#", 1)]);
|
||
ExpectNothing(mc4.Stream);
|
||
SendDisconnect(mc4.Stream);
|
||
|
||
// --- Network delete test ---
|
||
|
||
// Subscribe on S1 ("sub_one") and S2 ("sub_two") to "bar"
|
||
using var mcOne = MqttConnect(srv1Host, srv1Port, clientId: "sub_one", cleanSess: false);
|
||
CheckConnAckAccepted(mcOne.Stream);
|
||
MqttSubscribe(mcOne.Stream, 1, [("bar", 1)]);
|
||
MqttFlush(mcOne.Stream);
|
||
|
||
using var mcTwo = MqttConnect(srv2Host, srv2Port, clientId: "sub_two", cleanSess: false);
|
||
CheckConnAckAccepted(mcTwo.Stream);
|
||
MqttSubscribe(mcTwo.Stream, 1, [("bar", 1)]);
|
||
MqttFlush(mcTwo.Stream);
|
||
|
||
// Publish retained "msg1" from S2
|
||
MqttPublish(mp.Stream, 1, false, true, "bar", 1, "msg1"u8.ToArray());
|
||
CheckPubMsg(mcOne.Stream, "bar", PubQoS1, "msg1"u8.ToArray());
|
||
CheckPubMsg(mcTwo.Stream, "bar", PubQoS1, "msg1"u8.ToArray());
|
||
|
||
// Empty retained deletes it (network delete for S1)
|
||
MqttPublish(mp.Stream, 1, false, true, "bar", 1, []);
|
||
CheckPubMsg(mcOne.Stream, "bar", PubQoS1, []);
|
||
CheckPubMsg(mcTwo.Stream, "bar", PubQoS1, []);
|
||
|
||
// New retained message
|
||
MqttPublish(mp.Stream, 1, false, true, "bar", 1, "msg2"u8.ToArray());
|
||
CheckPubMsg(mcOne.Stream, "bar", PubQoS1, "msg2"u8.ToArray());
|
||
CheckPubMsg(mcTwo.Stream, "bar", PubQoS1, "msg2"u8.ToArray());
|
||
|
||
// Reconnect sub_one on S1: should get msg2 with retain flag
|
||
SendDisconnect(mcOne.Stream);
|
||
|
||
using var mcOneRecon = MqttConnect(srv1Host, srv1Port, clientId: "sub_one", cleanSess: false);
|
||
CheckConnAck(mcOneRecon.Stream, ConnAckAccepted, expectedSessionPresent: true);
|
||
MqttSubscribe(mcOneRecon.Stream, 1, [("bar", 1)]);
|
||
CheckPubMsg(mcOneRecon.Stream, "bar", PubQoS1 | PubFlagRetain, "msg2"u8.ToArray());
|
||
|
||
SendDisconnect(mcOneRecon.Stream);
|
||
SendDisconnect(mcTwo.Stream);
|
||
SendDisconnect(mp.Stream);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTClusterReplicasCount (mqtt_test.go:3525).
|
||
/// Parameterized test verifying that MQTT JetStream streams are created with
|
||
/// the correct replica count based on cluster size:
|
||
/// size=1 → replicas=1, size=2 → replicas=2, size=3 → replicas=3, size=5 → replicas=3
|
||
/// Checks $MQTT_msgs, $MQTT_rmsgs, and $MQTT_sess streams.
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2222
|
||
public void MQTTClusterReplicasCount_ShouldSucceed()
|
||
{
|
||
// Go test:
|
||
// for _, test := range []struct{ size, replicas int }{
|
||
// {1, 1}, {2, 2}, {3, 3}, {5, 3},
|
||
// } {
|
||
// if test.size == 1 { single server } else { cluster of test.size }
|
||
// mc, rc := testMQTTConnectRetry(clientID: "sub", cleanSess: false)
|
||
// testMQTTSub(1, mc, rc, "foo/#", qos: 1)
|
||
// testMQTTFlush(mc, nil, rc)
|
||
// js, _ := nc.JetStream()
|
||
// for _, sname := range []string{mqttStreamName, mqttRetainedMsgsStreamName, mqttSessStreamName} {
|
||
// si, _ := js.StreamInfo(sname)
|
||
// si.Config.Replicas == test.replicas
|
||
// }
|
||
// }
|
||
|
||
var testCases = new[] { (Size: 1, Replicas: 1), (Size: 2, Replicas: 2), (Size: 3, Replicas: 3), (Size: 5, Replicas: 3) };
|
||
var mqttStreams = new[] { "$MQTT_msgs", "$MQTT_rmsgs", "$MQTT_sess" };
|
||
|
||
foreach (var (size, expectedReplicas) in testCases)
|
||
{
|
||
var host = "127.0.0.1";
|
||
var port = 1883;
|
||
|
||
using var mc = MqttConnect(host, port, clientId: "sub", cleanSess: false);
|
||
CheckConnAckAccepted(mc.Stream);
|
||
MqttSubscribe(mc.Stream, 1, [("foo/#", 1)]);
|
||
MqttFlush(mc.Stream);
|
||
|
||
// Verify replica counts on $MQTT_msgs, $MQTT_rmsgs, $MQTT_sess
|
||
// This requires a JetStream API call to check stream info.
|
||
// nc := natsConnect(s.ClientURL())
|
||
// js, _ := nc.JetStream()
|
||
// for each stream, js.StreamInfo(sname).Config.Replicas == expectedReplicas
|
||
|
||
SendDisconnect(mc.Stream);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTClusterCanCreateSessionWithOnServerDown (mqtt_test.go:3582).
|
||
/// Tests that MQTT sessions can still be created when one server in a 3-server
|
||
/// cluster is down. After shutting down one server, waits for quorum and verifies
|
||
/// a new MQTT connection can still be established on a remaining server.
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2223
|
||
public void MQTTClusterCanCreateSessionWithOnServerDown_ShouldSucceed()
|
||
{
|
||
// Go test:
|
||
// cl := createJetStreamClusterWithTemplate(..., "MQTT", 3)
|
||
// o := cl.opts[0]
|
||
// mc, rc := testMQTTConnectRetry(cleanSess: true, S1)
|
||
// testMQTTCheckConnAck(rc, accepted, false)
|
||
// mc.Close()
|
||
//
|
||
// // Shutdown server 2
|
||
// cl.servers[1].Shutdown()
|
||
// cl.waitOnPeerCount(2)
|
||
// cl.waitOnLeader()
|
||
//
|
||
// // Connect to server 3; should still work with quorum
|
||
// o = cl.opts[2]
|
||
// mc, rc = testMQTTConnectRetry(cleanSess: true, S3, retries: 5)
|
||
// testMQTTCheckConnAck(rc, accepted, false)
|
||
|
||
// Initial connection on server 1
|
||
var host1 = "127.0.0.1";
|
||
var port1 = 1883;
|
||
using var mc1 = MqttConnect(host1, port1, cleanSess: true);
|
||
CheckConnAckAccepted(mc1.Stream);
|
||
|
||
// Server 2 would be shut down here (cl.servers[1].Shutdown())
|
||
// Wait for quorum (cl.waitOnPeerCount(2), cl.waitOnLeader())
|
||
|
||
// Connect to server 3: should succeed with 2/3 quorum
|
||
var host3 = "127.0.0.1";
|
||
var port3 = 1885;
|
||
using var mc3 = MqttConnect(host3, port3, cleanSess: true);
|
||
CheckConnAckAccepted(mc3.Stream);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTClusterPlacement (mqtt_test.go:3611).
|
||
/// Tests that MQTT JetStream assets (streams and consumers) are placed on the
|
||
/// leaf node cluster (SPOKE) rather than the hub when using a SuperCluster
|
||
/// topology (3 hub clusters × 2 servers each + 3 SPOKE leaf nodes).
|
||
/// Verifies that all streams and their consumers have cluster name "SPOKE"
|
||
/// and all replicas have names prefixed with "SPOKE-".
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2224
|
||
public void MQTTClusterPlacement_ShouldSucceed()
|
||
{
|
||
// Go test:
|
||
// sc := createJetStreamSuperCluster(t, 3, 2)
|
||
// c := sc.randomCluster()
|
||
// lnc := c.createLeafNodesWithTemplateAndStartPort(jsClusterTemplWithLeafAndMQTT, "SPOKE", 3, 22111)
|
||
// sc.waitOnPeerCount(9)
|
||
// sc.waitOnLeader()
|
||
//
|
||
// for i := 0; i < 10; i++ {
|
||
// mc, rc := testMQTTConnectRetry(cleanSess: true, lnc.opts[i%3])
|
||
// testMQTTCheckConnAck(rc, accepted, false)
|
||
// }
|
||
//
|
||
// nc := natsConnect(lnc.servers[0].ClientURL())
|
||
// js, _ := nc.JetStream()
|
||
// for si := range js.StreamsInfo() {
|
||
// si.Cluster.Name == "SPOKE"
|
||
// all replicas have "SPOKE-" prefix
|
||
// for ci := range js.ConsumersInfo(si.Config.Name) {
|
||
// ci.Cluster.Name == "SPOKE"
|
||
// all replicas have "SPOKE-" prefix
|
||
// }
|
||
// }
|
||
|
||
// SuperCluster: 3 hub clusters × 2 servers + 3 SPOKE leaf nodes
|
||
// Connect 10 MQTT clients to SPOKE leaf nodes
|
||
var spokeHost = "127.0.0.1";
|
||
var spokePorts = new[] { 22111, 22112, 22113 };
|
||
|
||
for (var i = 0; i < 10; i++)
|
||
{
|
||
var port = spokePorts[i % 3];
|
||
using var mc = MqttConnect(spokeHost, port, cleanSess: true);
|
||
CheckConnAckAccepted(mc.Stream);
|
||
}
|
||
|
||
// Verify all MQTT streams are on SPOKE cluster
|
||
// Verify all consumers are on SPOKE cluster
|
||
// Verify all replica names start with "SPOKE-"
|
||
// This requires JetStream API calls via NATS client.
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ported from Go TestMQTTLeafnodeWithoutJSToClusterWithJSNoSharedSysAcc (mqtt_test.go:3664).
|
||
/// Tests MQTT connectivity through a leafnode without JetStream to a cluster
|
||
/// that has JetStream. Runs 6 sub-cases varying how JetStream domain resolution
|
||
/// is configured:
|
||
/// 0: JsAccDefaultDomain="", JS enabled in leaf with extra account
|
||
/// 1: JsAccDefaultDomain="", JS disabled in leaf
|
||
/// 2: JsAccDefaultDomain=DOMAIN, JS disabled in leaf
|
||
/// 3: MQTT.JsDomain=DOMAIN, JS enabled in leaf with extra account
|
||
/// 4: MQTT.JsDomain=DOMAIN, JS disabled in leaf
|
||
/// 5: JsAccDefaultDomain=DOMAIN, JS enabled in leaf with extra account
|
||
/// For each case, subscribes on leafnode, publishes from leafnode and hub,
|
||
/// verifies cross-topology message delivery.
|
||
/// </summary>
|
||
[Fact(Skip = "deferred: requires running NATS cluster")] // T:2225
|
||
public void MQTTLeafnodeWithoutJSToClusterWithJSNoSharedSysAcc_ShouldSucceed()
|
||
{
|
||
// Go test:
|
||
// test := func(t *testing.T, resolution int) {
|
||
// // 3-server hub cluster
|
||
// getClusterOpts := func(name, i) { ... hub opts ... }
|
||
// o1, o2, o3 := getClusterOpts("S1",1), getClusterOpts("S2",2), getClusterOpts("S3",3)
|
||
// s1, s2, s3 := testMQTTRunServer(oi)
|
||
// checkClusterFormed(s1, s2, s3)
|
||
// waitForJetStreamLeader()
|
||
//
|
||
// // Leafnode with MQTT but no JS (varies by resolution case)
|
||
// lno := testMQTTDefaultOptions()
|
||
// lno.JetStream = false
|
||
// switch resolution {
|
||
// case 0: lno.Accounts = [NewAccount("unused")]; lno.JsAccDefaultDomain = {"$G": ""}
|
||
// case 1: lno.JsAccDefaultDomain = {"$G": ""}
|
||
// case 2: lno.JsAccDefaultDomain = {"$G": "DOMAIN"}
|
||
// case 3: lno.Accounts = [NewAccount("unused")]; lno.MQTT.JsDomain = "DOMAIN"
|
||
// case 4: lno.MQTT.JsDomain = "DOMAIN"
|
||
// case 5: lno.Accounts = [NewAccount("unused")]; lno.JsAccDefaultDomain = {"$G": "DOMAIN"}
|
||
// }
|
||
// lno.LeafNode.Remotes = [{ URLs: [o1,o2,o3 LeafNode ports] }]
|
||
// ln := RunServer(lno)
|
||
// checkLeafNodeConnected(ln)
|
||
//
|
||
// // Subscribe on leafnode
|
||
// mc, rc := testMQTTConnectRetry(clientID: "sub", cleanSess: true, lno.MQTT)
|
||
// testMQTTSub(1, mc, rc, "foo", qos: 1)
|
||
//
|
||
// connectAndPublish := func(o *Options) {
|
||
// mp := testMQTTConnectRetry(cleanSess: true, o.MQTT)
|
||
// testMQTTPublish(mp, 1, false, false, "foo", 1, "msg")
|
||
// }
|
||
//
|
||
// // Pub from leafnode → sub receives
|
||
// connectAndPublish(lno)
|
||
// testMQTTCheckPubMsg(mc, rc, "foo", PubQoS1, "msg")
|
||
//
|
||
// // Pub from hub server → sub receives
|
||
// connectAndPublish(o3)
|
||
// testMQTTCheckPubMsg(mc, rc, "foo", PubQoS1, "msg")
|
||
//
|
||
// // Subscribe on hub server
|
||
// mc2, rc2 := testMQTTConnectRetry(clientID: "sub2", cleanSess: true, o2.MQTT)
|
||
// testMQTTSub(1, mc2, rc2, "foo", qos: 1)
|
||
//
|
||
// // Pub from leafnode → hub sub receives
|
||
// connectAndPublish(lno)
|
||
// testMQTTCheckPubMsg(mc2, rc2, "foo", PubQoS1, "msg")
|
||
//
|
||
// // Pub from hub → hub sub receives
|
||
// connectAndPublish(o1)
|
||
// testMQTTCheckPubMsg(mc2, rc2, "foo", PubQoS1, "msg")
|
||
// }
|
||
//
|
||
// 6 subtests with resolution 0-5
|
||
|
||
var subCases = new (string Name, int Resolution)[]
|
||
{
|
||
("backwards-compatibility-default-js-enabled-in-leaf", 0),
|
||
("backwards-compatibility-default-js-disabled-in-leaf", 1),
|
||
("backwards-compatibility-domain-js-disabled-in-leaf", 2),
|
||
("mqtt-explicit-js-enabled-in-leaf", 3),
|
||
("mqtt-explicit-js-disabled-in-leaf", 4),
|
||
("backwards-compatibility-domain-js-enabled-in-leaf", 5),
|
||
};
|
||
|
||
foreach (var (name, resolution) in subCases)
|
||
{
|
||
// 3-server hub cluster + leafnode with MQTT
|
||
// Configuration varies per resolution case
|
||
var leafHost = "127.0.0.1";
|
||
var leafPort = 1883;
|
||
var hubHost = "127.0.0.1";
|
||
var hubPorts = new[] { 1884, 1885, 1886 };
|
||
|
||
// Subscribe on leafnode
|
||
using var mc = MqttConnect(leafHost, leafPort, clientId: "sub", cleanSess: true);
|
||
CheckConnAckAccepted(mc.Stream);
|
||
MqttSubscribe(mc.Stream, 1, [("foo", 1)]);
|
||
MqttFlush(mc.Stream);
|
||
|
||
// Publish from leafnode
|
||
using var mp1 = MqttConnect(leafHost, leafPort, cleanSess: true);
|
||
CheckConnAckAccepted(mp1.Stream);
|
||
MqttPublish(mp1.Stream, 1, false, false, "foo", 1, "msg"u8.ToArray());
|
||
CheckPubMsg(mc.Stream, "foo", PubQoS1, "msg"u8.ToArray());
|
||
SendDisconnect(mp1.Stream);
|
||
|
||
// Publish from hub server 3
|
||
using var mp2 = MqttConnect(hubHost, hubPorts[2], cleanSess: true);
|
||
CheckConnAckAccepted(mp2.Stream);
|
||
MqttPublish(mp2.Stream, 1, false, false, "foo", 1, "msg"u8.ToArray());
|
||
CheckPubMsg(mc.Stream, "foo", PubQoS1, "msg"u8.ToArray());
|
||
SendDisconnect(mp2.Stream);
|
||
|
||
// Subscribe on hub server 2
|
||
using var mc2 = MqttConnect(hubHost, hubPorts[1], clientId: "sub2", cleanSess: true);
|
||
CheckConnAckAccepted(mc2.Stream);
|
||
MqttSubscribe(mc2.Stream, 1, [("foo", 1)]);
|
||
MqttFlush(mc2.Stream);
|
||
|
||
// Publish from leafnode → hub sub receives
|
||
using var mp3 = MqttConnect(leafHost, leafPort, cleanSess: true);
|
||
CheckConnAckAccepted(mp3.Stream);
|
||
MqttPublish(mp3.Stream, 1, false, false, "foo", 1, "msg"u8.ToArray());
|
||
CheckPubMsg(mc2.Stream, "foo", PubQoS1, "msg"u8.ToArray());
|
||
SendDisconnect(mp3.Stream);
|
||
|
||
// Publish from hub server 1 → hub sub receives
|
||
using var mp4 = MqttConnect(hubHost, hubPorts[0], cleanSess: true);
|
||
CheckConnAckAccepted(mp4.Stream);
|
||
MqttPublish(mp4.Stream, 1, false, false, "foo", 1, "msg"u8.ToArray());
|
||
CheckPubMsg(mc2.Stream, "foo", PubQoS1, "msg"u8.ToArray());
|
||
SendDisconnect(mp4.Stream);
|
||
|
||
SendDisconnect(mc2.Stream);
|
||
SendDisconnect(mc.Stream);
|
||
}
|
||
}
|
||
}
|