dohertj2/lmxopcua
1
0
Fork 0
Code Issues 17 Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(mtconnect): make a /sample stream end loud, and correct the gap contract (Task 7 review)

Browse Source 
Code review of c46540ae approved the framing algorithm but moved the risk
onto the contract around it.

C1 (critical). SampleAsync returned normally on three different events:
EOF, the closing --boundary--, and a response that was not multipart at
all (where it yielded one parsed snapshot and finished). The seam
documents the stream as yielding until ct is cancelled, so a Task 11 pump
written to that contract would silently drop its subscription or hot-loop
reconnecting; the non-multipart leg reported a configuration error as a
healthy finished stream, which is the #485 quiet-successful-termination
shape aimed at the very next task. Every non-cancellation end now throws:
MTConnectStreamEndedException (ConnectionClosed / ClosingBoundary -
transient) or MTConnectStreamNotSupportedException (configuration -
reconnecting reproduces it forever), sharing one catchable base. The
non-multipart body is still parsed first so an Agent's own MTConnectError
text wins, but the document is NOT yielded.

I1. IsSequenceGap moved to IMTConnectAgentClient (static) and its first
parameter is now expectedFrom. The old name and doc were wrong for every
chunk after the first - the correct comparand is the PREVIOUS chunk's
NextSequence, and comparing against the opening "from" argument reports a
gap on every chunk once the ring buffer rolls, i.e. an endless /current
re-baseline storm. Both doc blocks also now record that cppagent answers
from < firstSequence with an OUT_OF_RANGE MTConnectError under HTTP 200,
which surfaces as InvalidDataException and NOT as a gap - so Task 11 must
treat that parse failure as a re-baseline trigger too.

I2. Every multipart test served the whole body from one buffer, so every
framing test completed in a single ReadAsync - the split-boundary path,
the split-header path and the multi-fill loop were correct by inspection
only. A ChunkedStream double (N bytes per read, N = 1/3/7) now drives the
fixtures through a split transport, plus a >16 KB document that forces a
buffer resize. Falsifiability: removing either scan overlap, or
collapsing the fill loop, fails ONLY the new chunked tests.

I3. Disposal mid-enumeration surfaces as OperationCanceledException via
an internal dispose-linked token, not a raw ObjectDisposedException that
Task 9's re-init would inflict on an enumerating pump.

I5. HeartbeatMs, SampleIntervalMs and SampleCount join RequestTimeoutMs
in construction-time positive-value validation. All three reach the query
string and an Agent answers heartbeat=0 with an HTTP-200 MTConnectError
that names no config key.

I4/M2. The no-Content-length framing fallback logs a one-shot warning
(the client takes an optional ILogger); a multipart response with no
boundary parameter fails fast instead of degrading into a watchdog
timeout.

M5/M6. Shared element/attribute reading rules extracted to MTConnectXml;
the probe parser documents why it alone does not trim BOM/whitespace. The
literal U+FEFF in source became a '' escape.

Also, from Task 8: MTConnectObservation gained IsStructured (default
false), set from element.HasElements. A DATA_SET/TABLE observation's
<Entry> children concatenate through element.Value to nonsense ("12" for
two entries) that the index would publish as Good, and only the parser
can tell that apart from a legitimate space-bearing Message. False for
CONDITION observations - their value comes from the element name, so
child content cannot corrupt it.

