diff --git a/docs/plans/2026-07-24-mtconnect-driver.md b/docs/plans/2026-07-24-mtconnect-driver.md index a3b3abef..4e6b733b 100644 --- a/docs/plans/2026-07-24-mtconnect-driver.md +++ b/docs/plans/2026-07-24-mtconnect-driver.md @@ -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 `` 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` diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs index b596827a..86b9c98d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs @@ -16,7 +16,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// /sample 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; 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. /// public interface IMTConnectAgentClient { @@ -38,13 +40,73 @@ public interface IMTConnectAgentClient /// /// Opens an Agent /sample long-poll stream starting at sequence - /// and yields one per multipart chunk the Agent sends, - /// indefinitely, until is cancelled. Each yielded result's - /// is load-bearing — the caller compares - /// it against the requested to detect a ring-buffer sequence gap and - /// re-baseline via . + /// and yields one per multipart chunk the Agent sends. /// + /// + /// + /// The only way this enumeration ends without throwing is being + /// cancelled. 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 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 + /// (transient — reconnect) and + /// (configuration — reconnecting will + /// reproduce it forever). + /// + /// + /// Advancing the cursor. After each chunk, the caller's next expected sequence is + /// that chunk's — not the + /// this call was opened with. Feed that running value to + /// and, on reconnect, to a fresh SampleAsync. + /// + /// /// The sequence number to resume streaming from. - /// Cancellation token; cancelling ends the stream. + /// Cancellation token; cancelling is the contracted way to end the stream. IAsyncEnumerable SampleAsync(long from, CancellationToken ct); + + /// + /// Has the Agent's ring buffer overflowed past the sequence the caller expected next? True + /// when the chunk's oldest retained sequence is strictly newer than + /// , meaning observations between the two were evicted before + /// the caller could read them and an incremental update would silently skip values. + /// + /// + /// + /// is a running cursor, not the original + /// from. It equals the from passed to only + /// for the first chunk; from the second chunk on it is the previous + /// chunk's . Comparing every chunk + /// against the original from instead reports a gap on every chunk once the + /// Agent's buffer has rolled past it — an endless /current re-baseline storm + /// against a perfectly healthy stream. + /// + /// + /// Equality is not a gap: firstSequence == expectedFrom is the ordinary + /// contiguous case where the Agent simply had nothing older to send. + /// + /// + /// This check alone does NOT cover every ring-buffer overflow. It can only judge + /// a chunk the Agent was willing to send — and cppagent answers a from that has + /// already fallen out of its buffer with an MTConnectError / OUT_OF_RANGE + /// document served under HTTP 200, which this client surfaces as an + /// out of , never as a chunk. + /// A pump that re-baselines only on IsSequenceGap will therefore fault instead of + /// recovering exactly when it has fallen furthest behind. Treat an OUT_OF_RANGE + /// parse failure as a re-baseline trigger (re-prime from , + /// resume from that snapshot's NextSequence) just like a detected gap. + /// + /// + /// The sequence the caller expected this chunk to start at. + /// The chunk the Agent answered with. + /// + /// true when observations were lost and the caller must re-baseline via + /// rather than trusting an incremental update. + /// + static bool IsSequenceGap(long expectedFrom, MTConnectStreamsResult chunk) + { + ArgumentNullException.ThrowIfNull(chunk); + + return chunk.FirstSequence > expectedFrom; + } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs index a30a4aa1..8ce654b1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs @@ -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 supposed 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 -/// and is bounded differently instead (below). +/// 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. /// /// /// Every call is deadline-bounded, but not all by the same mechanism. The unary legs @@ -41,61 +44,94 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// and the watchdog is what closes it here. /// /// -/// A non-positive RequestTimeoutMs is rejected at construction so an -/// operator-authored 0 can never come to mean "wait forever" (arch-review 01/S-6). +/// Every operator-authorable duration/count is rejected at construction if non-positive. +/// A 0 must never come to mean "wait forever" (arch-review 01/S-6), and the three +/// /sample query knobs additionally reach the wire: an Agent answers +/// heartbeat=0 with an MTConnectError under HTTP 200, which would otherwise +/// surface as a parse failure that never names the offending config key. +/// +/// +/// Disposal is safe against an in-flight enumeration. Task 9 re-initializes the +/// driver by disposing and rebuilding this client, which can happen while a pump is still +/// enumerating . Rather than letting the in-flight read fault with +/// an unclassified or 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 +/// . /// /// /// This class does not swallow failures into an empty model: an unreachable Agent, a non-2xx -/// status, or an unparseable body all throw. InitializeAsync (Task 9) is what catches -/// that and moves the driver to Faulted. +/// status, an unparseable body, and the end of a /sample stream all throw. See +/// for why the last of those is an exception rather +/// than a normal end of enumeration. InitializeAsync (Task 9) is what catches these +/// and moves the driver to Faulted. /// /// 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; /// Creates a client for the Agent described by . /// The driver options carrying the Agent URI, device scope, and per-call deadline. + /// + /// Optional logger. Used only for conditions an operator needs to see but which are not + /// failures — today, the degraded /sample framing mode. + /// /// is missing or is not an absolute HTTP(S) URI. - /// is not positive. - public MTConnectAgentClient(MTConnectDriverOptions options) - : this(options, handler: null) + /// + /// Any of , + /// , + /// or + /// is not positive. + /// + public MTConnectAgentClient(MTConnectDriverOptions options, ILogger? logger = null) + : this(options, handler: null, logger) { } /// - /// Test seam: as , but sends through - /// the supplied handler so the request/response round trip can be exercised with no socket. - /// The caller retains ownership of . + /// Test seam: as , but + /// sends through the supplied handler so the request/response round trip can be exercised + /// with no socket. The caller retains ownership of . /// - 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), - options.RequestTimeoutMs, - $"{nameof(MTConnectDriverOptions.RequestTimeoutMs)} must be positive; a non-positive per-call deadline would let a frozen Agent wedge the driver indefinitely."); - } + RequirePositive( + options.RequestTimeoutMs, + 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; } - /// - /// Has the Agent's ring buffer overflowed past what the caller asked for? True when the - /// chunk's oldest retained sequence is strictly 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: firstSequence == requestedFrom is the ordinary - /// contiguous case where the Agent simply had nothing older to send. - /// - /// The from sequence the caller passed to . - /// The chunk the Agent answered with. - /// - /// true when observations were lost and the caller must re-baseline via - /// rather than trusting an incremental update. - /// - public static bool IsSequenceGap(long requestedFrom, MTConnectStreamsResult chunk) - { - ArgumentNullException.ThrowIfNull(chunk); - - return chunk.FirstSequence > requestedFrom; - } - /// public async Task 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 /// public async Task 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 /// 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 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. /// private async Task 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 } /// Reads a non-multipart streaming body whole, under the unary deadline. - private async Task ReadWholeBodyAsync(Stream body, Uri uri, CancellationToken ct) + private async Task 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 boundary parameter of a multipart/* response, or null when the /// Agent answered with a single (non-multipart) document. /// - private static string? ResolveMultipartBoundary(HttpResponseMessage response) + /// + /// The response claims to be multipart/* but carries no boundary 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 that points an operator at the network instead of + /// at the malformed header. + /// + 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 /// /// /// Why hand-rolled. ASP.NET Core's MultipartReader is not in this - /// dependency set, and TrakHound's - /// MTConnectHttpClientStream — 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 MTConnectHttpClientStream — 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. /// /// /// Two framing paths, and the first one matters for latency. When the part carries /// a Content-length 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 next 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.) /// /// /// Every read is watchdog-bounded. 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. /// /// - 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; + + /// How the stream ended. Only meaningful once a read has returned null. + public MTConnectStreamEndReason EndReason { get; private set; } = MTConnectStreamEndReason.ConnectionClosed; /// /// Reads the next part's payload, or null once the closing delimiter is seen or - /// the Agent closes the connection. + /// the Agent closes the connection — see for which. /// public async Task 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 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) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs index 76bd41c9..740a2d71 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs @@ -136,10 +136,33 @@ public sealed record MTConnectStreamsResult( /// The observation's Agent-reported timestamp, normalized to UTC ( /// is always ). Becomes the OPC UA variable's SourceTimestamp. /// +/// +/// true when the Agent carried this observation's real content in child elements +/// rather than as text — an MTConnect 2.0 DATA_SET / TABLE observation, whose +/// content is a list of <Entry key="…"> elements. +/// +/// Why the flag exists, and why only the parser can set it. is +/// the element's concatenated descendant text, so a two-entry data set reads as the single +/// token "12" — 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 representation, and no value-shape heuristic can work, because +/// concatenated entries are indistinguishable from a legitimate space-bearing EVENT such as +/// a Message reading "Coolant level low". The one reliable discriminator — +/// that the observation element had element children — exists only while the XML is still +/// XML. +/// +/// +/// Deliberately a neutral fact about the wire shape, not a status: mapping it to a +/// quality code is the observation index's job (it codes these +/// BadNotSupported rather than publishing concatenated noise as Good). Defaults to +/// false, the shape of every ordinary scalar observation. +/// +/// public sealed record MTConnectObservation( string DataItemId, string Value, - DateTime TimestampUtc); + DateTime TimestampUtc, + bool IsStructured = false); /// /// Recursive helpers over the device/component tree. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs index 6f561050..6d2d7c51 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs @@ -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 <Component> element — so every element child of /// <Components> is a component. /// (3) The document is XML-namespaced and the namespace URI carries the MTConnect version -/// (urn:mtconnect.org:MTConnectDevices:1.3 … :2.0). Elements are therefore -/// matched on 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. xsi:type) can never be mistaken for a DataItem's type. +/// (urn:mtconnect.org:MTConnectDevices:1.3 … :2.0). Element matching and +/// attribute reading therefore go through , whose two rules +/// (match on local name, read attributes unqualified) are shared verbatim with +/// — see that type for why each rule exists. +/// +/// +/// One deliberate divergence from the streams parser: no BOM/whitespace trim here. +/// A /probe body arrives whole from HttpContent, so it begins exactly where +/// the Agent's document begins. A /sample 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 XmlReader rejects — hence the trim there and not here. +/// This asymmetry is intentional, not an oversight. /// /// /// Failure posture. Anything that is not a well-formed MTConnect device model throws @@ -54,6 +59,9 @@ internal static class MTConnectProbeParser /// private const int MaxComponentDepth = 64; + /// How 's shared error messages name this document. + private const string Subject = "/probe response"; + /// Parses a /probe response held as a string. /// The raw MTConnectDevices XML document. /// @@ -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 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 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 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."); - } - - /// - /// Reads an unqualified attribute, normalizing a present-but-empty value to null. - /// Only unqualified attributes are considered, so a namespace-prefixed attribute such as - /// xsi:type can never be read as a DataItem's type. - /// - 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; - } - - /// - /// Builds the message for a root element that is not MTConnectDevices, lifting the - /// Agent's own error text when the response is an MTConnectError document (which - /// Agents return under HTTP 200 for, e.g., an unknown device name). - /// - private static string DescribeUnexpectedRoot(XElement root) - { - var message = - $"MTConnect /probe response root element is <{root.Name.LocalName}>, expected ."; - - 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)}"; - } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs new file mode 100644 index 00000000..362dc16f --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs @@ -0,0 +1,161 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; + +/// +/// Base type for every way an Agent /sample long poll can stop delivering chunks for a +/// reason that is not the caller cancelling it. +/// +/// +/// +/// Why these are exceptions rather than a normal end of enumeration. +/// 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. +/// +/// +/// Catch this base type to handle "the stream is over" uniformly; catch the two derived +/// types to distinguish a transient end (the Agent closed a connection — reconnect) +/// from a configuration error (the Agent will never stream to this request — retrying +/// is pointless until an operator changes something). +/// +/// +public abstract class MTConnectStreamException : Exception +{ + /// Creates the exception with no message. + protected MTConnectStreamException() + { + } + + /// Creates the exception with a message. + /// The operator-facing description. + protected MTConnectStreamException(string message) + : base(message) + { + } + + /// Creates the exception with a message and an inner cause. + /// The operator-facing description. + /// The underlying cause. + protected MTConnectStreamException(string message, Exception innerException) + : base(message, innerException) + { + } +} + +/// How an Agent /sample stream ended. +public enum MTConnectStreamEndReason +{ + /// + /// 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 + /// . + /// + ConnectionClosed = 0, + + /// + /// The Agent wrote the closing --boundary-- delimiter, ending the multipart response + /// deliberately. Usually means the request was bounded (an Agent honouring a count + /// 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. + /// + ClosingBoundary = 1 +} + +/// +/// The Agent's /sample stream ended without the caller cancelling it. See +/// for why this is not a normal end of enumeration. +/// +public sealed class MTConnectStreamEndedException : MTConnectStreamException +{ + /// Creates the exception for a stream that ended for . + /// How the stream ended. + /// The /sample request URI whose stream ended. + /// How many chunks were yielded before the end. + 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; + } + + /// Creates the exception with a message. + /// The operator-facing description. + public MTConnectStreamEndedException(string message) + : base(message) + { + AgentUri = string.Empty; + } + + /// Creates the exception with a message and an inner cause. + /// The operator-facing description. + /// The underlying cause. + public MTConnectStreamEndedException(string message, Exception innerException) + : base(message, innerException) + { + AgentUri = string.Empty; + } + + /// Creates the exception with no message. + public MTConnectStreamEndedException() + { + AgentUri = string.Empty; + } + + /// How the stream ended. + public MTConnectStreamEndReason Reason { get; } + + /// The /sample request URI whose stream ended. + public string AgentUri { get; } + + /// How many chunks were yielded before the stream ended. + 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" + }; +} + +/// +/// The Agent answered a /sample request with something that cannot be consumed as a +/// stream at all — a single non-multipart document, or a multipart/* response with no +/// boundary parameter to frame it by. +/// +/// +/// This is a configuration error, not a transient one, and is deliberately typed apart +/// from : 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 multipart/x-mixed-replace). 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. +/// +public sealed class MTConnectStreamNotSupportedException : MTConnectStreamException +{ + /// Creates the exception with a message. + /// The operator-facing description. + public MTConnectStreamNotSupportedException(string message) + : base(message) + { + } + + /// Creates the exception with a message and an inner cause. + /// The operator-facing description. + /// The underlying cause. + public MTConnectStreamNotSupportedException(string message, Exception innerException) + : base(message, innerException) + { + } + + /// Creates the exception with no message. + public MTConnectStreamNotSupportedException() + { + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs index 9a9cfd21..20f0b642 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs @@ -79,6 +79,12 @@ internal static class MTConnectStreamsParser private const string ConditionContainer = "Condition"; + /// How 's shared error messages name this document. + private const string Subject = "streams response"; + + /// How those messages name the header element. + private const string HeaderOwner = "the
"; + /// /// 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
; 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 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); } /// @@ -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 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."); - } - - /// - /// Reads an unqualified attribute, normalizing a present-but-empty value to null. - /// Only unqualified attributes are considered, so a namespace-prefixed vendor attribute can - /// never be mistaken for an observation's identity or timestamp. - /// - 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
"); - - if (!long.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value)) - { - throw new InvalidDataException( - $"MTConnect streams response is malformed: the
has a non-numeric '{name}' attribute ('{raw}')."); - } - - return value; - } - - /// - /// Builds the message for a root element that is not MTConnectStreams, lifting the - /// Agent's own error text when the response is an MTConnectError document (which - /// Agents return under HTTP 200 for, e.g., a from outside the buffer). - /// - private static string DescribeUnexpectedRoot(XElement root) - { - var message = - $"MTConnect streams response root element is <{root.Name.LocalName}>, expected ."; - - 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)}"; - } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectXml.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectXml.cs new file mode 100644 index 00000000..63c31262 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectXml.cs @@ -0,0 +1,146 @@ +using System.Globalization; +using System.Xml.Linq; + +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; + +/// +/// The element/attribute reading rules shared by and +/// . 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. +/// +/// +/// +/// Rule 1: elements are matched on , ignoring the +/// namespace. The namespace URI carries the MTConnect version +/// (urn:mtconnect.org:MTConnectDevices:1.3 … :2.0), 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. +/// +/// +/// Rule 2: attributes are read unqualified. A prefixed attribute +/// (xsi:type on a document root, a vendor's x:dataItemId) must never be +/// mistaken for the real type / dataItemId, which would invent a data item the +/// probe never declared. +/// +/// +internal static class MTConnectXml +{ + /// Does have the given local name, whatever its namespace? + /// The element to test. + /// The unqualified name to match. + public static bool IsNamed(XElement element, string localName) => + string.Equals(element.Name.LocalName, localName, StringComparison.Ordinal); + + /// Direct element children of with the given local name. + /// The element whose children to filter. + /// The unqualified name to match. + public static IEnumerable ChildrenNamed(XElement parent, string localName) => + parent.Elements().Where(child => IsNamed(child, localName)); + + /// + /// Reads an unqualified attribute, normalizing a present-but-empty value to null. + /// See the type-level remarks for why only unqualified attributes are considered. + /// + /// The element carrying the attribute. + /// The unqualified attribute name. + public static string? OptionalAttribute(XElement element, string name) + { + var value = element.Attribute(name)?.Value; + + return string.IsNullOrWhiteSpace(value) ? null : value; + } + + /// Reads a required unqualified attribute, or throws naming the owner and the attribute. + /// The element carrying the attribute. + /// The unqualified attribute name. + /// How to describe the element in the error message. + /// How to describe the document in the error message (e.g. "/probe response"). + /// The attribute is absent or blank. + 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."); + } + + /// Reads an optional unqualified attribute as an . + /// The element carrying the attribute. + /// The unqualified attribute name. + /// How to describe the element in the error message. + /// How to describe the document in the error message. + /// The attribute is present but not an integer. + 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; + } + + /// Reads a required unqualified attribute as a . + /// The element carrying the attribute. + /// The unqualified attribute name. + /// How to describe the element in the error message. + /// How to describe the document in the error message. + /// The attribute is absent, blank, or not an integer. + 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; + } + + /// + /// 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 MTConnectError document — which + /// Agents return under HTTP 200, so EnsureSuccessStatusCode never fires and + /// this is the only place the operator can learn what the Agent objected to. + /// + /// The document's actual root element. + /// The root element name the caller required. + /// How to describe the document in the error message. + 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)}"; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs index 089aa83d..6df71fa6 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs @@ -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; /// /// Task 7 — the /current + /sample legs: /// turning an MTConnectStreams document into , the -/// ring-buffer sequence-gap signal (), and the +/// ring-buffer sequence-gap signal (), and the /// multipart/x-mixed-replace chunk framing + heartbeat watchdog on the long-poll leg. /// Every test runs against canned fixtures or a stubbed — 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(); - await foreach (var chunk in client.SampleAsync(from, Ct)) + await Should.ThrowAsync(async () => { - gaps.Add(MTConnectAgentClient.IsSequenceGap(from, chunk)); - from = chunk.NextSequence; - } + await foreach (var chunk in client.SampleAsync(from, Ct)) + { + // `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(); + var ex = await Should.ThrowAsync(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( + 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( + 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(); + await Should.ThrowAsync(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( + 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( + async () => await EnumerateAsync(client.SampleAsync(108, Ct))); + + ended.ShouldBeAssignableTo(); + 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 + + /// + /// Serving the whole body from one buffer completes every frame in a single + /// ReadAsync, 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. + /// + [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(async () => await pending); + ex.ShouldNotBeOfType(); + } + + [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(async () => await client.CurrentAsync(Ct)); + await Should.ThrowAsync( + 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(() => 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(() => 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(() => 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() + { + // 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( + """ + + 12 + + """)); + + 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( + """Coolant level low""")); + + 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( + """ + + vendor extension + + """)); + + 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> DrainAsync(IAsyncEnumerable source) { var results = new List(); - await foreach (var item in source) + try { - results.Add(item); + 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; } + /// Enumerates to completion, discarding chunks and letting every exception escape. + private static async Task EnumerateAsync(IAsyncEnumerable source) + { + await foreach (var _ in source) + { + // Drained for effect only. + } + } + + /// A streams document big enough to force the frame reader to grow its buffer. + private static string LargeStreamsDocument(int observationCount) + { + var samples = new StringBuilder(); + for (var i = 0; i < observationCount; i++) + { + samples.Append(CultureInfo.InvariantCulture, + $"""{i}.5"""); + } + + return $""" + +
+ + {samples} + + + """; + } + private static Dictionary ParseCurrent() => ParseFixture(CurrentFixture); private static Dictionary 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); + /// Serves an already-built multipart body verbatim (e.g. one with no closing delimiter). + public static StubHandler RespondingWithBody(string boundary, byte[] body) => + new(() => Multipart(boundary, new ByteArrayContent(body)), hang: false); + + /// + /// Serves a multipart body through a transport that hands back at most + /// bytes per ReadAsync, so boundaries, part + /// headers and documents are all split across reads. + /// + 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); + } + + /// A response that claims multipart but omits the boundary parameter. + 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); + /// Builds a multipart/x-mixed-replace body exactly as an Agent writes one. public static byte[] MultipartBody(string boundary, bool withContentLength, bool closing, params string[] parts) { @@ -769,6 +1185,97 @@ public sealed class MTConnectStreamsParseTests } } + /// Hands back at most bytesPerRead bytes per read — the split-transport double. + private sealed class ChunkedContent(byte[] body, int bytesPerRead) : HttpContent + { + protected override Task CreateContentReadStreamAsync() => + Task.FromResult(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 ReadAsync(Memory 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 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(); + } + + /// Captures warning-level log lines so the degraded-framing notice can be asserted. + private sealed class RecordingLogger : ILogger + { + public List Warnings { get; } = []; + + public IDisposable? BeginScope(TState state) + where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + if (logLevel == LogLevel.Warning) + { + Warnings.Add(formatter(state, exception)); + } + } + } + /// Serves , then never completes another read. private sealed class StallingContent(byte[] prefix) : HttpContent {