fix(mqtt): birth-typed Sparkplug writes and QoS 1 for Plain writes (#602 STAB-25/26)
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
@@ -74,9 +74,20 @@ public enum MqttWriteEncodeError
|
|||||||
/// </remarks>
|
/// </remarks>
|
||||||
public static class MqttWriteEncoder
|
public static class MqttWriteEncoder
|
||||||
{
|
{
|
||||||
/// <summary>QoS for an outbound write. See the remarks on <see cref="TryEncode"/>.</summary>
|
/// <summary>
|
||||||
|
/// QoS for a Sparkplug NCMD/DCMD command. See the remarks on <see cref="TryEncode"/>. Plain-mode
|
||||||
|
/// writes use <see cref="PlainWriteQos"/> instead.
|
||||||
|
/// </summary>
|
||||||
public const int WriteQos = 0;
|
public const int WriteQos = 0;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// QoS for a Plain-mode <c>writeTopic</c> publish (STAB-26). At QoS 1 the publish completes only
|
||||||
|
/// once the broker has PUBACKed (or refused) it, so a Good write status means "the broker accepted
|
||||||
|
/// the message" rather than "the bytes left this socket"; at QoS 0 a broker-side refusal or a
|
||||||
|
/// connection that dropped under the send was indistinguishable from success.
|
||||||
|
/// </summary>
|
||||||
|
public const int PlainWriteQos = 1;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Encodes one write for the tag <paramref name="def"/> under the driver's <paramref name="mode"/>.
|
/// Encodes one write for the tag <paramref name="def"/> under the driver's <paramref name="mode"/>.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -84,9 +95,16 @@ public static class MqttWriteEncoder
|
|||||||
/// <param name="def">The resolved tag definition.</param>
|
/// <param name="def">The resolved tag definition.</param>
|
||||||
/// <param name="value">The value to write.</param>
|
/// <param name="value">The value to write.</param>
|
||||||
/// <param name="encoded">The topic + payload to publish, when this returns <see cref="MqttWriteEncodeError.None"/>.</param>
|
/// <param name="encoded">The topic + payload to publish, when this returns <see cref="MqttWriteEncodeError.None"/>.</param>
|
||||||
|
/// <param name="effectiveDataType">
|
||||||
|
/// Sparkplug only (STAB-25): the type to encode the metric as when the tag did not author one — the
|
||||||
|
/// type the metric's live birth certificate declared, the same <c>authored ? authored : birth</c>
|
||||||
|
/// precedence the ingest path coerces reads with. <see langword="null"/> encodes as
|
||||||
|
/// <see cref="MqttTagDefinition.DataType"/>, which for an unauthored tag is only the record's
|
||||||
|
/// default and would command the metric with the wrong wire type. Ignored in Plain mode.
|
||||||
|
/// </param>
|
||||||
/// <returns>The encode outcome.</returns>
|
/// <returns>The encode outcome.</returns>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
/// <b>QoS 0, retain <see langword="false"/>.</b> Retain is the one that matters: a retained
|
/// <b>Retain <see langword="false"/> always; QoS 1 for Plain, 0 for Sparkplug commands.</b> Retain is the one that matters: a retained
|
||||||
/// command would be replayed by the broker to the device on every future subscribe — including
|
/// command would be replayed by the broker to the device on every future subscribe — including
|
||||||
/// the device's own reconnect — so a single stale write would be re-applied forever. This is the
|
/// the device's own reconnect — so a single stale write would be re-applied forever. This is the
|
||||||
/// same hazard <see cref="RebirthRequester"/> avoids for NCMD, and the reason neither is
|
/// same hazard <see cref="RebirthRequester"/> avoids for NCMD, and the reason neither is
|
||||||
@@ -96,13 +114,14 @@ public static class MqttWriteEncoder
|
|||||||
MqttMode mode,
|
MqttMode mode,
|
||||||
MqttTagDefinition def,
|
MqttTagDefinition def,
|
||||||
object? value,
|
object? value,
|
||||||
out MqttEncodedWrite encoded)
|
out MqttEncodedWrite encoded,
|
||||||
|
DriverDataType? effectiveDataType = null)
|
||||||
{
|
{
|
||||||
ArgumentNullException.ThrowIfNull(def);
|
ArgumentNullException.ThrowIfNull(def);
|
||||||
encoded = default;
|
encoded = default;
|
||||||
|
|
||||||
return mode == MqttMode.SparkplugB
|
return mode == MqttMode.SparkplugB
|
||||||
? TryEncodeSparkplug(def, value, out encoded)
|
? TryEncodeSparkplug(def, value, effectiveDataType, out encoded)
|
||||||
: TryEncodePlain(def, value, out encoded);
|
: TryEncodePlain(def, value, out encoded);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -163,7 +182,7 @@ public static class MqttWriteEncoder
|
|||||||
: value.ToString() ?? "";
|
: value.ToString() ?? "";
|
||||||
}
|
}
|
||||||
|
|
||||||
encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text));
|
encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text), PlainWriteQos);
|
||||||
return MqttWriteEncodeError.None;
|
return MqttWriteEncodeError.None;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -175,6 +194,7 @@ public static class MqttWriteEncoder
|
|||||||
private static MqttWriteEncodeError TryEncodeSparkplug(
|
private static MqttWriteEncodeError TryEncodeSparkplug(
|
||||||
MqttTagDefinition def,
|
MqttTagDefinition def,
|
||||||
object? value,
|
object? value,
|
||||||
|
DriverDataType? effectiveDataType,
|
||||||
out MqttEncodedWrite encoded)
|
out MqttEncodedWrite encoded)
|
||||||
{
|
{
|
||||||
encoded = default;
|
encoded = default;
|
||||||
@@ -206,7 +226,7 @@ public static class MqttWriteEncoder
|
|||||||
}
|
}
|
||||||
|
|
||||||
var metric = new Payload.Types.Metric { Name = metricName };
|
var metric = new Payload.Types.Metric { Name = metricName };
|
||||||
var error = ApplyValue(metric, def.DataType, value);
|
var error = ApplyValue(metric, def.DataTypeAuthored ? def.DataType : effectiveDataType ?? def.DataType, value);
|
||||||
if (error != MqttWriteEncodeError.None)
|
if (error != MqttWriteEncodeError.None)
|
||||||
{
|
{
|
||||||
return error;
|
return error;
|
||||||
@@ -227,7 +247,7 @@ public static class MqttWriteEncoder
|
|||||||
edgeNodeId,
|
edgeNodeId,
|
||||||
isDevice ? def.DeviceId : null);
|
isDevice ? def.DeviceId : null);
|
||||||
|
|
||||||
encoded = new MqttEncodedWrite(topic, payload.ToByteArray());
|
encoded = new MqttEncodedWrite(topic, payload.ToByteArray(), WriteQos);
|
||||||
return MqttWriteEncodeError.None;
|
return MqttWriteEncodeError.None;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -344,4 +364,5 @@ public static class MqttWriteEncoder
|
|||||||
/// <summary>One encoded outbound write: where it goes and what it carries.</summary>
|
/// <summary>One encoded outbound write: where it goes and what it carries.</summary>
|
||||||
/// <param name="Topic">The concrete topic to publish on.</param>
|
/// <param name="Topic">The concrete topic to publish on.</param>
|
||||||
/// <param name="Payload">The encoded message body.</param>
|
/// <param name="Payload">The encoded message body.</param>
|
||||||
public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload);
|
/// <param name="Qos">The MQTT QoS to publish at — see <see cref="MqttWriteEncoder.PlainWriteQos"/> / <see cref="MqttWriteEncoder.WriteQos"/>.</param>
|
||||||
|
public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload, int Qos = MqttWriteEncoder.WriteQos);
|
||||||
|
|||||||
@@ -544,6 +544,11 @@ public sealed class MqttDriver
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
/// <para>
|
/// <para>
|
||||||
|
/// <b>Also the write path's type (STAB-25).</b> <see cref="WriteAsync"/> encodes a Sparkplug
|
||||||
|
/// command with this same answer, so a birth-typed Int32 metric is commanded as Int32 rather than
|
||||||
|
/// as the record's default.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
/// <b>The precedence is the ingest path's, restated.</b> <c>SparkplugIngestor.PublishMetric</c>
|
/// <b>The precedence is the ingest path's, restated.</b> <c>SparkplugIngestor.PublishMetric</c>
|
||||||
/// coerces every value with <c>DataTypeAuthored ? authored : birth</c>; a discovery surface
|
/// coerces every value with <c>DataTypeAuthored ? authored : birth</c>; a discovery surface
|
||||||
/// that disagreed with it would advertise a type no value ever arrives as. The two must be
|
/// that disagreed with it would advertise a type no value ever arrives as. The two must be
|
||||||
@@ -674,7 +679,10 @@ public sealed class MqttDriver
|
|||||||
/// reference reject every other write in the same OPC UA request.
|
/// reference reject every other write in the same OPC UA request.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
/// <b>Good means "published", not "applied".</b> MQTT has no application-level write
|
/// <b>Good means "published", not "applied".</b> A Plain-mode write is published at QoS 1
|
||||||
|
/// (STAB-26), so "published" means the broker PUBACKed it — a broker-side refusal or a drop under
|
||||||
|
/// the send is Bad, not Good. Sparkplug NCMD/DCMD stay at QoS 0 (a redelivered command would be
|
||||||
|
/// re-applied). MQTT has no application-level write
|
||||||
/// acknowledgement, so a successful PUBLISH is the strongest truth available at this seam —
|
/// acknowledgement, so a successful PUBLISH is the strongest truth available at this seam —
|
||||||
/// the same optimistic-Good posture Galaxy's fire-and-forget write takes. Device-level truth
|
/// the same optimistic-Good posture Galaxy's fire-and-forget write takes. Device-level truth
|
||||||
/// arrives afterwards, as the echoed value on the tag's read topic, and corrects the node
|
/// arrives afterwards, as the echoed value on the tag's read topic, and corrects the node
|
||||||
@@ -709,7 +717,8 @@ public sealed class MqttDriver
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
var error = MqttWriteEncoder.TryEncode(_options.Mode, def, w.Value, out var encoded);
|
var error = MqttWriteEncoder.TryEncode(
|
||||||
|
_options.Mode, def, w.Value, out var encoded, ResolveDiscoveredDataType(def));
|
||||||
if (error != MqttWriteEncodeError.None)
|
if (error != MqttWriteEncodeError.None)
|
||||||
{
|
{
|
||||||
results[i] = new WriteResult(MapEncodeError(error));
|
results[i] = new WriteResult(MapEncodeError(error));
|
||||||
@@ -728,7 +737,7 @@ public sealed class MqttDriver
|
|||||||
try
|
try
|
||||||
{
|
{
|
||||||
await connection
|
await connection
|
||||||
.PublishAsync(encoded.Topic, encoded.Payload, MqttWriteEncoder.WriteQos, retain: false, cancellationToken)
|
.PublishAsync(encoded.Topic, encoded.Payload, encoded.Qos, retain: false, cancellationToken)
|
||||||
.ConfigureAwait(false);
|
.ConfigureAwait(false);
|
||||||
results[i] = new WriteResult(0u);
|
results[i] = new WriteResult(0u);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -350,6 +350,55 @@ public sealed class MqttWriteEncoderTests
|
|||||||
Payload.Parser.ParseFrom(encoded.Payload).Metrics[0].Name.ShouldBe("Node Control/Rebirth");
|
Payload.Parser.ParseFrom(encoded.Payload).Metrics[0].Name.ShouldBe("Node Control/Rebirth");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- STAB-25 / STAB-26 (#602) ---------------------------------------------------------------------
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void A_plain_write_publishes_at_QoS1_so_Good_means_the_broker_accepted_it()
|
||||||
|
{
|
||||||
|
MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("oven/set"), 42.5, out var encoded)
|
||||||
|
.ShouldBe(MqttWriteEncodeError.None);
|
||||||
|
|
||||||
|
encoded.Qos.ShouldBe(1);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void A_sparkplug_command_stays_at_QoS0()
|
||||||
|
{
|
||||||
|
// Pinned beside the Plain case so the two cannot be "unified" by accident: an NCMD/DCMD retry from
|
||||||
|
// QoS 1 redelivery would re-apply a command the device already took.
|
||||||
|
MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, Sparkplug("Filler1"), 1.0, out var encoded)
|
||||||
|
.ShouldBe(MqttWriteEncodeError.None);
|
||||||
|
|
||||||
|
encoded.Qos.ShouldBe(0);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void An_unauthored_sparkplug_type_is_commanded_with_the_birth_declared_type()
|
||||||
|
{
|
||||||
|
// DataTypeAuthored=false: the record's Float64 is only a default. The birth said Int32, so the DCMD
|
||||||
|
// must carry Int32 — a Double-typed command to an Int32 metric is the silent-drop shape.
|
||||||
|
var def = Sparkplug("Filler1") with { DataTypeAuthored = false };
|
||||||
|
|
||||||
|
MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 7, out var encoded, DriverDataType.Int32)
|
||||||
|
.ShouldBe(MqttWriteEncodeError.None);
|
||||||
|
|
||||||
|
var metric = Payload.Parser.ParseFrom(encoded.Payload).Metrics[0];
|
||||||
|
metric.Datatype.ShouldBe((uint)TahuDataType.Int32);
|
||||||
|
metric.IntValue.ShouldBe(7u);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void An_authored_sparkplug_type_wins_over_the_birth_type()
|
||||||
|
{
|
||||||
|
// Same precedence as the ingest path: authored beats birth.
|
||||||
|
var def = Sparkplug("Filler1", DriverDataType.Float64) with { DataTypeAuthored = true };
|
||||||
|
|
||||||
|
MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 7, out var encoded, DriverDataType.Int32)
|
||||||
|
.ShouldBe(MqttWriteEncodeError.None);
|
||||||
|
|
||||||
|
Payload.Parser.ParseFrom(encoded.Payload).Metrics[0].Datatype.ShouldBe((uint)TahuDataType.Double);
|
||||||
|
}
|
||||||
|
|
||||||
private static string EncodeTopic(MqttMode mode, MqttTagDefinition def)
|
private static string EncodeTopic(MqttMode mode, MqttTagDefinition def)
|
||||||
{
|
{
|
||||||
MqttWriteEncoder.TryEncode(mode, def, 1.0, out var encoded).ShouldBe(MqttWriteEncodeError.None);
|
MqttWriteEncoder.TryEncode(mode, def, 1.0, out var encoded).ShouldBe(MqttWriteEncodeError.None);
|
||||||
|
|||||||
Reference in New Issue
Block a user