275/275 green in this suite's own files.
This commit is contained in:
Joseph Doherty
2026-07-24 15:01:31 -04:00
parent ac0a284055
commit 5ba9f1be6f
9 changed files with 1232 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/2026-07-24-mtconnect-driver.md
+51
View File
@@ -385,6 +385,57 @@ dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests --filter "Ful
git add -A && git commit -m "feat(mtconnect): current/sample parse + sequence-gap detection (Task 7)"
```
### Task 7 review remediation (post-`c46540ae`) — three contract changes Tasks 9/11/13 must build against
Code review of `c46540ae` approved the framing algorithm but moved the risk onto the **contract**. Three
changes here are breaking relative to the sketch above; the rest are hardening.
1. **`SampleAsync` never ends normally except by cancellation.** The seam documents the stream as
yielding "indefinitely, until `ct` is cancelled", but the implementation returned normally on EOF, on
the closing `--boundary--`, and — worst — on a response that was not multipart at all, where it
yielded one parsed snapshot and finished. A pump written to the documented contract has no reason to
handle normal completion, so it would silently drop its subscription or hot-loop reconnecting; the
non-multipart leg additionally reported a **configuration error as a healthy finished stream**, which
is the #485 quiet-successful-termination shape aimed straight at Task 11. Every non-cancellation end
now throws: `MTConnectStreamEndedException` (with `MTConnectStreamEndReason.ConnectionClosed` /
`ClosingBoundary` — transient, reconnect) or `MTConnectStreamNotSupportedException` (configuration —
reconnecting reproduces it forever), sharing the base `MTConnectStreamException`. The non-multipart
body is still parsed first, so an Agent's own `MTConnectError` text still wins, but the document is
**not** yielded.
2. **`IsSequenceGap` moved to `IMTConnectAgentClient` (static) and its first parameter is now
`expectedFrom`.** The sketch's `requestedFrom` name and doc were wrong for every chunk after the
first: the correct comparand is the **previous chunk's `NextSequence`**, and comparing against the
opening `from` reports a gap on every chunk once the ring buffer rolls — an endless `/current`
re-baseline storm against a healthy stream. Rehomed onto the seam so Task 11 need not reference the
concrete client. **Task 11 must also treat an `OUT_OF_RANGE` `InvalidDataException` as a re-baseline
trigger:** cppagent answers a `from` below `firstSequence` with an `MTConnectError` under HTTP 200,
so ring-buffer overflow arrives as a *parse failure*, never as a gap-bearing chunk. `IsSequenceGap`
alone does not cover it. Documented on the seam.
3. **`MTConnectObservation` gained `IsStructured`** (default `false`), set from `element.HasElements`.
A DATA_SET/TABLE observation's `<Entry key=…>` children concatenate through `element.Value` to
nonsense (`"12"` for two entries) which the index would otherwise publish as Good, and only the
parser can tell that apart from a legitimate space-bearing `Message` — no value-shape heuristic
works. Deliberately **false for CONDITION** observations: their value comes from the element *name*,
so child content cannot corrupt it. The index maps the flag to a status; the parser does not.
Hardening in the same pass: disposal mid-enumeration now surfaces as `OperationCanceledException`
(via an internal dispose-linked token) rather than a raw `ObjectDisposedException`/`IOException` that
Task 9's re-init would otherwise inflict on an enumerating pump; `HeartbeatMs`, `SampleIntervalMs` and
`SampleCount` join `RequestTimeoutMs` in construction-time positive-value validation (all three reach
the query string, and an Agent answers `heartbeat=0` with an HTTP-200 `MTConnectError` that would name
no config key); a `multipart/*` response with no `boundary` parameter fails fast instead of degrading
into a watchdog timeout; the no-`Content-length` framing fallback logs a one-shot warning (the client
now takes an optional `ILogger`); and the shared element/attribute reading rules moved into
`MTConnectXml`, used by both parsers.
**Coverage gap closed, and worth remembering as a pattern.** Every multipart test served the whole body
from one buffer, so **every framing test completed in a single `ReadAsync`** — the split-boundary path,
the split-header path and the multi-fill loop were correct by inspection only, on the task whose
headline risk is framing. A `ChunkedStream` double returning N bytes per read (N = 1, 3, 7) now drives
the fixtures through a split transport. Falsifiability confirms the gap was real: removing the
split-boundary overlap, removing the split-header overlap, and collapsing `EnsureAsync`'s fill loop each
fail **only** the new chunked tests — every pre-existing multipart test stays green under all three.
---
## Task 8: `MTConnectObservationIndex` + `UNAVAILABLE → BadNoCommunication`
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs
+69 -7
View File
@@ -16,7 +16,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// <c>/sample</c> long poll. (Task 0 planned to build on the TrakHound MTConnect.NET libraries;
/// Tasks 6 and 7 proved they can neither parse a document nor frame the stream socket-free, and
/// dropped the references — see the Task 0 CORRECTION block in the plan.) A fake for tests only
/// needs to implement these three members.
/// needs to implement these three instance members; <see cref="IsSequenceGap"/> is a static
/// protocol rule that lives here, rather than on the implementation, so a consumer written
/// against this seam never has to reference the concrete client.
/// </remarks>
public interface IMTConnectAgentClient
{
@@ -38,13 +40,73 @@ public interface IMTConnectAgentClient
/// <summary>
/// Opens an Agent <c>/sample</c> long-poll stream starting at sequence <paramref name="from"/>
/// and yields one <see cref="MTConnectStreamsResult"/> per multipart chunk the Agent sends,
/// indefinitely, until <paramref name="ct"/> is cancelled. Each yielded result's
/// <see cref="MTConnectStreamsResult.FirstSequence"/> is load-bearing — the caller compares
/// it against the requested <paramref name="from"/> to detect a ring-buffer sequence gap and
/// re-baseline via <see cref="CurrentAsync"/>.
/// and yields one <see cref="MTConnectStreamsResult"/> per multipart chunk the Agent sends.
/// </summary>
/// <remarks>
/// <para>
/// <b>The only way this enumeration ends without throwing is <paramref name="ct"/> being
/// cancelled.</b> Every other way a stream can stop — the connection dropping, the Agent
/// writing its closing boundary, or the endpoint turning out not to stream at all —
/// throws an <see cref="MTConnectStreamException"/> naming which. A consumer therefore
/// never has to treat "the enumeration finished" as an ambiguous signal, and cannot
/// silently lose its subscription to a data path that quietly stopped (#485). See
/// <see cref="MTConnectStreamEndedException"/> (transient — reconnect) and
/// <see cref="MTConnectStreamNotSupportedException"/> (configuration — reconnecting will
/// reproduce it forever).
/// </para>
/// <para>
/// <b>Advancing the cursor.</b> After each chunk, the caller's next expected sequence is
/// that chunk's <see cref="MTConnectStreamsResult.NextSequence"/> — not the
/// <paramref name="from"/> this call was opened with. Feed that running value to
/// <see cref="IsSequenceGap"/> and, on reconnect, to a fresh <c>SampleAsync</c>.
/// </para>
/// </remarks>
/// <param name="from">The sequence number to resume streaming from.</param>
/// <param name="ct">Cancellation token; cancelling ends the stream.</param>
/// <param name="ct">Cancellation token; cancelling is the contracted way to end the stream.</param>
IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(long from, CancellationToken ct);
/// <summary>
/// Has the Agent's ring buffer overflowed past the sequence the caller expected next? True
/// when the chunk's oldest retained sequence is <i>strictly</i> newer than
/// <paramref name="expectedFrom"/>, meaning observations between the two were evicted before
/// the caller could read them and an incremental update would silently skip values.
/// </summary>
/// <remarks>
/// <para>
/// <b><paramref name="expectedFrom"/> is a running cursor, not the original
/// <c>from</c>.</b> It equals the <c>from</c> passed to <see cref="SampleAsync"/> only
/// for the <i>first</i> chunk; from the second chunk on it is the <b>previous</b>
/// chunk's <see cref="MTConnectStreamsResult.NextSequence"/>. Comparing every chunk
/// against the original <c>from</c> instead reports a gap on every chunk once the
/// Agent's buffer has rolled past it — an endless <c>/current</c> re-baseline storm
/// against a perfectly healthy stream.
/// </para>
/// <para>
/// Equality is not a gap: <c>firstSequence == expectedFrom</c> is the ordinary
/// contiguous case where the Agent simply had nothing older to send.
/// </para>
/// <para>
/// <b>This check alone does NOT cover every ring-buffer overflow.</b> It can only judge
/// a chunk the Agent was willing to send — and cppagent answers a <c>from</c> that has
/// already fallen out of its buffer with an <c>MTConnectError</c> / <c>OUT_OF_RANGE</c>
/// document served under <b>HTTP 200</b>, which this client surfaces as an
/// <see cref="InvalidDataException"/> out of <see cref="SampleAsync"/>, never as a chunk.
/// A pump that re-baselines only on <c>IsSequenceGap</c> will therefore fault instead of
/// recovering exactly when it has fallen furthest behind. Treat an <c>OUT_OF_RANGE</c>
/// parse failure as a re-baseline trigger (re-prime from <see cref="CurrentAsync"/>,
/// resume from that snapshot's <c>NextSequence</c>) just like a detected gap.
/// </para>
/// </remarks>
/// <param name="expectedFrom">The sequence the caller expected this chunk to start at.</param>
/// <param name="chunk">The chunk the Agent answered with.</param>
/// <returns>
/// <c>true</c> when observations were lost and the caller must re-baseline via
/// <see cref="CurrentAsync"/> rather than trusting an incremental update.
/// </returns>
static bool IsSequenceGap(long expectedFrom, MTConnectStreamsResult chunk)
{
ArgumentNullException.ThrowIfNull(chunk);
return chunk.FirstSequence > expectedFrom;
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs
+199 -90
View File
@@ -1,6 +1,7 @@
using System.Globalization;
using System.Runtime.CompilerServices;
using System.Text;
using Microsoft.Extensions.Logging;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
@@ -25,7 +26,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// a long poll that is <i>supposed</i> to outlive that bound indefinitely. Raising the one
/// timeout to suit the stream would un-bound the unary deadline — precisely what arch-review
/// R2-01 says must never happen — so the streaming leg gets its own client with
/// <see cref="Timeout.InfiniteTimeSpan"/> and is bounded differently instead (below).
/// <see cref="Timeout.InfiniteTimeSpan"/> and is bounded differently instead (below). In
/// production each client owns its own handler and connection pool; in tests both are built
/// over one injected handler, which the caller owns.
/// </para>
/// <para>
/// <b>Every call is deadline-bounded, but not all by the same mechanism.</b> The unary legs
@@ -41,61 +44,94 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// and the watchdog is what closes it here.
/// </para>
/// <para>
/// A non-positive <c>RequestTimeoutMs</c> is rejected at construction so an
/// operator-authored <c>0</c> can never come to mean "wait forever" (arch-review 01/S-6).
/// <b>Every operator-authorable duration/count is rejected at construction if non-positive.</b>
/// A <c>0</c> must never come to mean "wait forever" (arch-review 01/S-6), and the three
/// <c>/sample</c> query knobs additionally reach the wire: an Agent answers
/// <c>heartbeat=0</c> with an <c>MTConnectError</c> under HTTP 200, which would otherwise
/// surface as a parse failure that never names the offending config key.
/// </para>
/// <para>
/// <b>Disposal is safe against an in-flight enumeration.</b> Task 9 re-initializes the
/// driver by disposing and rebuilding this client, which can happen while a pump is still
/// enumerating <see cref="SampleAsync"/>. Rather than letting the in-flight read fault with
/// an unclassified <see cref="ObjectDisposedException"/> or <see cref="IOException"/> that no
/// caller is written to expect, every request runs under a token linked to an internal
/// dispose token, so disposal surfaces as an ordinary
/// <see cref="OperationCanceledException"/>.
/// </para>
/// <para>
/// This class does not swallow failures into an empty model: an unreachable Agent, a non-2xx
/// status, or an unparseable body all throw. <c>InitializeAsync</c> (Task 9) is what catches
/// that and moves the driver to Faulted.
/// status, an unparseable body, and <b>the end of a <c>/sample</c> stream</b> all throw. See
/// <see cref="MTConnectStreamException"/> for why the last of those is an exception rather
/// than a normal end of enumeration. <c>InitializeAsync</c> (Task 9) is what catches these
/// and moves the driver to Faulted.
/// </para>
/// </remarks>
public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
{
private readonly HttpClient _http;
private readonly HttpClient _streamHttp;
private readonly CancellationTokenSource _disposeCts = new();
private readonly ILogger? _logger;
private readonly Uri _probeUri;
private readonly Uri _currentUri;
private readonly string _sampleUriBase;
private readonly string _sampleQuerySuffix;
private readonly TimeSpan _requestTimeout;
private readonly TimeSpan _streamIdleTimeout;
private bool _disposed;
/// <summary>Creates a client for the Agent described by <paramref name="options"/>.</summary>
/// <param name="options">The driver options carrying the Agent URI, device scope, and per-call deadline.</param>
/// <param name="logger">
/// Optional logger. Used only for conditions an operator needs to see but which are not
/// failures — today, the degraded <c>/sample</c> framing mode.
/// </param>
/// <exception cref="ArgumentException"><see cref="MTConnectDriverOptions.AgentUri"/> is missing or is not an absolute HTTP(S) URI.</exception>
/// <exception cref="ArgumentOutOfRangeException"><see cref="MTConnectDriverOptions.RequestTimeoutMs"/> is not positive.</exception>
public MTConnectAgentClient(MTConnectDriverOptions options)
: this(options, handler: null)
/// <exception cref="ArgumentOutOfRangeException">
/// Any of <see cref="MTConnectDriverOptions.RequestTimeoutMs"/>,
/// <see cref="MTConnectDriverOptions.HeartbeatMs"/>,
/// <see cref="MTConnectDriverOptions.SampleIntervalMs"/> or
/// <see cref="MTConnectDriverOptions.SampleCount"/> is not positive.
/// </exception>
public MTConnectAgentClient(MTConnectDriverOptions options, ILogger? logger = null)
: this(options, handler: null, logger)
{
}
/// <summary>
/// Test seam: as <see cref="MTConnectAgentClient(MTConnectDriverOptions)"/>, but sends through
/// the supplied handler so the request/response round trip can be exercised with no socket.
/// The caller retains ownership of <paramref name="handler"/>.
/// Test seam: as <see cref="MTConnectAgentClient(MTConnectDriverOptions, ILogger)"/>, but
/// sends through the supplied handler so the request/response round trip can be exercised
/// with no socket. The caller retains ownership of <paramref name="handler"/>.
/// </summary>
internal MTConnectAgentClient(MTConnectDriverOptions options, HttpMessageHandler? handler)
internal MTConnectAgentClient(MTConnectDriverOptions options, HttpMessageHandler? handler, ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
if (options.RequestTimeoutMs <= 0)
{
throw new ArgumentOutOfRangeException(
nameof(options),
RequirePositive(
options.RequestTimeoutMs,
$"{nameof(MTConnectDriverOptions.RequestTimeoutMs)} must be positive; a non-positive per-call deadline would let a frozen Agent wedge the driver indefinitely.");
}
nameof(MTConnectDriverOptions.RequestTimeoutMs),
"a non-positive per-call deadline would let a frozen Agent wedge the driver indefinitely");
RequirePositive(
options.HeartbeatMs,
nameof(MTConnectDriverOptions.HeartbeatMs),
"it is sent to the Agent as 'heartbeat', which an Agent rejects with an MTConnectError under HTTP 200, and it is what bounds the /sample watchdog — a quiet connection would otherwise be indistinguishable from a dead one");
RequirePositive(
options.SampleIntervalMs,
nameof(MTConnectDriverOptions.SampleIntervalMs),
"it is sent to the Agent as 'interval'; zero would spin the streaming connection into a busy-loop");
RequirePositive(
options.SampleCount,
nameof(MTConnectDriverOptions.SampleCount),
"it is sent to the Agent as 'count'; a non-positive count asks the Agent for no observations");
_logger = logger;
_requestTimeout = TimeSpan.FromMilliseconds(options.RequestTimeoutMs);
// Two missed heartbeats plus one request timeout of network slack. Deriving the window from
// HeartbeatMs is what makes it a liveness check rather than a throughput assumption: a busy
// Agent and an idle one both emit at least one chunk per heartbeat. A non-positive
// HeartbeatMs is clamped away rather than rejected — the window then collapses to
// RequestTimeoutMs, which is still strictly positive, so the read can never be unbounded.
_streamIdleTimeout = TimeSpan.FromMilliseconds(
(2L * Math.Max(0, options.HeartbeatMs)) + options.RequestTimeoutMs);
// Agent and an idle one both emit at least one chunk per heartbeat.
_streamIdleTimeout = TimeSpan.FromMilliseconds((2L * options.HeartbeatMs) + options.RequestTimeoutMs);
_probeUri = BuildRequestUri(options, "probe");
_currentUri = BuildRequestUri(options, "current");
@@ -114,29 +150,11 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
_streamHttp.Timeout = Timeout.InfiniteTimeSpan;
}
/// <summary>
/// Has the Agent's ring buffer overflowed past what the caller asked for? True when the
/// chunk's oldest retained sequence is <i>strictly</i> newer than the sequence the caller
/// requested, which means observations between the two were evicted before the caller could
/// read them. Equality is not a gap: <c>firstSequence == requestedFrom</c> is the ordinary
/// contiguous case where the Agent simply had nothing older to send.
/// </summary>
/// <param name="requestedFrom">The <c>from</c> sequence the caller passed to <see cref="SampleAsync"/>.</param>
/// <param name="chunk">The chunk the Agent answered with.</param>
/// <returns>
/// <c>true</c> when observations were lost and the caller must re-baseline via
/// <see cref="CurrentAsync"/> rather than trusting an incremental update.
/// </returns>
public static bool IsSequenceGap(long requestedFrom, MTConnectStreamsResult chunk)
{
ArgumentNullException.ThrowIfNull(chunk);
return chunk.FirstSequence > requestedFrom;
}
/// <inheritdoc/>
public async Task<MTConnectProbeModel> ProbeAsync(CancellationToken ct)
{
ObjectDisposedException.ThrowIf(_disposed, this);
// ResponseContentRead (the default) buffers the whole body *under* this awaited, token-bound
// call, so the parse below reads from memory — no network read can outlive the deadline.
using var response = await SendAsync(_probeUri, ct).ConfigureAwait(false);
@@ -149,6 +167,8 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
/// <inheritdoc/>
public async Task<MTConnectStreamsResult> CurrentAsync(CancellationToken ct)
{
ObjectDisposedException.ThrowIf(_disposed, this);
using var response = await SendAsync(_currentUri, ct).ConfigureAwait(false);
await using var body = await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false);
@@ -161,34 +181,50 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
long from,
[EnumeratorCancellation] CancellationToken ct)
{
ObjectDisposedException.ThrowIf(_disposed, this);
var uri = BuildSampleUri(from);
// Linked to the caller's token AND to disposal, so tearing this client down mid-enumeration
// surfaces as a cancellation rather than as a raw ObjectDisposedException from the socket.
using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(ct, _disposeCts.Token);
var live = lifetime.Token;
using var request = new HttpRequestMessage(HttpMethod.Get, uri);
// ResponseHeadersRead, not the default ResponseContentRead: buffering a
// multipart/x-mixed-replace body that is designed never to end would hang here forever.
using var response = await SendStreamRequestAsync(request, uri, ct).ConfigureAwait(false);
using var response = await SendStreamRequestAsync(request, uri, lifetime).ConfigureAwait(false);
await using var body = await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false);
await using var body = await response.Content.ReadAsStreamAsync(live).ConfigureAwait(false);
var boundary = ResolveMultipartBoundary(response);
var boundary = ResolveMultipartBoundary(response, uri);
if (boundary is null)
{
// Not a multipart stream: the Agent answered with a single document. This is the shape
// an MTConnectError takes — Agents serve one under HTTP 200 — so the parse below is what
// surfaces the Agent's own error text rather than an empty, successful-looking stream.
yield return MTConnectStreamsParser.Parse(await ReadWholeBodyAsync(body, uri, ct).ConfigureAwait(false));
// Not a multipart stream: the Agent answered with a single document, which means it
// ignored the streaming request entirely. Parse it first — an MTConnectError arrives in
// exactly this shape (text/xml under HTTP 200) and the Agent's own error text is far
// more useful than ours — then report the misconfiguration. The parsed snapshot is
// deliberately NOT yielded: one document followed by a healthy-looking finished stream
// is how a dead data path disguises itself as a working one (#485).
var single = await ReadWholeBodyAsync(body, uri, lifetime).ConfigureAwait(false);
_ = MTConnectStreamsParser.Parse(single);
yield break;
throw new MTConnectStreamNotSupportedException(
$"MTConnect Agent answered the /sample request '{uri}' with a single '{response.Content.Headers.ContentType?.MediaType ?? "unknown"}' document instead of a multipart/x-mixed-replace stream. It parsed as a valid MTConnectStreams document but will never stream, so no subscription is possible; check that the URI addresses an MTConnect Agent directly and is not fronted by a proxy that buffers streaming responses.");
}
var reader = new MultipartStreamReader(body, boundary, _streamIdleTimeout, uri);
var reader = new MultipartStreamReader(body, boundary, _streamIdleTimeout, uri, _logger);
var delivered = 0L;
while (true)
{
var part = await reader.ReadNextPartAsync(ct).ConfigureAwait(false);
var part = await reader.ReadNextPartAsync(live).ConfigureAwait(false);
if (part is null)
{
yield break;
// The caller cancelling is the ONE contracted way out; anything else is reported.
live.ThrowIfCancellationRequested();
throw new MTConnectStreamEndedException(reader.EndReason, uri.AbsoluteUri, delivered);
}
// A frame carrying nothing but whitespace is transport filler, not a document. The
@@ -200,6 +236,8 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
continue;
}
delivered++;
yield return MTConnectStreamsParser.Parse(part);
}
}
@@ -207,8 +245,30 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
/// <inheritdoc/>
public void Dispose()
{
if (_disposed)
{
return;
}
_disposed = true;
// Cancel BEFORE disposing the clients so an in-flight read observes cancellation rather
// than a disposed handler.
_disposeCts.Cancel();
_http.Dispose();
_streamHttp.Dispose();
_disposeCts.Dispose();
}
private static void RequirePositive(int value, string optionName, string because)
{
if (value <= 0)
{
throw new ArgumentOutOfRangeException(
nameof(MTConnectDriverOptions),
value,
$"{optionName} must be positive; {because}.");
}
}
private Uri BuildSampleUri(long from) =>
@@ -217,7 +277,8 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
private async Task<HttpResponseMessage> SendAsync(Uri uri, CancellationToken ct)
{
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(ct, _disposeCts.Token);
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(lifetime.Token);
deadline.CancelAfter(_requestTimeout);
HttpResponseMessage response;
@@ -225,11 +286,11 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
{
response = await _http.GetAsync(uri, deadline.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
catch (OperationCanceledException) when (!lifetime.IsCancellationRequested)
{
// The caller did not cancel, so this is our own deadline (or HttpClient.Timeout) firing.
// Reported as a TimeoutException because a bare "A task was canceled" tells an operator
// nothing about which Agent stopped answering.
// Neither the caller nor disposal cancelled, so this is our own deadline (or
// HttpClient.Timeout) firing. Reported as a TimeoutException because a bare "A task was
// canceled" tells an operator nothing about which Agent stopped answering.
throw TimedOut(uri);
}
@@ -243,9 +304,9 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
/// this phase is bounded by the unary deadline; the body is bounded by the watchdog.
/// </summary>
private async Task<HttpResponseMessage> SendStreamRequestAsync(
HttpRequestMessage request, Uri uri, CancellationToken ct)
HttpRequestMessage request, Uri uri, CancellationTokenSource lifetime)
{
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(lifetime.Token);
deadline.CancelAfter(_requestTimeout);
HttpResponseMessage response;
@@ -255,7 +316,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, deadline.Token)
.ConfigureAwait(false);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
catch (OperationCanceledException) when (!lifetime.IsCancellationRequested)
{
throw TimedOut(uri);
}
@@ -266,9 +327,9 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
}
/// <summary>Reads a non-multipart streaming body whole, under the unary deadline.</summary>
private async Task<string> ReadWholeBodyAsync(Stream body, Uri uri, CancellationToken ct)
private async Task<string> ReadWholeBodyAsync(Stream body, Uri uri, CancellationTokenSource lifetime)
{
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(lifetime.Token);
deadline.CancelAfter(_requestTimeout);
using var reader = new StreamReader(body, Encoding.UTF8);
@@ -276,7 +337,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
{
return await reader.ReadToEndAsync(deadline.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
catch (OperationCanceledException) when (!lifetime.IsCancellationRequested)
{
throw TimedOut(uri);
}
@@ -289,7 +350,14 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
/// The <c>boundary</c> parameter of a <c>multipart/*</c> response, or <c>null</c> when the
/// Agent answered with a single (non-multipart) document.
/// </summary>
private static string? ResolveMultipartBoundary(HttpResponseMessage response)
/// <exception cref="MTConnectStreamNotSupportedException">
/// The response claims to be <c>multipart/*</c> but carries no <c>boundary</c> parameter, so
/// there is nothing to frame it by. Failed fast and named, because the alternative — falling
/// through to whole-body reading of a body that never ends — surfaces much later as a bare
/// watchdog <see cref="TimeoutException"/> that points an operator at the network instead of
/// at the malformed header.
/// </exception>
private static string? ResolveMultipartBoundary(HttpResponseMessage response, Uri uri)
{
var contentType = response.Content.Headers.ContentType;
if (contentType?.MediaType is null ||
@@ -302,7 +370,13 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
.FirstOrDefault(p => string.Equals(p.Name, "boundary", StringComparison.OrdinalIgnoreCase))?.Value?
.Trim('"');
return string.IsNullOrEmpty(boundary) ? null : boundary;
if (string.IsNullOrEmpty(boundary))
{
throw new MTConnectStreamNotSupportedException(
$"MTConnect Agent answered the /sample request '{uri}' with Content-Type '{contentType.MediaType}' but no 'boundary' parameter, so the multipart stream cannot be framed.");
}
return boundary;
}
private static Uri BuildRequestUri(MTConnectDriverOptions options, string request)
@@ -335,18 +409,22 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
/// <remarks>
/// <para>
/// <b>Why hand-rolled.</b> ASP.NET Core's <c>MultipartReader</c> is not in this
/// dependency set, and TrakHound's
/// <c>MTConnectHttpClientStream</c> — the one candidate the plan left open — takes a URL
/// and dials its own socket, so it can neither be fed an existing response stream nor be
/// exercised without a listener. See the Task 0 correction in the plan.
/// dependency set, and TrakHound's <c>MTConnectHttpClientStream</c> — the one candidate
/// the plan left open — takes a URL and dials its own socket, so it can neither be fed an
/// existing response stream nor be exercised without a listener. See the Task 0
/// correction in the plan.
/// </para>
/// <para>
/// <b>Two framing paths, and the first one matters for latency.</b> When the part carries
/// a <c>Content-length</c> header — every Agent in the wild writes one — the body is read
/// by length and the chunk is emitted the instant it is complete. Without it there is no
/// way to know a part has ended except by seeing the <i>next</i> boundary, so that path
/// necessarily emits each chunk one chunk late; it exists as a correctness fallback, not
/// as the expected path.
/// necessarily emits each chunk one chunk late: with the default 10 s heartbeat a value
/// can be up to a heartbeat stale on a driver built for low-latency subscribe. It is a
/// correctness fallback, not the expected path, and it logs a one-shot warning when it
/// engages so the degradation is visible rather than merely slow. (It cannot strand the
/// final part indefinitely — a clean end of stream returns the buffered tail — though a
/// watchdog timeout on a stalled Agent does discard it.)
/// </para>
/// <para>
/// <b>Every read is watchdog-bounded.</b> A silent peer faults the stream rather than
@@ -354,7 +432,8 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
/// malfunctioning Agent cannot grow it without limit.
/// </para>
/// </remarks>
private sealed class MultipartStreamReader(Stream stream, string boundary, TimeSpan idleTimeout, Uri uri)
private sealed class MultipartStreamReader(
Stream stream, string boundary, TimeSpan idleTimeout, Uri uri, ILogger? logger)
{
private const int MaxBufferBytes = 32 * 1024 * 1024;
@@ -365,10 +444,14 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
private byte[] _buffer = new byte[16 * 1024];
private int _length;
private bool _finished;
private bool _warnedNoContentLength;
/// <summary>How the stream ended. Only meaningful once a read has returned <c>null</c>.</summary>
public MTConnectStreamEndReason EndReason { get; private set; } = MTConnectStreamEndReason.ConnectionClosed;
/// <summary>
/// Reads the next part's payload, or <c>null</c> once the closing delimiter is seen or
/// the Agent closes the connection.
/// the Agent closes the connection — see <see cref="EndReason"/> for which.
/// </summary>
public async Task<string?> ReadNextPartAsync(CancellationToken ct)
{
@@ -380,42 +463,65 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
if (!await SkipThroughDelimiterAsync(ct).ConfigureAwait(false) ||
!await EnsureAsync(2, ct).ConfigureAwait(false))
{
_finished = true;
return null;
return Finish(MTConnectStreamEndReason.ConnectionClosed);
}
// "--boundary--" closes the stream.
if (_buffer[0] == (byte)'-' && _buffer[1] == (byte)'-')
{
_finished = true;
return null;
return Finish(MTConnectStreamEndReason.ClosingBoundary);
}
var headerEnd = await FindHeaderEndAsync(ct).ConfigureAwait(false);
if (headerEnd < 0)
{
_finished = true;
return null;
return Finish(MTConnectStreamEndReason.ConnectionClosed);
}
var headers = Encoding.ASCII.GetString(_buffer, 0, headerEnd);
Consume(headerEnd);
var body = ParseContentLength(headers) is { } declared
? await ReadByLengthAsync(declared, ct).ConfigureAwait(false)
: await ReadToNextDelimiterAsync(ct).ConfigureAwait(false);
byte[]? body;
if (ParseContentLength(headers) is { } declared)
{
body = await ReadByLengthAsync(declared, ct).ConfigureAwait(false);
}
else
{
WarnOnceAboutMissingContentLength();
body = await ReadToNextDelimiterAsync(ct).ConfigureAwait(false);
}
return body is null ? null : Encoding.UTF8.GetString(body);
}
private void WarnOnceAboutMissingContentLength()
{
if (_warnedNoContentLength)
{
return;
}
_warnedNoContentLength = true;
logger?.LogWarning(
"MTConnect /sample stream from '{AgentUri}' sends parts with no Content-length header. Falling back to boundary-scan framing, which cannot emit a chunk until the NEXT chunk begins, so every observation is delayed by up to one Agent heartbeat. Values will read stale; this is the Agent's framing, not a driver fault.",
uri);
}
private string? Finish(MTConnectStreamEndReason reason)
{
_finished = true;
EndReason = reason;
return null;
}
private async Task<byte[]?> ReadByLengthAsync(int declared, CancellationToken ct)
{
if (!await EnsureAsync(declared, ct).ConfigureAwait(false))
{
_finished = true;
Finish(MTConnectStreamEndReason.ConnectionClosed);
return null;
}
@@ -431,10 +537,11 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
var next = await FindDelimiterAsync(ct).ConfigureAwait(false);
if (next < 0)
{
// EOF with no closing delimiter: whatever is buffered is the last part.
// EOF with no closing delimiter: whatever is buffered is the last part. Returned
// rather than dropped, so a clean end of stream cannot strand a chunk.
var tail = _buffer[.._length];
_length = 0;
_finished = true;
Finish(MTConnectStreamEndReason.ConnectionClosed);
return tail;
}
@@ -477,6 +584,8 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
var searchedTo = 0;
while (true)
{
// Restart the scan far enough back that a delimiter split across two reads is still
// found — without the overlap, a boundary straddling a packet boundary is invisible.
var index = IndexOf(_delimiter, Math.Max(0, searchedTo - (_delimiter.Length - 1)));
if (index >= 0)
{
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs
+24 -1
View File
@@ -136,10 +136,33 @@ public sealed record MTConnectStreamsResult(
/// The observation's Agent-reported timestamp, normalized to UTC (<see cref="DateTime.Kind"/>
/// is always <see cref="DateTimeKind.Utc"/>). Becomes the OPC UA variable's SourceTimestamp.
/// </param>
/// <param name="IsStructured">
/// <c>true</c> when the Agent carried this observation's real content in <b>child elements</b>
/// rather than as text — an MTConnect 2.0 <c>DATA_SET</c> / <c>TABLE</c> observation, whose
/// content is a list of <c>&lt;Entry key="…"&gt;</c> elements.
/// <para>
/// <b>Why the flag exists, and why only the parser can set it.</b> <see cref="Value"/> is
/// the element's concatenated descendant text, so a two-entry data set reads as the single
/// token <c>"12"</c> — keys discarded, values run together. The observation index cannot
/// detect that after the fact: neither this record nor the tag definition carries the
/// DataItem's <c>representation</c>, and no value-shape heuristic can work, because
/// concatenated entries are indistinguishable from a legitimate space-bearing EVENT such as
/// a <c>Message</c> reading <c>"Coolant level low"</c>. The one reliable discriminator —
/// that the observation element had element children — exists only while the XML is still
/// XML.
/// </para>
/// <para>
/// Deliberately a <b>neutral fact about the wire shape</b>, not a status: mapping it to a
/// quality code is the observation index's job (it codes these
/// <c>BadNotSupported</c> rather than publishing concatenated noise as Good). Defaults to
/// <c>false</c>, the shape of every ordinary scalar observation.
/// </para>
/// </param>
public sealed record MTConnectObservation(
string DataItemId,
string Value,
DateTime TimestampUtc);
DateTime TimestampUtc,
bool IsStructured = false);
/// <summary>
/// Recursive helpers over the <see cref="IMTConnectComponentContainer"/> device/component tree.
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs
+33 -98
View File
@@ -1,4 +1,3 @@
using System.Globalization;
using System.Xml;
using System.Xml.Linq;
@@ -30,12 +29,18 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// literal <c>&lt;Component&gt;</c> element — so <i>every</i> element child of
/// <c>&lt;Components&gt;</c> is a component.
/// (3) The document is XML-namespaced and the namespace URI carries the MTConnect version
/// (<c>urn:mtconnect.org:MTConnectDevices:1.3</c> … <c>:2.0</c>). Elements are therefore
/// matched on <see cref="XName.LocalName"/> and the namespace is ignored entirely — nothing
/// version-specific is hard-coded, and a vendor-extension component in its own namespace
/// (whose standard child elements still inherit the default MTConnect namespace) is not
/// silently dropped. Attributes, by contrast, are read unqualified so a prefixed attribute
/// (e.g. <c>xsi:type</c>) can never be mistaken for a DataItem's <c>type</c>.
/// (<c>urn:mtconnect.org:MTConnectDevices:1.3</c> … <c>:2.0</c>). Element matching and
/// attribute reading therefore go through <see cref="MTConnectXml"/>, whose two rules
/// (match on local name, read attributes unqualified) are shared verbatim with
/// <see cref="MTConnectStreamsParser"/> — see that type for why each rule exists.
/// </para>
/// <para>
/// <b>One deliberate divergence from the streams parser: no BOM/whitespace trim here.</b>
/// A <c>/probe</c> body arrives whole from <c>HttpContent</c>, so it begins exactly where
/// the Agent's document begins. A <c>/sample</c> chunk, by contrast, is lifted out of a
/// multipart frame and can carry a byte-order mark or the framing's own line break ahead of
/// the XML declaration, which <c>XmlReader</c> rejects — hence the trim there and not here.
/// This asymmetry is intentional, not an oversight.
/// </para>
/// <para>
/// <b>Failure posture.</b> Anything that is not a well-formed MTConnect device model throws
@@ -54,6 +59,9 @@ internal static class MTConnectProbeParser
/// </summary>
private const int MaxComponentDepth = 64;
/// <summary>How <see cref="MTConnectXml"/>'s shared error messages name this document.</summary>
private const string Subject = "/probe response";
/// <summary>Parses a <c>/probe</c> response held as a string.</summary>
/// <param name="xml">The raw <c>MTConnectDevices</c> XML document.</param>
/// <exception cref="InvalidDataException">
@@ -106,12 +114,12 @@ internal static class MTConnectProbeParser
var root = document.Root
?? throw new InvalidDataException("MTConnect /probe response has no root element.");
if (!IsNamed(root, "MTConnectDevices"))
if (!MTConnectXml.IsNamed(root, "MTConnectDevices"))
{
throw new InvalidDataException(DescribeUnexpectedRoot(root));
throw new InvalidDataException(MTConnectXml.DescribeUnexpectedRoot(root, "MTConnectDevices", Subject));
}
var devicesContainer = ChildrenNamed(root, "Devices").FirstOrDefault()
var devicesContainer = MTConnectXml.ChildrenNamed(root, "Devices").FirstOrDefault()
?? throw new InvalidDataException(
"MTConnect /probe response has no <Devices> element; it does not describe a device model.");
@@ -127,15 +135,15 @@ internal static class MTConnectProbeParser
private static MTConnectDevice ReadDevice(XElement element) =>
new(
RequiredAttribute(element, "id", "Device"),
OptionalAttribute(element, "name"),
MTConnectXml.RequiredAttribute(element, "id", "Device", Subject),
MTConnectXml.OptionalAttribute(element, "name"),
ReadComponents(element, depth: 1),
ReadDataItems(element));
private static MTConnectComponent ReadComponent(XElement element, int depth) =>
new(
RequiredAttribute(element, "id", $"Component <{element.Name.LocalName}>"),
OptionalAttribute(element, "name"),
MTConnectXml.RequiredAttribute(element, "id", $"Component <{element.Name.LocalName}>", Subject),
MTConnectXml.OptionalAttribute(element, "name"),
ReadComponents(element, depth + 1),
ReadDataItems(element));
@@ -151,104 +159,31 @@ internal static class MTConnectProbeParser
$"MTConnect /probe response nests components more than {MaxComponentDepth} levels deep; refusing to recurse further.");
}
return ChildrenNamed(container, "Components")
return MTConnectXml.ChildrenNamed(container, "Components")
.SelectMany(components => components.Elements())
.Select(element => ReadComponent(element, depth))
.ToList();
}
private static IReadOnlyList<MTConnectDataItem> ReadDataItems(XElement container) =>
ChildrenNamed(container, "DataItems")
.SelectMany(dataItems => ChildrenNamed(dataItems, "DataItem"))
MTConnectXml.ChildrenNamed(container, "DataItems")
.SelectMany(dataItems => MTConnectXml.ChildrenNamed(dataItems, "DataItem"))
.Select(ReadDataItem)
.ToList();
private static MTConnectDataItem ReadDataItem(XElement element)
{
var id = RequiredAttribute(element, "id", "DataItem");
var id = MTConnectXml.RequiredAttribute(element, "id", "DataItem", Subject);
return new MTConnectDataItem(
id,
OptionalAttribute(element, "name"),
RequiredAttribute(element, "category", $"DataItem '{id}'"),
RequiredAttribute(element, "type", $"DataItem '{id}'"),
OptionalAttribute(element, "subType"),
OptionalAttribute(element, "units"),
OptionalAttribute(element, "representation"),
OptionalIntAttribute(element, "sampleCount", id));
MTConnectXml.OptionalAttribute(element, "name"),
MTConnectXml.RequiredAttribute(element, "category", $"DataItem '{id}'", Subject),
MTConnectXml.RequiredAttribute(element, "type", $"DataItem '{id}'", Subject),
MTConnectXml.OptionalAttribute(element, "subType"),
MTConnectXml.OptionalAttribute(element, "units"),
MTConnectXml.OptionalAttribute(element, "representation"),
MTConnectXml.OptionalIntAttribute(element, "sampleCount", $"DataItem '{id}'", Subject));
}
private static bool IsNamed(XElement element, string localName) =>
string.Equals(element.Name.LocalName, localName, StringComparison.Ordinal);
private static IEnumerable<XElement> ChildrenNamed(XElement parent, string localName) =>
parent.Elements().Where(child => IsNamed(child, localName));
private static string RequiredAttribute(XElement element, string name, string owner)
{
var value = OptionalAttribute(element, name);
return value ?? throw new InvalidDataException(
$"MTConnect /probe response is malformed: {owner} has no '{name}' attribute.");
}
/// <summary>
/// Reads an unqualified attribute, normalizing a present-but-empty value to <c>null</c>.
/// Only unqualified attributes are considered, so a namespace-prefixed attribute such as
/// <c>xsi:type</c> can never be read as a DataItem's <c>type</c>.
/// </summary>
private static string? OptionalAttribute(XElement element, string name)
{
var value = element.Attribute(name)?.Value;
return string.IsNullOrWhiteSpace(value) ? null : value;
}
private static int? OptionalIntAttribute(XElement element, string name, string dataItemId)
{
var raw = OptionalAttribute(element, name);
if (raw is null)
{
return null;
}
if (!int.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value))
{
throw new InvalidDataException(
$"MTConnect /probe response is malformed: DataItem '{dataItemId}' has a non-numeric '{name}' attribute ('{raw}').");
}
return value;
}
/// <summary>
/// Builds the message for a root element that is not <c>MTConnectDevices</c>, lifting the
/// Agent's own error text when the response is an <c>MTConnectError</c> document (which
/// Agents return under HTTP 200 for, e.g., an unknown device name).
/// </summary>
private static string DescribeUnexpectedRoot(XElement root)
{
var message =
$"MTConnect /probe response root element is <{root.Name.LocalName}>, expected <MTConnectDevices>.";
if (!IsNamed(root, "MTConnectError"))
{
return message;
}
var errors = root.Descendants()
.Where(e => IsNamed(e, "Error"))
.Select(e =>
{
var code = OptionalAttribute(e, "errorCode");
var text = e.Value.Trim();
return code is null ? text : $"{code}: {text}";
})
.Where(text => text.Length > 0)
.ToList();
return errors.Count == 0
? message
: $"{message} The Agent reported: {string.Join("; ", errors)}";
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs
+161
View File
@@ -0,0 +1,161 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// <summary>
/// Base type for every way an Agent <c>/sample</c> long poll can stop delivering chunks for a
/// reason that is <b>not</b> the caller cancelling it.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why these are exceptions rather than a normal end of enumeration.</b>
/// <see cref="IMTConnectAgentClient.SampleAsync"/> is contracted to yield indefinitely until
/// its token is cancelled, so a pump written to that contract has no reason to handle normal
/// completion — it would either drop the subscription silently or hot-loop reconnecting on
/// an enumerator that ends immediately. Returning normally would make "the data path
/// stopped" indistinguishable from "the data path is idle", which is the
/// quiet-successful-looking-termination shape that produced this repo's #485 defect. Every
/// non-cancellation end is therefore loud, typed, and carries the Agent URI.
/// </para>
/// <para>
/// Catch this base type to handle "the stream is over" uniformly; catch the two derived
/// types to distinguish a <i>transient</i> end (the Agent closed a connection — reconnect)
/// from a <i>configuration</i> error (the Agent will never stream to this request — retrying
/// is pointless until an operator changes something).
/// </para>
/// </remarks>
public abstract class MTConnectStreamException : Exception
{
/// <summary>Creates the exception with no message.</summary>
protected MTConnectStreamException()
{
}
/// <summary>Creates the exception with a message.</summary>
/// <param name="message">The operator-facing description.</param>
protected MTConnectStreamException(string message)
: base(message)
{
}
/// <summary>Creates the exception with a message and an inner cause.</summary>
/// <param name="message">The operator-facing description.</param>
/// <param name="innerException">The underlying cause.</param>
protected MTConnectStreamException(string message, Exception innerException)
: base(message, innerException)
{
}
}
/// <summary>How an Agent <c>/sample</c> stream ended.</summary>
public enum MTConnectStreamEndReason
{
/// <summary>
/// The transport closed mid-stream — the connection dropped, the Agent restarted, or a proxy
/// timed the connection out. Ordinary and transient: reconnect from the last
/// <see cref="MTConnectStreamsResult.NextSequence"/>.
/// </summary>
ConnectionClosed = 0,
/// <summary>
/// The Agent wrote the closing <c>--boundary--</c> delimiter, ending the multipart response
/// deliberately. Usually means the request was bounded (an Agent honouring a <c>count</c>
/// that exhausted, or an administrative stop) rather than a fault. Also reconnectable, but
/// worth distinguishing in logs: a stream that keeps closing cleanly points at the request's
/// parameters, not at the network.
/// </summary>
ClosingBoundary = 1
}
/// <summary>
/// The Agent's <c>/sample</c> stream ended without the caller cancelling it. See
/// <see cref="MTConnectStreamException"/> for why this is not a normal end of enumeration.
/// </summary>
public sealed class MTConnectStreamEndedException : MTConnectStreamException
{
/// <summary>Creates the exception for a stream that ended for <paramref name="reason"/>.</summary>
/// <param name="reason">How the stream ended.</param>
/// <param name="agentUri">The <c>/sample</c> request URI whose stream ended.</param>
/// <param name="chunksDelivered">How many chunks were yielded before the end.</param>
public MTConnectStreamEndedException(MTConnectStreamEndReason reason, string agentUri, long chunksDelivered)
: base($"MTConnect /sample stream from '{agentUri}' ended after {chunksDelivered} chunk(s) without the caller cancelling it ({Describe(reason)}). The stream is contracted to run until cancelled, so this is reported rather than returned; resume from the last chunk's nextSequence.")
{
Reason = reason;
AgentUri = agentUri;
ChunksDelivered = chunksDelivered;
}
/// <summary>Creates the exception with a message.</summary>
/// <param name="message">The operator-facing description.</param>
public MTConnectStreamEndedException(string message)
: base(message)
{
AgentUri = string.Empty;
}
/// <summary>Creates the exception with a message and an inner cause.</summary>
/// <param name="message">The operator-facing description.</param>
/// <param name="innerException">The underlying cause.</param>
public MTConnectStreamEndedException(string message, Exception innerException)
: base(message, innerException)
{
AgentUri = string.Empty;
}
/// <summary>Creates the exception with no message.</summary>
public MTConnectStreamEndedException()
{
AgentUri = string.Empty;
}
/// <summary>How the stream ended.</summary>
public MTConnectStreamEndReason Reason { get; }
/// <summary>The <c>/sample</c> request URI whose stream ended.</summary>
public string AgentUri { get; }
/// <summary>How many chunks were yielded before the stream ended.</summary>
public long ChunksDelivered { get; }
private static string Describe(MTConnectStreamEndReason reason) =>
reason switch
{
MTConnectStreamEndReason.ClosingBoundary => "the Agent wrote the closing multipart boundary",
_ => "the connection closed mid-stream"
};
}
/// <summary>
/// The Agent answered a <c>/sample</c> request with something that cannot be consumed as a
/// stream at all — a single non-multipart document, or a <c>multipart/*</c> response with no
/// <c>boundary</c> parameter to frame it by.
/// </summary>
/// <remarks>
/// This is a <b>configuration</b> error, not a transient one, and is deliberately typed apart
/// from <see cref="MTConnectStreamEndedException"/>: reconnecting will produce the identical
/// response forever. The usual cause is an endpoint that is not an MTConnect Agent (a reverse
/// proxy, a load balancer's health page, or an Agent fronted by something that buffers and
/// collapses <c>multipart/x-mixed-replace</c>). Reported instead of yielding the one document
/// the Agent did return, because a single snapshot followed by a healthy-looking finished
/// stream is exactly how a dead data path disguises itself as a working one.
/// </remarks>
public sealed class MTConnectStreamNotSupportedException : MTConnectStreamException
{
/// <summary>Creates the exception with a message.</summary>
/// <param name="message">The operator-facing description.</param>
public MTConnectStreamNotSupportedException(string message)
: base(message)
{
}
/// <summary>Creates the exception with a message and an inner cause.</summary>
/// <param name="message">The operator-facing description.</param>
/// <param name="innerException">The underlying cause.</param>
public MTConnectStreamNotSupportedException(string message, Exception innerException)
: base(message, innerException)
{
}
/// <summary>Creates the exception with no message.</summary>
public MTConnectStreamNotSupportedException()
{
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs
+26 -82
View File
@@ -79,6 +79,12 @@ internal static class MTConnectStreamsParser
private const string ConditionContainer = "Condition";
/// <summary>How <see cref="MTConnectXml"/>'s shared error messages name this document.</summary>
private const string Subject = "streams response";
/// <summary>How those messages name the header element.</summary>
private const string HeaderOwner = "the <Header>";
/// <summary>
/// The three elements whose element children are observations. A ComponentStream may carry
/// any subset of them, including none.
@@ -105,7 +111,7 @@ internal static class MTConnectStreamsParser
{
// A chunk lifted out of a multipart frame can carry a byte-order mark or the framing's
// trailing line break; XmlReader rejects either ahead of the XML declaration.
document = XDocument.Parse(xml.Trim('', ' ', '\t', '\r', '\n'));
document = XDocument.Parse(xml.Trim('\uFEFF', ' ', '\t', '\r', '\n'));
}
catch (XmlException ex)
{
@@ -140,19 +146,19 @@ internal static class MTConnectStreamsParser
var root = document.Root
?? throw new InvalidDataException("MTConnect streams response has no root element.");
if (!IsNamed(root, "MTConnectStreams"))
if (!MTConnectXml.IsNamed(root, "MTConnectStreams"))
{
throw new InvalidDataException(DescribeUnexpectedRoot(root));
throw new InvalidDataException(MTConnectXml.DescribeUnexpectedRoot(root, "MTConnectStreams", Subject));
}
var header = ChildrenNamed(root, "Header").FirstOrDefault()
var header = MTConnectXml.ChildrenNamed(root, "Header").FirstOrDefault()
?? throw new InvalidDataException(
"MTConnect streams response has no <Header>; without it the sequence bookkeeping the sample pump runs on is unknowable.");
return new MTConnectStreamsResult(
RequiredLongAttribute(header, "instanceId"),
RequiredLongAttribute(header, "nextSequence"),
RequiredLongAttribute(header, "firstSequence"),
MTConnectXml.RequiredLongAttribute(header, "instanceId", HeaderOwner, Subject),
MTConnectXml.RequiredLongAttribute(header, "nextSequence", HeaderOwner, Subject),
MTConnectXml.RequiredLongAttribute(header, "firstSequence", HeaderOwner, Subject),
ReadObservations(root));
}
@@ -168,7 +174,7 @@ internal static class MTConnectStreamsParser
foreach (var container in root.Descendants().Where(IsObservationContainer))
{
var isCondition = IsNamed(container, ConditionContainer);
var isCondition = MTConnectXml.IsNamed(container, ConditionContainer);
foreach (var element in container.Elements())
{
observations.Add(ReadObservation(element, isCondition));
@@ -180,12 +186,20 @@ internal static class MTConnectStreamsParser
private static MTConnectObservation ReadObservation(XElement element, bool isCondition)
{
var dataItemId = RequiredAttribute(element, "dataItemId", $"Observation <{element.Name.LocalName}>");
var dataItemId = MTConnectXml.RequiredAttribute(element, "dataItemId", $"Observation <{element.Name.LocalName}>", Subject);
return new MTConnectObservation(
dataItemId,
isCondition ? ConditionState(element) : element.Value.Trim(),
ReadTimestampUtc(element, dataItemId));
ReadTimestampUtc(element, dataItemId),
// A DATA_SET/TABLE observation keeps its content in <Entry key="…"> children, so the
// text read above concatenates them into nonsense ("12" for two entries). Flagged here
// because this is the only layer that can still see the distinction — see
// MTConnectObservation.IsStructured. Never true for a CONDITION: its value comes from
// the element NAME, so child content cannot corrupt it, and flagging one would throw
// away a perfectly well-determined state.
IsStructured: !isCondition && element.HasElements);
}
/// <summary>
@@ -203,7 +217,7 @@ internal static class MTConnectStreamsParser
private static DateTime ReadTimestampUtc(XElement element, string dataItemId)
{
var raw = RequiredAttribute(element, "timestamp", $"Observation '{dataItemId}'");
var raw = MTConnectXml.RequiredAttribute(element, "timestamp", $"Observation '{dataItemId}'", Subject);
if (!DateTime.TryParse(
raw,
@@ -221,76 +235,6 @@ internal static class MTConnectStreamsParser
}
private static bool IsObservationContainer(XElement element) =>
ObservationContainers.Any(name => IsNamed(element, name));
ObservationContainers.Any(name => MTConnectXml.IsNamed(element, name));
private static bool IsNamed(XElement element, string localName) =>
string.Equals(element.Name.LocalName, localName, StringComparison.Ordinal);
private static IEnumerable<XElement> ChildrenNamed(XElement parent, string localName) =>
parent.Elements().Where(child => IsNamed(child, localName));
private static string RequiredAttribute(XElement element, string name, string owner)
{
var value = OptionalAttribute(element, name);
return value ?? throw new InvalidDataException(
$"MTConnect streams response is malformed: {owner} has no '{name}' attribute.");
}
/// <summary>
/// Reads an unqualified attribute, normalizing a present-but-empty value to <c>null</c>.
/// Only unqualified attributes are considered, so a namespace-prefixed vendor attribute can
/// never be mistaken for an observation's identity or timestamp.
/// </summary>
private static string? OptionalAttribute(XElement element, string name)
{
var value = element.Attribute(name)?.Value;
return string.IsNullOrWhiteSpace(value) ? null : value;
}
private static long RequiredLongAttribute(XElement header, string name)
{
var raw = RequiredAttribute(header, name, "The <Header>");
if (!long.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value))
{
throw new InvalidDataException(
$"MTConnect streams response is malformed: the <Header> has a non-numeric '{name}' attribute ('{raw}').");
}
return value;
}
/// <summary>
/// Builds the message for a root element that is not <c>MTConnectStreams</c>, lifting the
/// Agent's own error text when the response is an <c>MTConnectError</c> document (which
/// Agents return under HTTP 200 for, e.g., a <c>from</c> outside the buffer).
/// </summary>
private static string DescribeUnexpectedRoot(XElement root)
{
var message =
$"MTConnect streams response root element is <{root.Name.LocalName}>, expected <MTConnectStreams>.";
if (!IsNamed(root, "MTConnectError"))
{
return message;
}
var errors = root.Descendants()
.Where(e => IsNamed(e, "Error"))
.Select(e =>
{
var code = OptionalAttribute(e, "errorCode");
var text = e.Value.Trim();
return code is null ? text : $"{code}: {text}";
})
.Where(text => text.Length > 0)
.ToList();
return errors.Count == 0
? message
: $"{message} The Agent reported: {string.Join("; ", errors)}";
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectXml.cs
+146
View File
@@ -0,0 +1,146 @@
using System.Globalization;
using System.Xml.Linq;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// <summary>
/// The element/attribute reading rules shared by <see cref="MTConnectProbeParser"/> and
/// <see cref="MTConnectStreamsParser"/>. Both documents are the same dialect of XML and must be
/// read by the same rules; keeping one copy is what stops the two parsers drifting apart on the
/// two rules below, either of which fails silently rather than loudly.
/// </summary>
/// <remarks>
/// <para>
/// <b>Rule 1: elements are matched on <see cref="XName.LocalName"/>, ignoring the
/// namespace.</b> The namespace URI carries the MTConnect version
/// (<c>urn:mtconnect.org:MTConnectDevices:1.3</c> … <c>:2.0</c>), so a fully-qualified match
/// would hard-code a version. It also silently drops vendor-extension elements declared in
/// their own namespace, whose standard children still inherit the default MTConnect
/// namespace — the browse tree simply comes back short, with no error anywhere.
/// </para>
/// <para>
/// <b>Rule 2: attributes are read <i>unqualified</i>.</b> A prefixed attribute
/// (<c>xsi:type</c> on a document root, a vendor's <c>x:dataItemId</c>) must never be
/// mistaken for the real <c>type</c> / <c>dataItemId</c>, which would invent a data item the
/// probe never declared.
/// </para>
/// </remarks>
internal static class MTConnectXml
{
/// <summary>Does <paramref name="element"/> have the given local name, whatever its namespace?</summary>
/// <param name="element">The element to test.</param>
/// <param name="localName">The unqualified name to match.</param>
public static bool IsNamed(XElement element, string localName) =>
string.Equals(element.Name.LocalName, localName, StringComparison.Ordinal);
/// <summary>Direct element children of <paramref name="parent"/> with the given local name.</summary>
/// <param name="parent">The element whose children to filter.</param>
/// <param name="localName">The unqualified name to match.</param>
public static IEnumerable<XElement> ChildrenNamed(XElement parent, string localName) =>
parent.Elements().Where(child => IsNamed(child, localName));
/// <summary>
/// Reads an unqualified attribute, normalizing a present-but-empty value to <c>null</c>.
/// See the type-level remarks for why only unqualified attributes are considered.
/// </summary>
/// <param name="element">The element carrying the attribute.</param>
/// <param name="name">The unqualified attribute name.</param>
public static string? OptionalAttribute(XElement element, string name)
{
var value = element.Attribute(name)?.Value;
return string.IsNullOrWhiteSpace(value) ? null : value;
}
/// <summary>Reads a required unqualified attribute, or throws naming the owner and the attribute.</summary>
/// <param name="element">The element carrying the attribute.</param>
/// <param name="name">The unqualified attribute name.</param>
/// <param name="owner">How to describe the element in the error message.</param>
/// <param name="subject">How to describe the document in the error message (e.g. "/probe response").</param>
/// <exception cref="InvalidDataException">The attribute is absent or blank.</exception>
public static string RequiredAttribute(XElement element, string name, string owner, string subject)
{
var value = OptionalAttribute(element, name);
return value ?? throw new InvalidDataException(
$"MTConnect {subject} is malformed: {owner} has no '{name}' attribute.");
}
/// <summary>Reads an optional unqualified attribute as an <see cref="int"/>.</summary>
/// <param name="element">The element carrying the attribute.</param>
/// <param name="name">The unqualified attribute name.</param>
/// <param name="owner">How to describe the element in the error message.</param>
/// <param name="subject">How to describe the document in the error message.</param>
/// <exception cref="InvalidDataException">The attribute is present but not an integer.</exception>
public static int? OptionalIntAttribute(XElement element, string name, string owner, string subject)
{
var raw = OptionalAttribute(element, name);
if (raw is null)
{
return null;
}
if (!int.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value))
{
throw new InvalidDataException(
$"MTConnect {subject} is malformed: {owner} has a non-numeric '{name}' attribute ('{raw}').");
}
return value;
}
/// <summary>Reads a required unqualified attribute as a <see cref="long"/>.</summary>
/// <param name="element">The element carrying the attribute.</param>
/// <param name="name">The unqualified attribute name.</param>
/// <param name="owner">How to describe the element in the error message.</param>
/// <param name="subject">How to describe the document in the error message.</param>
/// <exception cref="InvalidDataException">The attribute is absent, blank, or not an integer.</exception>
public static long RequiredLongAttribute(XElement element, string name, string owner, string subject)
{
var raw = RequiredAttribute(element, name, owner, subject);
if (!long.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value))
{
throw new InvalidDataException(
$"MTConnect {subject} is malformed: {owner} has a non-numeric '{name}' attribute ('{raw}').");
}
return value;
}
/// <summary>
/// Builds the error message for a document whose root is not what was expected, lifting the
/// Agent's own error text when the response is an <c>MTConnectError</c> document — which
/// Agents return under <b>HTTP 200</b>, so <c>EnsureSuccessStatusCode</c> never fires and
/// this is the only place the operator can learn what the Agent objected to.
/// </summary>
/// <param name="root">The document's actual root element.</param>
/// <param name="expectedRootName">The root element name the caller required.</param>
/// <param name="subject">How to describe the document in the error message.</param>
public static string DescribeUnexpectedRoot(XElement root, string expectedRootName, string subject)
{
var message =
$"MTConnect {subject} root element is <{root.Name.LocalName}>, expected <{expectedRootName}>.";
if (!IsNamed(root, "MTConnectError"))
{
return message;
}
var errors = root.Descendants()
.Where(e => IsNamed(e, "Error"))
.Select(e =>
{
var code = OptionalAttribute(e, "errorCode");
var text = e.Value.Trim();
return code is null ? text : $"{code}: {text}";
})
.Where(text => text.Length > 0)
.ToList();
return errors.Count == 0
? message
: $"{message} The Agent reported: {string.Join("; ", errors)}";
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs
+517 -10
View File
@@ -1,7 +1,9 @@
using System.Diagnostics;
using System.Globalization;
using System.Net;
using System.Net.Http.Headers;
using System.Text;
using Microsoft.Extensions.Logging;
using Shouldly;
using Xunit;
@@ -10,7 +12,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
/// <summary>
/// Task 7 — the <c>/current</c> + <c>/sample</c> legs: <see cref="MTConnectStreamsParser"/>
/// turning an <c>MTConnectStreams</c> document into <see cref="MTConnectStreamsResult"/>, the
/// ring-buffer sequence-gap signal (<see cref="MTConnectAgentClient.IsSequenceGap"/>), and the
/// ring-buffer sequence-gap signal (<see cref="IMTConnectAgentClient.IsSequenceGap"/>), and the
/// <c>multipart/x-mixed-replace</c> chunk framing + heartbeat watchdog on the long-poll leg.
/// Every test runs against canned fixtures or a stubbed <see cref="HttpMessageHandler"/> — no
/// socket is opened anywhere in this file.
@@ -179,7 +181,7 @@ public sealed class MTConnectStreamsParseTests
{
var chunk = MTConnectStreamsParser.Parse(File.ReadAllText(GapFixture));
MTConnectAgentClient.IsSequenceGap(requestedFrom: 1, chunk).ShouldBeTrue();
IMTConnectAgentClient.IsSequenceGap(expectedFrom: 1, chunk).ShouldBeTrue();
}
[Fact]
@@ -188,18 +190,18 @@ public sealed class MTConnectStreamsParseTests
// The falsifiability leg: a helper hard-wired to true would pass the gap test alone.
var contiguous = MTConnectStreamsParser.Parse(File.ReadAllText(SampleFixture));
MTConnectAgentClient.IsSequenceGap(requestedFrom: 108, contiguous).ShouldBeFalse();
IMTConnectAgentClient.IsSequenceGap(expectedFrom: 108, contiguous).ShouldBeFalse();
}
[Theory]
[InlineData(4999L, true)] // requested older than the buffer holds — data was lost
[InlineData(4999L, true)] // expected older than the buffer holds — data was lost
[InlineData(5000L, false)] // exactly the oldest retained sequence — nothing was lost
[InlineData(5001L, false)] // the agent simply had nothing older to send
public void Sequence_gap_is_strictly_firstSequence_greater_than_requested_from(long requestedFrom, bool expected)
public void Sequence_gap_is_strictly_firstSequence_greater_than_expected_from(long expectedFrom, bool expected)
{
var chunk = MTConnectStreamsParser.Parse(File.ReadAllText(GapFixture));
MTConnectAgentClient.IsSequenceGap(requestedFrom, chunk).ShouldBe(expected);
IMTConnectAgentClient.IsSequenceGap(expectedFrom, chunk).ShouldBe(expected);
}
// --------------------------------------------------------------- legal but degenerate documents
@@ -499,11 +501,17 @@ public sealed class MTConnectStreamsParseTests
var from = 108L;
var gaps = new List<bool>();
await Should.ThrowAsync<MTConnectStreamEndedException>(async () =>
{
await foreach (var chunk in client.SampleAsync(from, Ct))
{
gaps.Add(MTConnectAgentClient.IsSequenceGap(from, chunk));
// `from` is the RUNNING cursor, not the original argument — comparing every chunk
// against the opening `from` would report a gap on every chunk once the Agent's
// buffer rolls past it (I1).
gaps.Add(IMTConnectAgentClient.IsSequenceGap(from, chunk));
from = chunk.NextSequence;
}
});
gaps.ShouldBe([false, true]);
from.ShouldBe(5005);
@@ -541,7 +549,7 @@ public sealed class MTConnectStreamsParseTests
var chunks = await DrainAsync(client.SampleAsync(113, Ct));
chunks.ShouldHaveSingleItem().Observations.ShouldBeEmpty();
MTConnectAgentClient.IsSequenceGap(113, chunks[0]).ShouldBeFalse();
IMTConnectAgentClient.IsSequenceGap(113, chunks[0]).ShouldBeFalse();
}
[Fact(Timeout = 30_000)]
@@ -623,14 +631,107 @@ public sealed class MTConnectStreamsParseTests
}
[Fact]
public async Task SampleAsync_ends_cleanly_when_the_agent_closes_the_stream()
public async Task SampleAsync_reports_the_closing_boundary_rather_than_ending_quietly()
{
// C1. The seam contracts SampleAsync to run until cancelled, so a pump has no reason to
// handle normal completion: returning here would either drop the subscription silently or
// hot-loop reconnecting on an enumerator that ends immediately (#485).
var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct));
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var chunks = await DrainAsync(client.SampleAsync(108, Ct));
var chunks = new List<MTConnectStreamsResult>();
var ex = await Should.ThrowAsync<MTConnectStreamEndedException>(async () =>
{
await foreach (var chunk in client.SampleAsync(108, Ct))
{
chunks.Add(chunk);
}
});
chunks.ShouldHaveSingleItem();
ex.Reason.ShouldBe(MTConnectStreamEndReason.ClosingBoundary);
ex.ChunksDelivered.ShouldBe(1);
ex.AgentUri.ShouldContain("/sample?from=108");
}
[Fact]
public async Task SampleAsync_reports_a_connection_that_drops_mid_stream()
{
// No closing boundary — the body simply stops. Distinguished from a deliberate close so an
// operator can tell a flaky network from an Agent honouring a bounded request.
var truncated = StubHandler.MultipartBody(
"bnd", withContentLength: true, closing: false, await File.ReadAllTextAsync(SampleFixture, Ct));
var handler = StubHandler.RespondingWithBody("bnd", truncated);
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var ex = await Should.ThrowAsync<MTConnectStreamEndedException>(
async () => await EnumerateAsync(client.SampleAsync(108, Ct)));
ex.Reason.ShouldBe(MTConnectStreamEndReason.ConnectionClosed);
ex.ChunksDelivered.ShouldBe(1);
}
[Fact]
public async Task SampleAsync_reports_a_non_multipart_answer_as_a_configuration_error()
{
// The worst C1 shape: the Agent ignored the streaming request entirely. Yielding the one
// snapshot it did return and finishing would report a dead subscription as a healthy one.
var handler = StubHandler.RespondingWith(await File.ReadAllTextAsync(CurrentFixture, Ct));
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var ex = await Should.ThrowAsync<MTConnectStreamNotSupportedException>(
async () => await EnumerateAsync(client.SampleAsync(108, Ct)));
ex.Message.ShouldContain("multipart/x-mixed-replace");
ex.Message.ShouldContain("text/xml");
}
[Fact]
public async Task SampleAsync_yields_nothing_at_all_from_a_non_multipart_answer()
{
// Explicitly pinned: the parsed document must NOT reach the caller. One snapshot plus a
// finished stream is precisely how a dead data path disguises itself as a working one.
var handler = StubHandler.RespondingWith(await File.ReadAllTextAsync(CurrentFixture, Ct));
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var chunks = new List<MTConnectStreamsResult>();
await Should.ThrowAsync<MTConnectStreamNotSupportedException>(async () =>
{
await foreach (var chunk in client.SampleAsync(108, Ct))
{
chunks.Add(chunk);
}
});
chunks.ShouldBeEmpty();
}
[Fact]
public async Task SampleAsync_fails_fast_when_a_multipart_response_carries_no_boundary_parameter()
{
// M2. Falling through to whole-body reading would surface only as a watchdog TimeoutException
// much later, pointing an operator at the network instead of at the malformed header.
var handler = StubHandler.RespondingWithBoundarylessMultipart();
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var ex = await Should.ThrowAsync<MTConnectStreamNotSupportedException>(
async () => await EnumerateAsync(client.SampleAsync(1, Ct)));
ex.Message.ShouldContain("boundary");
}
[Fact]
public async Task A_stream_end_and_a_stream_misconfiguration_share_one_catchable_base_type()
{
// A pump that only wants "the stream is over" should not have to name both.
var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct));
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var ended = await Should.ThrowAsync<MTConnectStreamEndedException>(
async () => await EnumerateAsync(client.SampleAsync(108, Ct)));
ended.ShouldBeAssignableTo<MTConnectStreamException>();
typeof(MTConnectStreamException).IsAssignableFrom(typeof(MTConnectStreamNotSupportedException)).ShouldBeTrue();
}
[Fact]
@@ -644,6 +745,255 @@ public sealed class MTConnectStreamsParseTests
handler.RequestCount.ShouldBe(0);
}
// ------------------------------------------- I2: framing when the transport splits every read
/// <summary>
/// Serving the whole body from one buffer completes every frame in a single
/// <c>ReadAsync</c>, so the split-boundary path, the split-header path and the multi-fill
/// loop are never entered — on the one behaviour whose headline risk is framing. These drive
/// the same fixtures through a transport that hands back only N bytes at a time; at N = 1
/// every delimiter, every header line and every document is split maximally.
/// </summary>
[Theory]
[InlineData(1)]
[InlineData(3)]
[InlineData(7)]
public async Task SampleAsync_frames_correctly_when_the_transport_splits_every_boundary_and_header(int bytesPerRead)
{
var handler = StubHandler.RespondingWithChunkedMultipart(
"--------bnd",
bytesPerRead,
withContentLength: true,
await File.ReadAllTextAsync(SampleFixture, Ct),
await File.ReadAllTextAsync(GapFixture, Ct));
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var chunks = await DrainAsync(client.SampleAsync(108, Ct));
chunks.Select(c => c.FirstSequence).ShouldBe([108L, 5000L]);
chunks.Select(c => c.NextSequence).ShouldBe([113L, 5005L]);
chunks[0].Observations.Count.ShouldBe(5);
}
[Theory]
[InlineData(1)]
[InlineData(3)]
[InlineData(7)]
public async Task SampleAsync_boundary_scan_framing_survives_a_split_transport_too(int bytesPerRead)
{
var handler = StubHandler.RespondingWithChunkedMultipart(
"bnd",
bytesPerRead,
withContentLength: false,
await File.ReadAllTextAsync(SampleFixture, Ct),
await File.ReadAllTextAsync(GapFixture, Ct));
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var chunks = await DrainAsync(client.SampleAsync(108, Ct));
chunks.Select(c => c.FirstSequence).ShouldBe([108L, 5000L]);
}
[Fact]
public async Task SampleAsync_grows_its_buffer_for_a_part_larger_than_the_initial_read_buffer()
{
// Forces Array.Resize + a Consume across the grown buffer, which a 2 KB fixture never does.
var large = LargeStreamsDocument(observationCount: 900);
Encoding.UTF8.GetByteCount(large).ShouldBeGreaterThan(16 * 1024);
var handler = StubHandler.RespondingWithChunkedMultipart(
"bnd", bytesPerRead: 1024, withContentLength: true, large, large);
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
var chunks = await DrainAsync(client.SampleAsync(1, Ct));
chunks.Count.ShouldBe(2);
chunks[0].Observations.Count.ShouldBe(900);
chunks[1].Observations.Count.ShouldBe(900);
}
// ------------------------------------------------------- I3: disposal during a live enumeration
[Fact(Timeout = 30_000)]
public async Task Disposing_the_client_mid_enumeration_surfaces_as_cancellation()
{
// Task 9 re-initializes by disposing and rebuilding this client, which can happen while a
// pump is still enumerating. An unclassified ObjectDisposedException/IOException from the
// socket is something no caller is written to expect.
var prefix = StubHandler.MultipartBody(
"bnd", withContentLength: true, closing: false, await File.ReadAllTextAsync(SampleFixture, Ct));
var handler = StubHandler.RespondingWithStallingStream("bnd", prefix);
var client = NewClient(handler, new MTConnectDriverOptions
{
AgentUri = "http://agent:5000",
RequestTimeoutMs = 5000,
HeartbeatMs = 20_000
});
await using var chunks = client.SampleAsync(108, Ct).GetAsyncEnumerator(Ct);
(await chunks.MoveNextAsync()).ShouldBeTrue();
var pending = chunks.MoveNextAsync().AsTask();
client.Dispose();
var ex = await Should.ThrowAsync<OperationCanceledException>(async () => await pending);
ex.ShouldNotBeOfType<ObjectDisposedException>();
}
[Fact]
public async Task Using_a_disposed_client_throws_object_disposed_rather_than_dialling()
{
var handler = StubHandler.Hanging();
var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
client.Dispose();
await Should.ThrowAsync<ObjectDisposedException>(async () => await client.CurrentAsync(Ct));
await Should.ThrowAsync<ObjectDisposedException>(
async () => await EnumerateAsync(client.SampleAsync(1, Ct)));
handler.RequestCount.ShouldBe(0);
}
[Fact]
public void Dispose_is_idempotent()
{
var client = NewClient(StubHandler.Hanging(), new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
client.Dispose();
Should.NotThrow(client.Dispose);
}
// ------------------------------------------- I4: the degraded framing mode is operator-visible
[Fact]
public async Task A_part_without_content_length_warns_once_that_framing_is_degraded()
{
var logger = new RecordingLogger();
var handler = StubHandler.RespondingWithMultipart(
"bnd",
withContentLength: false,
await File.ReadAllTextAsync(SampleFixture, Ct),
await File.ReadAllTextAsync(GapFixture, Ct));
using var client = new MTConnectAgentClient(
new MTConnectDriverOptions { AgentUri = "http://agent:5000" }, handler, logger);
await DrainAsync(client.SampleAsync(108, Ct));
// Once, not once per part — a per-chunk warning on a long poll is its own outage.
logger.Warnings.Count.ShouldBe(1);
logger.Warnings[0].ShouldContain("Content-length");
logger.Warnings[0].ShouldContain("http://agent:5000/sample");
}
[Fact]
public async Task The_normal_content_length_path_warns_about_nothing()
{
var logger = new RecordingLogger();
var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct));
using var client = new MTConnectAgentClient(
new MTConnectDriverOptions { AgentUri = "http://agent:5000" }, handler, logger);
await DrainAsync(client.SampleAsync(108, Ct));
logger.Warnings.ShouldBeEmpty();
}
// ---------------------------------------- I5: operator-authorable options that reach the wire
[Theory]
[InlineData(0)]
[InlineData(-1)]
public void Ctor_rejects_a_non_positive_heartbeat(int heartbeatMs)
{
// heartbeat=0 goes on the query string, and an Agent answers it with an MTConnectError
// under HTTP 200 — surfacing as a parse failure that never names the offending key.
var ex = Should.Throw<ArgumentOutOfRangeException>(() => new MTConnectAgentClient(
new MTConnectDriverOptions { AgentUri = "http://agent:5000", HeartbeatMs = heartbeatMs }));
ex.Message.ShouldContain(nameof(MTConnectDriverOptions.HeartbeatMs));
}
[Theory]
[InlineData(0)]
[InlineData(-1)]
public void Ctor_rejects_a_non_positive_sample_interval(int sampleIntervalMs)
{
var ex = Should.Throw<ArgumentOutOfRangeException>(() => new MTConnectAgentClient(
new MTConnectDriverOptions { AgentUri = "http://agent:5000", SampleIntervalMs = sampleIntervalMs }));
ex.Message.ShouldContain(nameof(MTConnectDriverOptions.SampleIntervalMs));
}
[Theory]
[InlineData(0)]
[InlineData(-1)]
public void Ctor_rejects_a_non_positive_sample_count(int sampleCount)
{
var ex = Should.Throw<ArgumentOutOfRangeException>(() => new MTConnectAgentClient(
new MTConnectDriverOptions { AgentUri = "http://agent:5000", SampleCount = sampleCount }));
ex.Message.ShouldContain(nameof(MTConnectDriverOptions.SampleCount));
}
// ------------------------------------------- structured (DATA_SET / TABLE) observation signal
[Fact]
public void An_observation_whose_content_is_child_entries_is_flagged_structured()
{
// <Entry> children concatenate to "12" through element.Value — keys gone, values run
// together. The index codes this BadNotSupported rather than publishing noise as Good.
var result = MTConnectStreamsParser.Parse(StreamsDocument(
"""
<Events><PathFeedrateOverride dataItemId="ds" timestamp="2026-07-24T12:00:00Z" count="2">
<Entry key="V1">1</Entry><Entry key="V2">2</Entry>
</PathFeedrateOverride></Events>
"""));
var observation = result.Observations.ShouldHaveSingleItem();
observation.IsStructured.ShouldBeTrue();
observation.DataItemId.ShouldBe("ds");
}
[Fact]
public void A_plain_text_observation_containing_spaces_is_NOT_flagged_structured()
{
// The load-bearing case: this proves the signal is "the element had element children", not
// a whitespace heuristic. A Message reading "Coolant level low" is perfectly valid data and
// is textually indistinguishable from concatenated entries.
var result = MTConnectStreamsParser.Parse(StreamsDocument(
"""<Events><Message dataItemId="msg" timestamp="2026-07-24T12:00:00Z">Coolant level low</Message></Events>"""));
var observation = result.Observations.ShouldHaveSingleItem();
observation.IsStructured.ShouldBeFalse();
observation.Value.ShouldBe("Coolant level low");
}
[Fact]
public void Every_observation_in_the_committed_fixtures_is_unstructured()
{
// Includes the space-separated TIME_SERIES vector, which is text and must stay unflagged.
var current = MTConnectStreamsParser.Parse(File.ReadAllText(CurrentFixture));
current.Observations.ShouldAllBe(o => !o.IsStructured);
}
[Fact]
public void A_condition_is_never_flagged_structured_even_with_child_content()
{
// A condition's value comes from the element NAME, so child content cannot corrupt it.
// Flagging one would throw away a perfectly well-determined state.
var result = MTConnectStreamsParser.Parse(StreamsDocument(
"""
<Condition><Fault dataItemId="c" timestamp="2026-07-24T12:00:00Z" type="SYSTEM">
<Detail>vendor extension</Detail>
</Fault></Condition>
"""));
var observation = result.Observations.ShouldHaveSingleItem();
observation.IsStructured.ShouldBeFalse();
observation.Value.ShouldBe("Fault");
}
// ------------------------------------------------------------------------------------ helpers
private static CancellationToken Ct => TestContext.Current.CancellationToken;
@@ -654,14 +1004,51 @@ public sealed class MTConnectStreamsParseTests
private static async Task<List<MTConnectStreamsResult>> DrainAsync(IAsyncEnumerable<MTConnectStreamsResult> source)
{
var results = new List<MTConnectStreamsResult>();
try
{
await foreach (var item in source)
{
results.Add(item);
}
}
catch (MTConnectStreamEndedException)
{
// Expected under the C1 contract: every non-cancellation end of a /sample stream is
// loud. Tests that care about HOW it ended assert on the exception directly.
}
return results;
}
/// <summary>Enumerates to completion, discarding chunks and letting every exception escape.</summary>
private static async Task EnumerateAsync(IAsyncEnumerable<MTConnectStreamsResult> source)
{
await foreach (var _ in source)
{
// Drained for effect only.
}
}
/// <summary>A streams document big enough to force the frame reader to grow its buffer.</summary>
private static string LargeStreamsDocument(int observationCount)
{
var samples = new StringBuilder();
for (var i = 0; i < observationCount; i++)
{
samples.Append(CultureInfo.InvariantCulture,
$"""<Position dataItemId="d{i}" timestamp="2026-07-24T12:00:00Z" sequence="{i + 1}">{i}.5</Position>""");
}
return $"""
<MTConnectStreams xmlns="urn:mtconnect.org:MTConnectStreams:1.3">
<Header instanceId="1655000000" firstSequence="1" lastSequence="{observationCount}" nextSequence="{observationCount + 1}"/>
<Streams><DeviceStream name="D" uuid="d"><ComponentStream component="Path" componentId="p">
<Samples>{samples}</Samples>
</ComponentStream></DeviceStream></Streams>
</MTConnectStreams>
""";
}
private static Dictionary<string, MTConnectObservation> ParseCurrent() => ParseFixture(CurrentFixture);
private static Dictionary<string, MTConnectObservation> ParseFixture(string path) =>
@@ -717,6 +1104,35 @@ public sealed class MTConnectStreamsParseTests
public static StubHandler RespondingWithStallingStream(string boundary, byte[] prefix) =>
new(() => Multipart(boundary, new StallingContent(prefix)), hang: false);
/// <summary>Serves an already-built multipart body verbatim (e.g. one with no closing delimiter).</summary>
public static StubHandler RespondingWithBody(string boundary, byte[] body) =>
new(() => Multipart(boundary, new ByteArrayContent(body)), hang: false);
/// <summary>
/// Serves a multipart body through a transport that hands back at most
/// <paramref name="bytesPerRead"/> bytes per <c>ReadAsync</c>, so boundaries, part
/// headers and documents are all split across reads.
/// </summary>
public static StubHandler RespondingWithChunkedMultipart(
string boundary, int bytesPerRead, bool withContentLength, params string[] parts)
{
var body = MultipartBody(boundary, withContentLength, closing: true, parts);
return new StubHandler(() => Multipart(boundary, new ChunkedContent(body, bytesPerRead)), hang: false);
}
/// <summary>A response that claims multipart but omits the <c>boundary</c> parameter.</summary>
public static StubHandler RespondingWithBoundarylessMultipart() =>
new(
() =>
{
var content = new ByteArrayContent([]);
content.Headers.ContentType = new MediaTypeHeaderValue("multipart/x-mixed-replace");
return new HttpResponseMessage(HttpStatusCode.OK) { Content = content };
},
hang: false);
/// <summary>Builds a <c>multipart/x-mixed-replace</c> body exactly as an Agent writes one.</summary>
public static byte[] MultipartBody(string boundary, bool withContentLength, bool closing, params string[] parts)
{
@@ -769,6 +1185,97 @@ public sealed class MTConnectStreamsParseTests
}
}
/// <summary>Hands back at most <c>bytesPerRead</c> bytes per read — the split-transport double.</summary>
private sealed class ChunkedContent(byte[] body, int bytesPerRead) : HttpContent
{
protected override Task<Stream> CreateContentReadStreamAsync() =>
Task.FromResult<Stream>(new ChunkedStream(body, bytesPerRead));
protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) =>
stream.WriteAsync(body, 0, body.Length);
protected override bool TryComputeLength(out long length)
{
length = body.Length;
return true;
}
}
private sealed class ChunkedStream(byte[] body, int bytesPerRead) : Stream
{
private int _position;
public override bool CanRead => true;
public override bool CanSeek => false;
public override bool CanWrite => false;
public override long Length => body.Length;
public override long Position
{
get => _position;
set => throw new NotSupportedException();
}
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
if (_position >= body.Length)
{
return ValueTask.FromResult(0);
}
var count = Math.Min(Math.Min(bytesPerRead, buffer.Length), body.Length - _position);
body.AsMemory(_position, count).CopyTo(buffer);
_position += count;
return ValueTask.FromResult(count);
}
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
public override void Flush()
{
}
public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
}
/// <summary>Captures warning-level log lines so the degraded-framing notice can be asserted.</summary>
private sealed class RecordingLogger : ILogger
{
public List<string> Warnings { get; } = [];
public IDisposable? BeginScope<TState>(TState state)
where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(
LogLevel logLevel,
EventId eventId,
TState state,
Exception? exception,
Func<TState, Exception?, string> formatter)
{
if (logLevel == LogLevel.Warning)
{
Warnings.Add(formatter(state, exception));
}
}
}
/// <summary>Serves <paramref name="prefix"/>, then never completes another read.</summary>
private sealed class StallingContent(byte[] prefix) : HttpContent
{