Joseph Doherty
e7e84e0577
2: only a PUBACKed (QoS >= 1) publish advances LastSuccessfulRead; a QoS 0 Sparkplug command completing means only that the bytes left the local socket. 3: a no-verdict SUBSCRIBE caused by the caller's own cancellation now throws OperationCanceledException (refs still Bad + pending), so the breaker no longer counts a pipeline timeout as a failure. 4: the rebirth wait is the window capped at 15 s, below the AdminUI 20 s per-call timeout. 6: a 10-topic test pins the 8-in-flight publish bound. 7: partial-level wildcards and $share filters are pinned never to confirm. 8: the intended init-vs-shutdown supersede is commented at the re-check. Refs #744 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
468 lines
23 KiB
C#
468 lines
23 KiB
C#
using System.Diagnostics;
|
||
using System.Text.Json;
|
||
using Shouldly;
|
||
using Xunit;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
|
||
|
||
/// <summary>
|
||
/// The driver's own session lifecycle, against the <see cref="MiniBroker"/> fake: what
|
||
/// <see cref="MqttDriver.InitializeAsync"/> does to the session it is <i>replacing</i>, and what a
|
||
/// mis-authored duration knob does to the two paths whose whole job is to not fail
|
||
/// (<see cref="MqttDriver.ShutdownAsync"/> / <see cref="MqttDriver.DisposeAsync"/>).
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// Plain mode deliberately: it is the only mode whose initialize needs no SUBSCRIBE.
|
||
/// <see cref="MiniBroker"/> answers CONNECT, SUBSCRIBE (unless told not to) and PINGREQ, and never
|
||
/// PUBLISHes, so ingest is driven through <c>Subscriptions.HandleMessage</c>.
|
||
/// </remarks>
|
||
[Trait("Category", "Unit")]
|
||
public sealed class MqttDriverSessionLifecycleTests
|
||
{
|
||
/// <summary>
|
||
/// <b>Driver.Mqtt-002.</b> <c>InitializeCoreAsync</c> publishes <c>_connection</c> — an
|
||
/// <see cref="IAsyncDisposable"/> — so re-initializing a live driver without tearing down first
|
||
/// stranded a connected, supervised broker session that no later dispose could reach, and the
|
||
/// abandoned session's reconnect supervisor kept it alive against the broker indefinitely.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// The <c>IDriver</c> contract allows re-initialising a live driver with no
|
||
/// <c>ShutdownAsync</c>, and the #527 retry path's <c>teardownFirst</c> flag does not stop a
|
||
/// retry overlapping a still-running attempt, so the driver must not rely on a teardown.
|
||
/// </para>
|
||
/// <para>
|
||
/// The assertion is the broker's <b>live socket count</b>, not the driver's own view: a leak
|
||
/// is precisely the case where the driver believes it holds one session and the broker is
|
||
/// serving two. <see cref="MiniBroker.AcceptedCount"/> is asserted alongside it as the
|
||
/// control — without it, a second initialize that never dialled at all would satisfy
|
||
/// "exactly one live connection" just as well.
|
||
/// </para>
|
||
/// </remarks>
|
||
[Fact]
|
||
public async Task InitializeAsync_OnALiveDriver_TearsDownThePreviousSession()
|
||
{
|
||
using var broker = new MiniBroker();
|
||
var options = BrokerOptions(broker.Port);
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
await WaitForAsync(() => broker.LiveConnections == 1, "the first session to be established");
|
||
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
|
||
broker.AcceptedCount.ShouldBe(2, "the second initialize must genuinely open a new session");
|
||
await WaitForAsync(
|
||
() => broker.LiveConnections == 1,
|
||
"the replaced session to be torn down rather than left live and unreachable");
|
||
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
|
||
}
|
||
|
||
/// <summary>
|
||
/// <b>Driver.Mqtt-003.</b> <c>connectTimeoutSeconds</c> carries a <c>[Range]</c> annotation that
|
||
/// nothing validates on the runtime path — the blob is deserialized straight out of the deploy
|
||
/// artifact. A negative value is outside the legal range of both
|
||
/// <see cref="SemaphoreSlim.WaitAsync(TimeSpan)"/> and <see cref="CancellationTokenSource"/>, so
|
||
/// it threw <see cref="ArgumentOutOfRangeException"/> out of shutdown and dispose — the two calls
|
||
/// that must always complete, and the ones a host makes while tearing a bad deployment down.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task ShutdownAndDispose_WithANegativeConnectTimeout_DoNotThrow()
|
||
{
|
||
var options = BrokerOptions(1) with { ConnectTimeoutSeconds = -1, KeepAliveSeconds = -1 };
|
||
var driver = new MqttDriver(options, "d", null);
|
||
|
||
await Should.NotThrowAsync(async () =>
|
||
await driver.ShutdownAsync(TestContext.Current.CancellationToken));
|
||
await Should.NotThrowAsync(async () => await driver.DisposeAsync());
|
||
}
|
||
|
||
/// <summary>
|
||
/// The control for the test above: the normalization must not quietly rewrite a <i>legal</i>
|
||
/// value, or a deployment's authored connect budget would silently become the default.
|
||
/// </summary>
|
||
[Trait("Category", "SlowUnit")]
|
||
[Fact]
|
||
public async Task InitializeAsync_WithAnAuthoredConnectTimeout_KeepsIt()
|
||
{
|
||
using var broker = new MiniBroker { Blackhole = true };
|
||
var options = BrokerOptions(broker.Port) with { ConnectTimeoutSeconds = 2 };
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
|
||
var sw = Stopwatch.StartNew();
|
||
await Should.ThrowAsync<TimeoutException>(async () =>
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken));
|
||
sw.Stop();
|
||
|
||
sw.Elapsed.ShouldBeLessThan(
|
||
TimeSpan.FromSeconds(10),
|
||
"the authored 2s budget must still bound the connect, not the 15s default");
|
||
}
|
||
|
||
/// <summary>
|
||
/// <b>End-to-end through <see cref="MqttDriver"/> (archreview 05/STAB-18).</b> Deliver a value,
|
||
/// let the real broker drop the real session, and the driver publishes
|
||
/// <c>BadNoCommunication</c> with the value retained; feed the <b>identical</b> payload back —
|
||
/// the retained replay a broker sends on reconnect — and it publishes Good again.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// <b>What this adds over <c>MqttSubscriptionManagerRecoveryTests</c>:</b> the wiring. Those
|
||
/// drive <c>OnConnectionLost()</c> directly, so they stay green even if
|
||
/// <c>MqttSubscriptionManager.AttachTo</c> never subscribes to
|
||
/// <see cref="MqttConnection.ConnectionLost"/> — an unforwarded seam is this repo's
|
||
/// best-documented way to ship an inert fix. Here the edge comes from a genuine socket drop
|
||
/// through MQTTnet's own <c>DisconnectedAsync</c>.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Vacuity rule:</b> the recovery payload is byte-identical to the pre-outage one. A test
|
||
/// that replays a <i>different</i> value passes against a build with no stale at all — the
|
||
/// value change alone would defeat the suppression — and is not the guard.
|
||
/// </para>
|
||
/// <para>
|
||
/// Ingest is fed through <c>Subscriptions.HandleMessage</c>, the method MQTTnet's dispatcher
|
||
/// calls, because <see cref="MiniBroker"/> answers CONNECT, SUBSCRIBE and PINGREQ only: it
|
||
/// can drop a session (which is what this test needs from it) but cannot PUBLISH.
|
||
/// </para>
|
||
/// </remarks>
|
||
[Trait("Category", "SlowUnit")]
|
||
[Fact]
|
||
public async Task ABrokerDrop_MarksCachedRefsBadNoCommunication_AndTheIdenticalReplayRecovers()
|
||
{
|
||
const string rawPath = "Plant/Mqtt/dev1/Temp";
|
||
const string topic = "factory/oven/temp";
|
||
const uint good = 0x00000000u;
|
||
const uint badNoCommunication = 0x80310000u;
|
||
|
||
using var broker = new MiniBroker();
|
||
var options = BrokerOptions(broker.Port) with
|
||
{
|
||
RawTags =
|
||
[
|
||
new RawTagEntry(
|
||
rawPath,
|
||
$$"""{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
|
||
WriteIdempotent: false),
|
||
],
|
||
};
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
||
|
||
var raised = new List<DataValueSnapshot>();
|
||
driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e.Snapshot); };
|
||
|
||
driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
||
lock (raised) raised.Count.ShouldBe(1, "the first value must publish");
|
||
|
||
broker.DropAll();
|
||
await WaitForAsync(
|
||
() => { lock (raised) return raised.Count == 2; },
|
||
"the broker drop to publish a BadNoCommunication snapshot");
|
||
DataValueSnapshot staled;
|
||
lock (raised) staled = raised[1];
|
||
staled.StatusCode.ShouldBe(badNoCommunication);
|
||
staled.Value.ShouldBe(21.5, "the last known value is retained through the outage");
|
||
|
||
// The reconnect's retained replay — the IDENTICAL payload.
|
||
driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: true);
|
||
|
||
DataValueSnapshot recovered;
|
||
lock (raised) { raised.Count.ShouldBe(3); recovered = raised[2]; }
|
||
recovered.StatusCode.ShouldBe(good);
|
||
recovered.Value.ShouldBe(21.5);
|
||
}
|
||
|
||
/// <summary>
|
||
/// <b>R06:M-1, end to end.</b> A SUBSCRIBE the broker never answers used to come back from
|
||
/// <see cref="MqttDriver.SubscribeAsync"/> as a live handle while the topic left every retry set, so
|
||
/// <c>DriverInstanceActor</c> never re-subscribed and the tag stayed dark behind a Healthy connection.
|
||
/// It must throw (the actor's reconcile then retries), and the retry against a broker that answers
|
||
/// must establish and deliver — the positive control.
|
||
/// </summary>
|
||
[Trait("Category", "SlowUnit")]
|
||
[Fact]
|
||
public async Task ASubscribeTheBrokerNeverAnswers_Throws_AndTheRetryEstablishes()
|
||
{
|
||
const string rawPath = "Plant/Mqtt/dev1/Temp";
|
||
const string topic = "factory/oven/temp";
|
||
|
||
using var broker = new MiniBroker { AnswerSubscribe = false };
|
||
var options = BrokerOptions(broker.Port) with
|
||
{
|
||
RawTags =
|
||
[
|
||
new RawTagEntry(
|
||
rawPath,
|
||
$$"""{"topic":"{{topic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
|
||
WriteIdempotent: false),
|
||
],
|
||
};
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
|
||
await Should.ThrowAsync<InvalidOperationException>(() =>
|
||
driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken));
|
||
driver.GetHealth().State.ShouldBe(DriverState.Healthy, "the connection itself is fine — that was the trap");
|
||
|
||
broker.AnswerSubscribe = true;
|
||
var handle = await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
||
|
||
var raised = new List<DataChangeEventArgs>();
|
||
driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e); };
|
||
driver.Subscriptions.HandleMessage(topic, "7"u8.ToArray(), retained: false);
|
||
lock (raised)
|
||
{
|
||
raised.ShouldHaveSingleItem().SubscriptionHandle.ShouldBe(handle);
|
||
raised[0].Snapshot.StatusCode.ShouldBe(0u);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// <b>R06:M-3.</b> A refused initial connect used to rethrow without disposing the
|
||
/// <see cref="MqttConnection"/> (and its <c>IMqttClient</c>) it had just built. Control: a
|
||
/// successful initialize leaves its connection live, so the seam is not reporting "disposed" for
|
||
/// everything.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InitializeAsync_RefusedConnect_DisposesTheConnectionItBuilt()
|
||
{
|
||
using var broker = new MiniBroker { ConnAckReturnCode = MiniBroker.ReturnCodeNotAuthorized };
|
||
var options = BrokerOptions(broker.Port);
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
var built = new List<MqttConnection>();
|
||
driver.ConnectionCreatedForTests = c => built.Add(c);
|
||
|
||
await Should.ThrowAsync<Exception>(() => driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken));
|
||
|
||
built.ShouldHaveSingleItem().State.ShouldBe(MqttConnectionState.Disposed);
|
||
|
||
broker.ConnAckReturnCode = MiniBroker.ReturnCodeAccepted;
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
built.Count.ShouldBe(2);
|
||
built[1].State.ShouldBe(MqttConnectionState.Connected);
|
||
}
|
||
|
||
/// <summary>
|
||
/// <b>R06:M-10 + the lifecycle-generation rule.</b> A shutdown that times out on the lifecycle gate
|
||
/// tears down ungated; the wedged initialize that held the gate must then NOT publish its freshly
|
||
/// connected session behind that teardown, and its late health write must not overwrite the
|
||
/// shutdown's. The positive controls: the wedged init really did reach the broker (AcceptedCount),
|
||
/// and a later initialize connects and reports Healthy.
|
||
/// </summary>
|
||
[Trait("Category", "SlowUnit")]
|
||
[Fact]
|
||
public async Task AWedgedInit_CompletingAfterATimedOutShutdown_PublishesNothing()
|
||
{
|
||
using var broker = new MiniBroker();
|
||
var options = BrokerOptions(broker.Port) with { ConnectTimeoutSeconds = 1 };
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
|
||
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
driver.BeforeConnectHookForTests = async _ =>
|
||
{
|
||
entered.TrySetResult();
|
||
await release.Task;
|
||
};
|
||
|
||
var initialize = Task.Run(() => driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken));
|
||
await entered.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||
|
||
// Times out on the gate (1 s) and tears down ungated.
|
||
await driver.ShutdownAsync(TestContext.Current.CancellationToken);
|
||
driver.GetHealth().State.ShouldBe(DriverState.Unknown);
|
||
|
||
driver.BeforeConnectHookForTests = null;
|
||
release.SetResult();
|
||
await Should.ThrowAsync<OperationCanceledException>(() => initialize);
|
||
|
||
broker.AcceptedCount.ShouldBe(1, "the wedged initialize did connect — the control for the absence below");
|
||
await WaitForAsync(() => broker.LiveConnections == 0, "the superseded session to be torn down");
|
||
driver.GetHealth().State.ShouldBe(DriverState.Unknown, "a superseded init must not overwrite newer health");
|
||
driver.GetHostStatuses().ShouldHaveSingleItem().State.ShouldBe(HostState.Unknown);
|
||
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
await WaitForAsync(() => broker.LiveConnections == 1, "the fresh session");
|
||
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
|
||
}
|
||
|
||
private static RawTagEntry WritableTag(string rawPath, string writeTopic) => new(
|
||
rawPath,
|
||
$$"""{"topic":"{{writeTopic}}/state","payloadFormat":"Scalar","dataType":"Float64","writeTopic":"{{writeTopic}}"}""",
|
||
WriteIdempotent: false);
|
||
|
||
/// <summary>
|
||
/// <b>R06:M-8.</b> A write batch used to publish strictly serially, one awaited QoS 1 round-trip per
|
||
/// request — against a stalled broker, N writes × the publish deadline. Different topics now publish
|
||
/// concurrently: all four PUBLISHes reach a broker that never PUBACKs well inside ONE 30 s deadline
|
||
/// (serial code would have sent exactly one). Cancelling the batch then fills every slot, so the
|
||
/// batch shape holds.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task AWriteBatch_PublishesDifferentTopicsConcurrently()
|
||
{
|
||
using var broker = new MiniBroker(); // never PUBACKs
|
||
var options = BrokerOptions(broker.Port) with
|
||
{
|
||
ConnectTimeoutSeconds = 30,
|
||
RawTags = [.. Enumerable.Range(0, 4).Select(i => WritableTag($"Plant/Mqtt/dev1/Sp{i}", $"cmd/sp{i}"))],
|
||
};
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
|
||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||
var write = driver.WriteAsync(
|
||
[.. Enumerable.Range(0, 4).Select(i => new WriteRequest($"Plant/Mqtt/dev1/Sp{i}", (double)i))],
|
||
cts.Token);
|
||
|
||
await WaitForAsync(() => broker.Published.Count == 4, "all four PUBLISHes to be in flight at once");
|
||
write.IsCompleted.ShouldBeFalse("no PUBACK has arrived, so nothing can have completed");
|
||
|
||
await cts.CancelAsync();
|
||
var results = await write.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||
results.Count.ShouldBe(4);
|
||
results.ShouldAllBe(r => r.StatusCode == 0x80050000u); // BadCommunicationError, every slot filled
|
||
}
|
||
|
||
/// <summary>
|
||
/// Writes to the SAME topic in one batch stay in request order — last write wins at the device —
|
||
/// while the batch as a whole is concurrent. The broker PUBACKs here, so every slot is Good — and
|
||
/// those acks advance <c>LastSuccessfulRead</c> (writes feed health).
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task AWriteBatch_KeepsSameTopicWritesInRequestOrder()
|
||
{
|
||
using var broker = new MiniBroker { AckPublish = true };
|
||
var options = BrokerOptions(broker.Port) with
|
||
{
|
||
RawTags = [WritableTag("Plant/Mqtt/dev1/Sp", "cmd/sp"), WritableTag("Plant/Mqtt/dev1/Other", "cmd/other")],
|
||
};
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
driver.GetHealth().LastSuccessfulRead.ShouldBeNull("nothing has answered yet");
|
||
|
||
var results = await driver.WriteAsync(
|
||
[
|
||
new WriteRequest("Plant/Mqtt/dev1/Sp", 1.0),
|
||
new WriteRequest("Plant/Mqtt/dev1/Other", 9.0),
|
||
new WriteRequest("Plant/Mqtt/dev1/Sp", 2.0),
|
||
new WriteRequest("Plant/Mqtt/dev1/Sp", 3.0),
|
||
],
|
||
TestContext.Current.CancellationToken);
|
||
|
||
results.ShouldAllBe(r => r.StatusCode == 0u);
|
||
broker.Published.Where(p => p.Topic == "cmd/sp")
|
||
.Select(p => System.Text.Encoding.UTF8.GetString(p.Payload))
|
||
.ShouldBe(["1", "2", "3"]);
|
||
|
||
// A PUBACK is the broker answering, so writes feed the evidence clock the comms-loss watchdog reads.
|
||
driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Review minor 6: the bound itself. Ten topics against a broker that never PUBACKs: exactly
|
||
/// <see cref="MqttDriver.MaxConcurrentPublishes"/> PUBLISHes go out and the count then holds — a
|
||
/// regression to unbounded concurrency sends all ten. The stability check re-reads the count across
|
||
/// several condition polls rather than sleeping once.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task AWriteBatch_NeverHasMoreThanTheBoundInFlight()
|
||
{
|
||
using var broker = new MiniBroker(); // never PUBACKs
|
||
var options = BrokerOptions(broker.Port) with
|
||
{
|
||
ConnectTimeoutSeconds = 30,
|
||
RawTags = [.. Enumerable.Range(0, 10).Select(i => WritableTag($"Plant/Mqtt/dev1/Sp{i}", $"cmd/sp{i}"))],
|
||
};
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
|
||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||
var write = driver.WriteAsync(
|
||
[.. Enumerable.Range(0, 10).Select(i => new WriteRequest($"Plant/Mqtt/dev1/Sp{i}", (double)i))],
|
||
cts.Token);
|
||
|
||
await WaitForAsync(() => broker.Published.Count >= MqttDriver.MaxConcurrentPublishes, "the bound to fill");
|
||
for (var i = 0; i < 10; i++)
|
||
{
|
||
await Task.Delay(50, TestContext.Current.CancellationToken);
|
||
broker.Published.Count.ShouldBe(MqttDriver.MaxConcurrentPublishes, "a slot freed with no PUBACK received");
|
||
}
|
||
|
||
await cts.CancelAsync();
|
||
var results = await write.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||
results.Count.ShouldBe(10);
|
||
results.ShouldAllBe(r => r.StatusCode == 0x80050000u);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Review minor 2: a QoS 0 publish (a Sparkplug command) completing only means the bytes reached the
|
||
/// local socket buffer; the broker has not answered, so it must not advance <c>LastSuccessfulRead</c>.
|
||
/// The positive control is <see cref="AWriteBatch_KeepsSameTopicWritesInRequestOrder"/>, where a QoS 1
|
||
/// PUBACK does advance it.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task ASparkplugQos0Command_DoesNotAdvanceLastSuccessfulRead()
|
||
{
|
||
using var broker = new MiniBroker();
|
||
var options = BrokerOptions(broker.Port) with
|
||
{
|
||
Mode = MqttMode.SparkplugB,
|
||
Sparkplug = new MqttSparkplugOptions { GroupId = "G", RequestRebirthOnGap = false },
|
||
RawTags =
|
||
[
|
||
new RawTagEntry(
|
||
"Plant/Mqtt/spb/Setpoint",
|
||
"""{"groupId":"G","edgeNodeId":"E","metricName":"Setpoint","dataType":"Float64"}""",
|
||
WriteIdempotent: false),
|
||
],
|
||
};
|
||
await using var driver = new MqttDriver(options, "d", null);
|
||
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
|
||
var before = driver.GetHealth().LastSuccessfulRead;
|
||
var publishedBefore = broker.Published.Count;
|
||
|
||
var results = await driver.WriteAsync(
|
||
[new WriteRequest("Plant/Mqtt/spb/Setpoint", 4.5)], TestContext.Current.CancellationToken);
|
||
|
||
results.ShouldHaveSingleItem().StatusCode.ShouldBe(0u);
|
||
await WaitForAsync(
|
||
() => broker.Published.Skip(publishedBefore).Any(p => p.Topic == "spBv1.0/G/NCMD/E"),
|
||
"the NCMD to reach the broker — the control that the write really went out");
|
||
driver.GetHealth().LastSuccessfulRead.ShouldBe(before);
|
||
}
|
||
|
||
private static async Task WaitForAsync(Func<bool> condition, string what)
|
||
{
|
||
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
|
||
while (DateTime.UtcNow < deadline)
|
||
{
|
||
if (condition())
|
||
{
|
||
return;
|
||
}
|
||
|
||
await Task.Delay(25, TestContext.Current.CancellationToken);
|
||
}
|
||
|
||
throw new Xunit.Sdk.XunitException($"Timed out waiting for {what}.");
|
||
}
|
||
|
||
private static MqttDriverOptions BrokerOptions(int port) => new()
|
||
{
|
||
Mode = MqttMode.Plain,
|
||
Host = "127.0.0.1",
|
||
Port = port,
|
||
UseTls = false,
|
||
ProtocolVersion = MqttProtocolVersion.V311,
|
||
KeepAliveSeconds = 300,
|
||
ConnectTimeoutSeconds = 5,
|
||
ReconnectMinBackoffSeconds = 1,
|
||
ReconnectMaxBackoffSeconds = 2,
|
||
};
|
||
|
||
private static string Json(MqttDriverOptions options) => JsonSerializer.Serialize(options, MqttJson.Options);
|
||
}
|