diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md
index f8a6f5a4..5a064550 100644
--- a/docs/drivers/Mqtt.md
+++ b/docs/drivers/Mqtt.md
@@ -467,10 +467,27 @@ Both publish with `retain: false`. A **Plain** write goes at **QoS 1** (STAB-26)
once the broker PUBACKs it, so `Good` means "the broker accepted it", and a broker refusal or a connection
that died under the send is `Bad`. **Sparkplug commands stay at QoS 0** — a redelivered NCMD/DCMD would be
re-applied. A Sparkplug metric the tag did not give a `dataType` is commanded with the type its live birth
-certificate declared (STAB-25), the same authored-beats-birth precedence reads use. Retain is the one that
+certificate declared (STAB-25), the same authored-beats-birth precedence reads use — using the birth's **own**
+Sparkplug datatype, not the lossy driver-type mapping, so Int8/UInt8/Text/Uuid/Bytes/File metrics are commanded
+with their real type; a birth type the encoder cannot represent (any `*Array`, DataSet, Template, …) is refused
+`BadNotSupported` rather than sent as another type. Retain is the one that
matters and is deliberately not configurable: a retained command would be replayed by the broker to the device on *every* future
subscribe — including the device's own reconnect — so one stale write would be re-applied forever.
+**Plain QoS 1 is at-least-once.** Know the consequences before pointing a Plain `writeTopic` at a non-idempotent
+device topic:
+
+- A lost PUBACK can make a device subscribed at QoS ≥ 1 with a persistent session receive the command **twice**
+ (at QoS 0 delivery was downgraded and a duplicate could not happen).
+- A lost PUBACK, or the `ConnectTimeoutSeconds` (15 s) deadline, reports `Bad` and the node is **reverted**, even
+ though the broker may already have delivered the write. `Good` is "the broker accepted it", `Bad` is "not
+ confirmed", never "not delivered".
+- Each write holds `DriverInstanceActor`'s mailbox until the PUBACK arrives — up to 15 s per write, one after
+ another; comparable to a Modbus round trip.
+- A broker advertising MQTT 5 Maximum QoS 0 will disconnect on every Plain write.
+
+There is no per-tag QoS option; make the receiving topic idempotent (absolute setpoints, not increments).
+
### `writeTopic` is separate from `topic`, and its absence is meaningful
A tag commonly reads `…/state` and writes `…/set`, so the destination is its own authored field. A
diff --git a/docs/drivers/Sql.md b/docs/drivers/Sql.md
index a31617ca..7b7ad3ae 100644
--- a/docs/drivers/Sql.md
+++ b/docs/drivers/Sql.md
@@ -95,6 +95,13 @@ that holds an INTEGER station id can be authored either `"7"` or `7`. A JSON obj
is not a value and is refused. On **PostgreSQL** a string bind is sent untyped (`NpgsqlDbType.Unknown`,
`PostgresDialect.ConfigureParameter`, STAB-24) so the server infers its type from the compared column;
bound as `text`, a `"7"` against an `integer` key fails with `operator does not exist: integer = text`.
+
+⚠️ **Known limitation — key normalisation.** Row slice-back matches the returned key cell by its *text* form
+(`Convert.ToString`, ordinal), while the server now matches by the column's inferred type. An authored key the
+server treats as equal but whose text differs therefore selects a row that the driver then cannot find, and the
+tag reads **no-row**: `"007"` against an `integer` column (returns `7`), `"1.50"` against `numeric` (returns
+`1.5`), and an upper-case GUID against a `uuid` column (returned lower-case). Author keys in the column's
+canonical text form. SQL Server behaves the same way.
(They were read asymmetrically until 2026-08-10, so a numerically
authored `keyValue` refused the tag while the identically-bound `whereValue` was served — and the
deploy gate reported the key as *missing* when it was right there in the blob.)
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
index 63bce6d1..982df163 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
@@ -239,14 +239,10 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
_logger.LogError(ex,
"AbLegacy driver initialise failed. Driver={DriverInstanceId}", _driverInstanceId);
- // Tear down any probe loops and cached state that were created before the failure so
- // that a caller who catches and abandons (rather than retrying via ReinitializeAsync)
- // doesn't leave orphaned background tasks, CancellationTokenSources, and libplctag
- // handles alive. Mirrors the body of ShutdownAsync without awaiting the poll engine
- // (nothing has been subscribed yet at init time).
- foreach (var state in _devices.Values)
- try { state.ProbeCts?.Cancel(); } catch { }
- JoinProbeLoopsSync();
+ // Release cached state created before the failure so a caller who catches and abandons
+ // (rather than retrying via ReinitializeAsync) doesn't leave CancellationTokenSources and
+ // libplctag handles alive. No probe-loop join here: the loops start as the LAST statement of
+ // the try and nothing after that can throw, so a failure never has a running loop to join.
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
index 3da7bf61..a84f454a 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
@@ -341,6 +341,14 @@ public sealed class MTConnectDriver
///
private long _sessionGeneration;
+ ///
+ /// Pumps retired by because they were bound to a replaced
+ /// session (STAB-20), kept so can join them — a pump nobody
+ /// awaits can still run its default-verdict arm (Degrade, the reconnect counter) after
+ /// the new session has written Healthy. Guarded by .
+ ///
+ private readonly List<(CancellationTokenSource Cts, Task Task)> _supersededPumps = [];
+
///
/// The Agent answered /sample with something that cannot be streamed at all, so the
/// pump gave up. Latched — reconnecting reproduces the identical answer forever, and a
@@ -1806,6 +1814,33 @@ public sealed class MTConnectDriver
}
}
+ /// Disposes and forgets superseded pumps that have finished. Caller must hold .
+ private void PruneSupersededPumpsCore()
+ {
+ for (var i = _supersededPumps.Count - 1; i >= 0; i--)
+ {
+ if (!_supersededPumps[i].Task.IsCompleted)
+ {
+ continue;
+ }
+
+ _supersededPumps[i].Cts.Dispose();
+ _supersededPumps.RemoveAt(i);
+ }
+ }
+
+ /// The retired-but-unjoined pump tasks (test barrier).
+ internal Task[] SupersededPumpTasks
+ {
+ get
+ {
+ lock (_subscriptionLock)
+ {
+ return [.. _supersededPumps.Select(p => p.Task)];
+ }
+ }
+ }
+
///
/// Starts the shared /sample pump, if there is anything to pump and anything to pump
/// from. Caller must hold .
@@ -1836,6 +1871,16 @@ public sealed class MTConnectDriver
try { _pumpCts?.Cancel(); }
catch (ObjectDisposedException) { /* already going away */ }
+ // Retain (don't drop) the retired pump so shutdown can join it; its source is disposed
+ // by whoever observes it complete.
+ PruneSupersededPumpsCore();
+ if (_pumpCts is not null && _pumpTask is not null)
+ {
+ _supersededPumps.Add((_pumpCts, _pumpTask));
+ _pumpCts = null;
+ }
+ _pumpTask = null;
+
_logger.LogInformation(
"MTConnect driver {DriverInstanceId} replaced a /sample pump bound to a retired Agent session.",
_driverInstanceId);
@@ -2726,15 +2771,17 @@ public sealed class MTConnectDriver
// One publish, so a concurrent lock-free reader sees the new client, the new probe
// cache, and the pump's opening cursor together or none of them.
+ // The generation bump and the session write are ONE critical section: a Subscribe landing
+ // between them would start a pump on the OLD session stamped with the NEW generation, and
+ // the restart's own StartSampleStreamCore would then treat it as current.
lock (_subscriptionLock)
{
+ Volatile.Write(
+ ref _session,
+ new AgentSession(client, probe, current.InstanceId, current.NextSequence));
_sessionGeneration++;
}
- Volatile.Write(
- ref _session,
- new AgentSession(client, probe, current.InstanceId, current.NextSequence));
-
var index = new MTConnectObservationIndex(binding.IndexTags);
index.Apply(current);
Volatile.Write(ref _index, index);
@@ -2864,6 +2911,7 @@ public sealed class MTConnectDriver
{
CancellationTokenSource? cts;
Task? task;
+ (CancellationTokenSource Cts, Task Task)[] retired;
lock (_subscriptionLock)
{
@@ -2871,6 +2919,27 @@ public sealed class MTConnectDriver
task = _pumpTask;
_pumpCts = null;
_pumpTask = null;
+ retired = [.. _supersededPumps];
+ _supersededPumps.Clear();
+ }
+
+ // Join pumps retired by a session replacement (already cancelled): they must not outlive
+ // teardown and write health / the reconnect counter over the next session.
+ foreach (var (retiredCts, retiredTask) in retired)
+ {
+ try
+ {
+ await retiredCts.CancelAsync().ConfigureAwait(false);
+ await retiredTask.WaitAsync(TeardownWait, cancellationToken).ConfigureAwait(false);
+ retiredCts.Dispose();
+ }
+ catch (Exception ex)
+ {
+ _logger.LogDebug(
+ ex,
+ "MTConnect driver {DriverInstanceId} could not join a superseded /sample pump within {TeardownWait}; it is cancelled and abandoned.",
+ _driverInstanceId, TeardownWait);
+ }
}
if (cts is null)
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs
index 5ebd5d96..cb564b30 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs
@@ -102,6 +102,14 @@ public static class MqttWriteEncoder
/// , 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.
///
+ ///
+ /// Sparkplug only: the metric's OWN birth-declared , used when the tag
+ /// authored no type. Preferred over because the
+ /// ToDriverDataType map is lossy (Int8→Int16, Uint8→Uint16, Text/Uuid/Bytes/File→String,
+ /// *Array→scalar), so commanding through it sends the wrong Tahu datatype. A birth type the
+ /// encoder cannot represent (arrays, DataSet, Template, …) is REFUSED as
+ /// — a wrong type is never sent.
+ ///
/// The encode outcome.
///
/// Retain always; QoS 1 for Plain, 0 for Sparkplug commands. Retain is the one that matters: a retained
@@ -115,13 +123,14 @@ public static class MqttWriteEncoder
MqttTagDefinition def,
object? value,
out MqttEncodedWrite encoded,
- DriverDataType? effectiveDataType = null)
+ DriverDataType? effectiveDataType = null,
+ SparkplugDataType? birthDataType = null)
{
ArgumentNullException.ThrowIfNull(def);
encoded = default;
return mode == MqttMode.SparkplugB
- ? TryEncodeSparkplug(def, value, effectiveDataType, out encoded)
+ ? TryEncodeSparkplug(def, value, effectiveDataType, birthDataType, out encoded)
: TryEncodePlain(def, value, out encoded);
}
@@ -195,6 +204,7 @@ public static class MqttWriteEncoder
MqttTagDefinition def,
object? value,
DriverDataType? effectiveDataType,
+ SparkplugDataType? birthDataType,
out MqttEncodedWrite encoded)
{
encoded = default;
@@ -226,7 +236,9 @@ public static class MqttWriteEncoder
}
var metric = new Payload.Types.Metric { Name = metricName };
- var error = ApplyValue(metric, def.DataTypeAuthored ? def.DataType : effectiveDataType ?? def.DataType, value);
+ var error = !def.DataTypeAuthored && birthDataType is { } birth
+ ? ApplyBirthValue(metric, birth, value)
+ : ApplyValue(metric, def.DataTypeAuthored ? def.DataType : effectiveDataType ?? def.DataType, value);
if (error != MqttWriteEncodeError.None)
{
return error;
@@ -251,6 +263,63 @@ public static class MqttWriteEncoder
return MqttWriteEncodeError.None;
}
+ ///
+ /// Sets the metric's value field and Datatype from the metric's own birth-declared Sparkplug
+ /// type, for the types whose driver-type mapping is lossy. Types with a 1:1 driver mapping delegate
+ /// to .
+ ///
+ /// The metric being built.
+ /// The birth certificate's declared datatype.
+ /// The value to convert.
+ /// The encode outcome.
+ private static MqttWriteEncodeError ApplyBirthValue(Payload.Types.Metric metric, SparkplugDataType birth, object value)
+ {
+ try
+ {
+ switch (birth)
+ {
+ case TahuDataType.Int8:
+ metric.Datatype = (uint)TahuDataType.Int8;
+ metric.IntValue = unchecked((uint)Convert.ToSByte(value, CultureInfo.InvariantCulture));
+ return MqttWriteEncodeError.None;
+
+ case TahuDataType.Uint8:
+ metric.Datatype = (uint)TahuDataType.Uint8;
+ metric.IntValue = Convert.ToByte(value, CultureInfo.InvariantCulture);
+ return MqttWriteEncodeError.None;
+
+ case TahuDataType.Text:
+ case TahuDataType.Uuid:
+ metric.Datatype = (uint)birth;
+ metric.StringValue = value as string ?? Convert.ToString(value, CultureInfo.InvariantCulture) ?? "";
+ return MqttWriteEncodeError.None;
+
+ case TahuDataType.Bytes:
+ case TahuDataType.File:
+ if (value is not byte[] bytes)
+ {
+ return MqttWriteEncodeError.ValueNotRepresentable;
+ }
+
+ metric.Datatype = (uint)birth;
+ metric.BytesValue = ByteString.CopyFrom(bytes);
+ return MqttWriteEncodeError.None;
+ }
+ }
+ catch (Exception ex) when (ex is InvalidCastException or FormatException or OverflowException)
+ {
+ return MqttWriteEncodeError.ValueNotRepresentable;
+ }
+
+ // Arrays, DataSet, Template, PropertySet, Unknown: refused — never sent as a scalar of another type.
+ if (birth.IsSparkplugArray() || birth.ToDriverDataType() is null)
+ {
+ return MqttWriteEncodeError.UnsupportedDataType;
+ }
+
+ return ApplyValue(metric, birth.ToDriverDataType()!.Value, value);
+ }
+
///
/// Sets the metric's typed value field and its Datatype from the tag's declared type.
///
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
index 36f5182b..fb39ed43 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
@@ -88,6 +88,9 @@ public sealed class MqttDriver
///
private const uint StatusBadNotWritable = 0x803B0000u;
+ /// OPC UA BadNotSupported — the value's wire type has no safe command encoding.
+ private const uint StatusBadNotSupported = 0x803D0000u;
+
/// The supplied value cannot be represented in the tag's declared data type.
private const uint StatusBadTypeMismatch = 0x80740000u;
@@ -579,6 +582,24 @@ public sealed class MqttDriver
return table?.ResolveByName(metric)?.DataType.ToDriverDataType() ?? def.DataType;
}
+ ///
+ /// The metric's own birth-declared Sparkplug datatype (unmapped), or when the
+ /// tag authored a type, is not a complete Sparkplug binding, or no birth has been seen. Fed to the
+ /// write encoder so lossy driver-type mappings (Int8, Text, Bytes, arrays, …) are not commanded with
+ /// the wrong wire type.
+ ///
+ private Org.Eclipse.Tahu.Protobuf.DataType? ResolveBirthDataType(MqttTagDefinition def)
+ {
+ if (def.DataTypeAuthored || _sparkplug is not { } sparkplug
+ || def.GroupId is not { } groupId || def.EdgeNodeId is not { } edgeNodeId || def.MetricName is not { } metric)
+ {
+ return null;
+ }
+
+ return sparkplug.Births.Find(new SparkplugScope(groupId, edgeNodeId, def.DeviceId))
+ ?.ResolveByName(metric)?.DataType;
+ }
+
// ---- ISubscribable (straight through to the manager) ----
///
@@ -718,7 +739,7 @@ public sealed class MqttDriver
}
var error = MqttWriteEncoder.TryEncode(
- _options.Mode, def, w.Value, out var encoded, ResolveDiscoveredDataType(def));
+ _options.Mode, def, w.Value, out var encoded, ResolveDiscoveredDataType(def), ResolveBirthDataType(def));
if (error != MqttWriteEncodeError.None)
{
results[i] = new WriteResult(MapEncodeError(error));
@@ -772,7 +793,7 @@ public sealed class MqttDriver
private static uint MapEncodeError(MqttWriteEncodeError error) => error switch
{
MqttWriteEncodeError.NoWriteDestination => StatusBadNotWritable,
- MqttWriteEncodeError.UnsupportedDataType => StatusBadNotWritable,
+ MqttWriteEncodeError.UnsupportedDataType => StatusBadNotSupported,
_ => StatusBadTypeMismatch,
};
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyShutdownJoinTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyShutdownJoinTests.cs
index 7b610135..c9cb080b 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyShutdownJoinTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyShutdownJoinTests.cs
@@ -51,4 +51,34 @@ public sealed class AbLegacyShutdownJoinTests
factory.Tags["S:0"].Disposed.ShouldBeTrue();
}
+
+ /// The synchronous Dispose() path joins the parked loop too (it used to cancel and dispose under it).
+ [Fact]
+ public async Task Dispose_WaitsForProbeLoopParkedInInit_ThenProbeRuntimeIsDisposed()
+ {
+ var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var factory = new FakeAbLegacyTagFactory
+ {
+ Customise = p => new ParkedInitTag(p, entered, release),
+ };
+ var drv = new AbLegacyDriver(new AbLegacyDriverOptions
+ {
+ Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0", AbLegacyPlcFamily.Slc500)],
+ Probe = new AbLegacyProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(50), ProbeAddress = "S:0" },
+ }, "drv-join-dispose", factory);
+ await drv.InitializeAsync("{}", CancellationToken.None);
+ await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+ var dispose = Task.Run(drv.Dispose);
+
+ await Task.Delay(300);
+ dispose.IsCompleted.ShouldBeFalse("Dispose must join the probe loop parked in runtime init");
+
+ // Positive control: releasing the parked call lets Dispose finish, with the probe runtime released.
+ release.SetResult();
+ await dispose.WaitAsync(TimeSpan.FromSeconds(5));
+
+ factory.Tags["S:0"].Disposed.ShouldBeTrue();
+ }
}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
index 738f7a5f..710c8f1d 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
@@ -35,6 +35,7 @@ public sealed class MTConnectPumpGenerationTests
"mt-gen",
_ => clients.Dequeue());
await driver.InitializeAsync("{}", Ct);
+ Task[] retired = [];
var blockOnce = 0;
var blocking = new ManualResetEventSlim(false);
@@ -75,6 +76,11 @@ public sealed class MTConnectPumpGenerationTests
blocking.Set();
await restart.WaitAsync(Watchdog, Ct);
+ // The pump retired by the restart is retained so shutdown can join it (it used to be dropped
+ // un-awaited and could write health after the new session was Healthy).
+ retired = driver.SupersededPumpTasks;
+ retired.ShouldNotBeEmpty();
+
// The restart installed a new client; the standing subscription must be streaming from IT.
await clientB.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
clientA.IsDisposed.ShouldBeTrue();
@@ -84,6 +90,10 @@ public sealed class MTConnectPumpGenerationTests
blocking.Set();
await driver.ShutdownAsync(CancellationToken.None);
}
+
+ // Shutdown joined every retired pump and forgot them.
+ retired.ShouldAllBe(t => t.IsCompleted);
+ driver.SupersededPumpTasks.ShouldBeEmpty();
}
}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs
index 56784606..480440b2 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs
@@ -287,6 +287,56 @@ public sealed class MqttWriteEncoderTests
MqttMode.SparkplugB, Sparkplug("Filler1", DriverDataType.Reference), "ref", out _)
.ShouldBe(MqttWriteEncodeError.UnsupportedDataType);
+ // ---- Birth-typed writes keep the birth's OWN Sparkplug datatype (review R5 MINOR-4) ----------
+
+ private static Payload.Types.Metric EncodeBirth(TahuDataType birth, object value)
+ {
+ var error = MqttWriteEncoder.TryEncode(
+ MqttMode.SparkplugB, Sparkplug("Filler1", DriverDataType.Float64) with { DataTypeAuthored = false }, value,
+ out var encoded, birth.ToDriverDataType(), birth);
+ error.ShouldBe(MqttWriteEncodeError.None);
+ return Payload.Parser.ParseFrom(encoded.Payload).Metrics.Single();
+ }
+
+ [Theory]
+ [InlineData(TahuDataType.Int8, (sbyte)-5)]
+ [InlineData(TahuDataType.Uint8, (byte)200)]
+ [InlineData(TahuDataType.Text, "hello")]
+ [InlineData(TahuDataType.Uuid, "123e4567-e89b-12d3-a456-426614174000")]
+ public void A_birth_typed_write_is_commanded_with_the_births_own_datatype(TahuDataType birth, object value)
+ => ((TahuDataType)EncodeBirth(birth, value).Datatype).ShouldBe(birth);
+
+ [Fact]
+ public void A_birth_typed_bytes_write_carries_bytes_with_the_bytes_datatype()
+ {
+ var metric = EncodeBirth(TahuDataType.Bytes, new byte[] { 1, 2, 3 });
+ ((TahuDataType)metric.Datatype).ShouldBe(TahuDataType.Bytes);
+ metric.BytesValue.ToByteArray().ShouldBe([1, 2, 3]);
+ }
+
+ [Theory]
+ [InlineData(TahuDataType.Int32Array)]
+ [InlineData(TahuDataType.DoubleArray)]
+ [InlineData(TahuDataType.DataSet)]
+ [InlineData(TahuDataType.Template)]
+ public void A_birth_type_with_no_command_encoding_is_refused_never_sent_as_another_type(TahuDataType birth)
+ {
+ MqttWriteEncoder.TryEncode(
+ MqttMode.SparkplugB, Sparkplug("Filler1") with { DataTypeAuthored = false }, 1, out var encoded,
+ birth.ToDriverDataType(), birth)
+ .ShouldBe(MqttWriteEncodeError.UnsupportedDataType);
+ encoded.Payload.ShouldBeNull();
+ }
+
+ [Fact]
+ public void An_authored_type_still_wins_over_the_birth_type()
+ {
+ var def = Sparkplug("Filler1", DriverDataType.Int32) with { DataTypeAuthored = true };
+ MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 7, out var encoded, null, TahuDataType.Text)
+ .ShouldBe(MqttWriteEncodeError.None);
+ ((TahuDataType)Payload.Parser.ParseFrom(encoded.Payload).Metrics.Single().Datatype).ShouldBe(TahuDataType.Int32);
+ }
+
[Fact]
public void The_mode_decides_the_shape_not_the_blob()
{