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>
|
||||
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;
|
||||
|
||||
/// <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>
|
||||
/// Encodes one write for the tag <paramref name="def"/> under the driver's <paramref name="mode"/>.
|
||||
/// </summary>
|
||||
@@ -84,9 +95,16 @@ public static class MqttWriteEncoder
|
||||
/// <param name="def">The resolved tag definition.</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="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>
|
||||
/// <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
|
||||
/// 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
|
||||
@@ -96,13 +114,14 @@ public static class MqttWriteEncoder
|
||||
MqttMode mode,
|
||||
MqttTagDefinition def,
|
||||
object? value,
|
||||
out MqttEncodedWrite encoded)
|
||||
out MqttEncodedWrite encoded,
|
||||
DriverDataType? effectiveDataType = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(def);
|
||||
encoded = default;
|
||||
|
||||
return mode == MqttMode.SparkplugB
|
||||
? TryEncodeSparkplug(def, value, out encoded)
|
||||
? TryEncodeSparkplug(def, value, effectiveDataType, out encoded)
|
||||
: TryEncodePlain(def, value, out encoded);
|
||||
}
|
||||
|
||||
@@ -163,7 +182,7 @@ public static class MqttWriteEncoder
|
||||
: 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;
|
||||
}
|
||||
|
||||
@@ -175,6 +194,7 @@ public static class MqttWriteEncoder
|
||||
private static MqttWriteEncodeError TryEncodeSparkplug(
|
||||
MqttTagDefinition def,
|
||||
object? value,
|
||||
DriverDataType? effectiveDataType,
|
||||
out MqttEncodedWrite encoded)
|
||||
{
|
||||
encoded = default;
|
||||
@@ -206,7 +226,7 @@ public static class MqttWriteEncoder
|
||||
}
|
||||
|
||||
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)
|
||||
{
|
||||
return error;
|
||||
@@ -227,7 +247,7 @@ public static class MqttWriteEncoder
|
||||
edgeNodeId,
|
||||
isDevice ? def.DeviceId : null);
|
||||
|
||||
encoded = new MqttEncodedWrite(topic, payload.ToByteArray());
|
||||
encoded = new MqttEncodedWrite(topic, payload.ToByteArray(), WriteQos);
|
||||
return MqttWriteEncodeError.None;
|
||||
}
|
||||
|
||||
@@ -344,4 +364,5 @@ public static class MqttWriteEncoder
|
||||
/// <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="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);
|
||||
|
||||
Reference in New Issue
Block a user