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

feat(mtconnect): current/sample parse + sequence-gap detection (Task 7)

Browse Source 
Adds the /current and /sample legs to MTConnectAgentClient:

- MTConnectStreamsParser: hand-rolled System.Xml.Linq parse of an
  MTConnectStreams document into MTConnectStreamsResult. Observations are
  every element child of <Samples>/<Events>/<Condition>, keyed by
  dataItemId (the element NAME is the data item's type in PascalCase, so
  a fixed name set would drop observations). A CONDITION's value is its
  element name, not its text - <Normal/> is empty - and <Unavailable/>
  normalizes onto the same UNAVAILABLE sentinel a Sample/Event carries as
  text, so Task 8's no-comms mapping sees one token. Timestamps are
  normalized to DateTimeKind.Utc explicitly.
- IsSequenceGap(requestedFrom, chunk) => chunk.FirstSequence >
  requestedFrom, so Task 11's pump can re-baseline after a ring-buffer
  overflow. Equality is NOT a gap.
- SampleAsync: ResponseHeadersRead + a hand-rolled
  multipart/x-mixed-replace frame reader (Content-length fast path,
  boundary-scan fallback), bounded by a heartbeat watchdog derived from
  HeartbeatMs so a silent Agent faults instead of wedging the pump - the
  S7 frozen-peer shape (arch-review R2-01). The long poll gets its own
  HttpClient with an infinite Timeout so the unary /probe + /current
  deadline stays bounded.

Task 0 package decision, option (c): both TrakHound PackageReferences and
their PackageVersions are dropped. MTConnectHttpClientStream takes a URL
and dials its own socket (no handler/stream seam, so it cannot serve a
socket-free unit suite), and it emits already-parsed documents through the
formatter lookup Task 6 proved is missing - framing-only reuse was never
actually on offer. The driver now depends on nothing but the BCL. Recorded
in the plan's Task 0 CORRECTION block.

Also folds in two Task 6 review carry-overs: the stale
IMTConnectAgentClient doc comment claiming a TrakHound-backed
implementation, and a probe test constructing a mixed-namespace vendor
extension (<x:Pump> with default-namespace <DataItems> children) that
backs the parser's ignore-the-namespace claim with a test.

187/187 green.
This commit is contained in:
Joseph Doherty
2026-07-24 14:32:36 -04:00
parent d3aaa4a411
commit cf43f8d0ab
9 changed files with 1740 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs
+8 -2
View File
@@ -9,8 +9,14 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// canned probe/current/sample XML, with no sockets involved.
/// </summary>
/// <remarks>
/// The production implementation (Task 6/7) is built on the TrakHound MTConnect.NET client
/// libraries (Task 0's decision); a fake for tests only needs to implement these three members.
/// The production implementation, <see cref="MTConnectAgentClient"/>, carries no backend NuGet
/// at all: it is an <see cref="HttpClient"/> over the Agent's three request paths, feeding
/// hand-rolled <c>System.Xml.Linq</c> parsers (<c>MTConnectProbeParser</c> /
/// <c>MTConnectStreamsParser</c>) plus a <c>multipart/x-mixed-replace</c> frame reader for the
/// <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.
/// </remarks>
public interface IMTConnectAgentClient
{
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs
+496 -21
View File
@@ -1,3 +1,7 @@
using System.Globalization;
using System.Runtime.CompilerServices;
using System.Text;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// <summary>
@@ -8,19 +12,37 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// <remarks>
/// <para>
/// <b>The constructor opens no connection.</b> It only composes the request URIs and builds
/// an <see cref="HttpClient"/> (which itself dials nothing until a request is issued), so a
/// throwaway instance is safe to construct — the universal discovery browser's
/// <c>CanBrowse</c> pattern depends on that.
/// its <see cref="HttpClient"/>s (which themselves dial nothing until a request is issued),
/// so a throwaway instance is safe to construct — the universal discovery browser's
/// <c>CanBrowse</c> pattern depends on that. <see cref="SampleAsync"/> likewise dials
/// nothing until it is enumerated.
/// </para>
/// <para>
/// <b>Every call is deadline-bounded.</b> Each request runs under a
/// <see cref="CancellationTokenSource"/> linked to the caller's token and cancelled after
/// <see cref="MTConnectDriverOptions.RequestTimeoutMs"/>, <i>and</i>
/// <see cref="HttpClient.Timeout"/> is set to the same bound. A frozen Agent surfaces as a
/// <see cref="TimeoutException"/> rather than wedging the caller — the lesson from this
/// repo's S7 read-leg wall-clock gap (arch-review R2-01). A non-positive timeout is rejected
/// at construction so an operator-authored <c>0</c> can never come to mean "wait forever"
/// (arch-review 01/S-6).
/// <b>Two <see cref="HttpClient"/>s, and the split is deliberate.</b>
/// <see cref="HttpClient.Timeout"/> is a per-instance, whole-response deadline, so a single
/// client cannot serve both legs: the unary <c>/probe</c> and <c>/current</c> calls must be
/// bounded by <see cref="MTConnectDriverOptions.RequestTimeoutMs"/>, while <c>/sample</c> is
/// 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).
/// </para>
/// <para>
/// <b>Every call is deadline-bounded, but not all by the same mechanism.</b> The unary legs
/// run under a <see cref="CancellationTokenSource"/> linked to the caller's token and
/// cancelled after <c>RequestTimeoutMs</c>. The streaming leg bounds its <i>header</i> phase
/// the same way (a peer that accepts the connection and never answers must not wedge the
/// pump), then hands the body to a <b>heartbeat watchdog</b>: MTConnect Agents emit a
/// keep-alive chunk every <see cref="MTConnectDriverOptions.HeartbeatMs"/> precisely so a
/// client can tell "quiet but alive" from "dead", so every body read is bounded by a
/// two-missed-heartbeat window and a silent Agent surfaces as a
/// <see cref="TimeoutException"/>. This is the exact shape of this repo's S7 frozen-peer
/// production defect (arch-review R2-01) — an async read that ignores the socket deadline —
/// 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).
/// </para>
/// <para>
/// This class does not swallow failures into an empty model: an unreachable Agent, a non-2xx
@@ -31,8 +53,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
{
private readonly HttpClient _http;
private readonly HttpClient _streamHttp;
private readonly Uri _probeUri;
private readonly Uri _currentUri;
private readonly string _sampleUriBase;
private readonly string _sampleQuerySuffix;
private readonly TimeSpan _requestTimeout;
private readonly TimeSpan _streamIdleTimeout;
/// <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>
@@ -61,10 +88,50 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
}
_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);
_probeUri = BuildRequestUri(options, "probe");
_currentUri = BuildRequestUri(options, "current");
_sampleUriBase = BuildRequestUri(options, "sample").AbsoluteUri;
_sampleQuerySuffix = string.Create(
CultureInfo.InvariantCulture,
$"&interval={options.SampleIntervalMs}&count={options.SampleCount}&heartbeat={options.HeartbeatMs}");
_http = handler is null ? new HttpClient() : new HttpClient(handler, disposeHandler: false);
_http.Timeout = _requestTimeout;
_streamHttp = handler is null ? new HttpClient() : new HttpClient(handler, disposeHandler: false);
// See the class remarks: a long poll must outlive the unary deadline by design, and the
// heartbeat watchdog — not this property — is what keeps it bounded.
_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/>
@@ -80,18 +147,73 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
}
/// <inheritdoc/>
public Task<MTConnectStreamsResult> CurrentAsync(CancellationToken ct) =>
// Task 7 adds the /current leg (MTConnectStreamsParser); deliberately not stubbed as an
// empty result, which would read as a working call that silently returns nothing.
throw new NotImplementedException("The MTConnect /current leg lands in Task 7.");
public async Task<MTConnectStreamsResult> CurrentAsync(CancellationToken ct)
{
using var response = await SendAsync(_currentUri, ct).ConfigureAwait(false);
await using var body = await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false);
return MTConnectStreamsParser.Parse(body);
}
/// <inheritdoc/>
public IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(long from, CancellationToken ct) =>
// Task 7 adds the /sample long-poll leg (multipart chunk framing + sequence-gap detection).
throw new NotImplementedException("The MTConnect /sample leg lands in Task 7.");
public async IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(
long from,
[EnumeratorCancellation] CancellationToken ct)
{
var uri = BuildSampleUri(from);
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);
await using var body = await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false);
var boundary = ResolveMultipartBoundary(response);
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));
yield break;
}
var reader = new MultipartStreamReader(body, boundary, _streamIdleTimeout, uri);
while (true)
{
var part = await reader.ReadNextPartAsync(ct).ConfigureAwait(false);
if (part is null)
{
yield break;
}
// A frame carrying nothing but whitespace is transport filler, not a document. The
// Agent's real keep-alive is a well-formed observation-free MTConnectStreams document,
// and that one IS yielded — it advances the sequence, so swallowing it would strand the
// caller's `from` behind the Agent's buffer.
if (string.IsNullOrWhiteSpace(part))
{
continue;
}
yield return MTConnectStreamsParser.Parse(part);
}
}
/// <inheritdoc/>
public void Dispose() => _http.Dispose();
public void Dispose()
{
_http.Dispose();
_streamHttp.Dispose();
}
private Uri BuildSampleUri(long from) =>
new(string.Create(CultureInfo.InvariantCulture, $"{_sampleUriBase}?from={from}{_sampleQuerySuffix}"),
UriKind.Absolute);
private async Task<HttpResponseMessage> SendAsync(Uri uri, CancellationToken ct)
{
@@ -108,8 +230,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
// 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.
throw new TimeoutException(
$"MTConnect Agent request to '{uri}' did not complete within {_requestTimeout.TotalMilliseconds:F0} ms.");
throw TimedOut(uri);
}
response.EnsureSuccessStatusCode();
@@ -117,6 +238,73 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
return response;
}
/// <summary>
/// Issues the <c>/sample</c> request and returns as soon as the response headers land. Only
/// 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)
{
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
deadline.CancelAfter(_requestTimeout);
HttpResponseMessage response;
try
{
response = await _streamHttp
.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, deadline.Token)
.ConfigureAwait(false);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
throw TimedOut(uri);
}
response.EnsureSuccessStatusCode();
return response;
}
/// <summary>Reads a non-multipart streaming body whole, under the unary deadline.</summary>
private async Task<string> ReadWholeBodyAsync(Stream body, Uri uri, CancellationToken ct)
{
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
deadline.CancelAfter(_requestTimeout);
using var reader = new StreamReader(body, Encoding.UTF8);
try
{
return await reader.ReadToEndAsync(deadline.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
throw TimedOut(uri);
}
}
private TimeoutException TimedOut(Uri uri) =>
new($"MTConnect Agent request to '{uri}' did not complete within {_requestTimeout.TotalMilliseconds:F0} ms.");
/// <summary>
/// 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)
{
var contentType = response.Content.Headers.ContentType;
if (contentType?.MediaType is null ||
!contentType.MediaType.StartsWith("multipart/", StringComparison.OrdinalIgnoreCase))
{
return null;
}
var boundary = contentType.Parameters
.FirstOrDefault(p => string.Equals(p.Name, "boundary", StringComparison.OrdinalIgnoreCase))?.Value?
.Trim('"');
return string.IsNullOrEmpty(boundary) ? null : boundary;
}
private static Uri BuildRequestUri(MTConnectDriverOptions options, string request)
{
var agentUri = options.AgentUri?.Trim();
@@ -139,4 +327,291 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
return new Uri($"{root}{deviceSegment}/{request}", UriKind.Absolute);
}
/// <summary>
/// Frames a <c>multipart/x-mixed-replace</c> body into individual part payloads, reading
/// incrementally and never buffering more than the part currently in flight.
/// </summary>
/// <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.
/// </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.
/// </para>
/// <para>
/// <b>Every read is watchdog-bounded.</b> A silent peer faults the stream rather than
/// parking a task forever (arch-review R2-01), and the buffer is capped so a hostile or
/// malfunctioning Agent cannot grow it without limit.
/// </para>
/// </remarks>
private sealed class MultipartStreamReader(Stream stream, string boundary, TimeSpan idleTimeout, Uri uri)
{
private const int MaxBufferBytes = 32 * 1024 * 1024;
private static readonly byte[] CrLfCrLf = "\r\n\r\n"u8.ToArray();
private static readonly byte[] LfLf = "\n\n"u8.ToArray();
private readonly byte[] _delimiter = Encoding.ASCII.GetBytes("--" + boundary);
private byte[] _buffer = new byte[16 * 1024];
private int _length;
private bool _finished;
/// <summary>
/// Reads the next part's payload, or <c>null</c> once the closing delimiter is seen or
/// the Agent closes the connection.
/// </summary>
public async Task<string?> ReadNextPartAsync(CancellationToken ct)
{
if (_finished)
{
return null;
}
if (!await SkipThroughDelimiterAsync(ct).ConfigureAwait(false) ||
!await EnsureAsync(2, ct).ConfigureAwait(false))
{
_finished = true;
return null;
}
// "--boundary--" closes the stream.
if (_buffer[0] == (byte)'-' && _buffer[1] == (byte)'-')
{
_finished = true;
return null;
}
var headerEnd = await FindHeaderEndAsync(ct).ConfigureAwait(false);
if (headerEnd < 0)
{
_finished = true;
return null;
}
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);
return body is null ? null : Encoding.UTF8.GetString(body);
}
private async Task<byte[]?> ReadByLengthAsync(int declared, CancellationToken ct)
{
if (!await EnsureAsync(declared, ct).ConfigureAwait(false))
{
_finished = true;
return null;
}
var body = _buffer[..declared];
Consume(declared);
return body;
}
private async Task<byte[]?> ReadToNextDelimiterAsync(CancellationToken ct)
{
var next = await FindDelimiterAsync(ct).ConfigureAwait(false);
if (next < 0)
{
// EOF with no closing delimiter: whatever is buffered is the last part.
var tail = _buffer[.._length];
_length = 0;
_finished = true;
return tail;
}
var body = _buffer[..TrimTrailingLineBreak(next)];
// Leave the delimiter itself in place; the next call's skip consumes it.
Consume(next);
return body;
}
/// <summary>Strips the single line break that terminates a part body before its delimiter.</summary>
private int TrimTrailingLineBreak(int end)
{
if (end >= 2 && _buffer[end - 2] == (byte)'\r' && _buffer[end - 1] == (byte)'\n')
{
return end - 2;
}
return end >= 1 && _buffer[end - 1] == (byte)'\n' ? end - 1 : end;
}
/// <summary>Advances past the next boundary delimiter; <c>false</c> at end of stream.</summary>
private async Task<bool> SkipThroughDelimiterAsync(CancellationToken ct)
{
var index = await FindDelimiterAsync(ct).ConfigureAwait(false);
if (index < 0)
{
return false;
}
Consume(index + _delimiter.Length);
return true;
}
private async Task<int> FindDelimiterAsync(CancellationToken ct)
{
var searchedTo = 0;
while (true)
{
var index = IndexOf(_delimiter, Math.Max(0, searchedTo - (_delimiter.Length - 1)));
if (index >= 0)
{
return index;
}
searchedTo = _length;
if (!await FillAsync(ct).ConfigureAwait(false))
{
return -1;
}
}
}
/// <summary>
/// Finds the end of the part's header block — the first blank line after the delimiter.
/// A part with no headers at all is legal, in which case the blank line sits at offset 0.
/// </summary>
private async Task<int> FindHeaderEndAsync(CancellationToken ct)
{
var searchedTo = 0;
while (true)
{
var crlf = IndexOf(CrLfCrLf, Math.Max(0, searchedTo - 3));
var lf = IndexOf(LfLf, Math.Max(0, searchedTo - 1));
if (crlf >= 0 && (lf < 0 || crlf <= lf))
{
return crlf + CrLfCrLf.Length;
}
if (lf >= 0)
{
return lf + LfLf.Length;
}
searchedTo = _length;
if (!await FillAsync(ct).ConfigureAwait(false))
{
return -1;
}
}
}
private static int? ParseContentLength(string headers)
{
foreach (var line in headers.Split('\n'))
{
var trimmed = line.Trim();
if (!trimmed.StartsWith("content-length:", StringComparison.OrdinalIgnoreCase))
{
continue;
}
var raw = trimmed["content-length:".Length..].Trim();
return int.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value) &&
value >= 0
? value
: null;
}
return null;
}
private async Task<bool> EnsureAsync(int count, CancellationToken ct)
{
while (_length < count)
{
if (!await FillAsync(ct).ConfigureAwait(false))
{
return false;
}
}
return true;
}
/// <summary>
/// Reads at least one more byte, bounded by the heartbeat watchdog. Returns <c>false</c>
/// at end of stream and throws <see cref="TimeoutException"/> when the Agent goes silent.
/// </summary>
private async Task<bool> FillAsync(CancellationToken ct)
{
if (_length == _buffer.Length)
{
if (_buffer.Length >= MaxBufferBytes)
{
throw new InvalidDataException(
$"MTConnect /sample stream from '{uri}' sent a part larger than {MaxBufferBytes} bytes without a boundary; refusing to buffer further.");
}
Array.Resize(ref _buffer, Math.Min(_buffer.Length * 2, MaxBufferBytes));
}
using var watchdog = CancellationTokenSource.CreateLinkedTokenSource(ct);
watchdog.CancelAfter(idleTimeout);
int read;
try
{
read = await stream.ReadAsync(_buffer.AsMemory(_length), watchdog.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
throw new TimeoutException(
$"MTConnect /sample stream from '{uri}' sent no data (not even a heartbeat) within {idleTimeout.TotalMilliseconds:F0} ms; treating the Agent as unreachable rather than waiting indefinitely.");
}
if (read <= 0)
{
return false;
}
_length += read;
return true;
}
private int IndexOf(byte[] pattern, int start)
{
if (start >= _length)
{
return -1;
}
var index = _buffer.AsSpan(start, _length - start).IndexOf(pattern.AsSpan());
return index < 0 ? -1 : index + start;
}
private void Consume(int count)
{
Buffer.BlockCopy(_buffer, count, _buffer, 0, _length - count);
_length -= count;
}
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs
+2 -2
View File
@@ -11,8 +11,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// </summary>
/// <remarks>
/// <para>
/// <b>Why hand-rolled rather than TrakHound.</b> The referenced TrakHound packages
/// (<c>MTConnect.NET-Common</c> + <c>-HTTP</c> 6.9.0.2) contain <b>no</b>
/// <b>Why hand-rolled rather than TrakHound.</b> The TrakHound packages Task 0 pinned
/// (<c>MTConnect.NET-Common</c> + <c>-HTTP</c> 6.9.0.2 — both dropped in Task 7) contain <b>no</b>
/// <c>IResponseDocumentFormatter</c> implementation — the XML formatter ships in the
/// separate, unreferenced <c>MTConnect.NET-XML</c> package — so
/// <c>ResponseDocumentFormatter.CreateDevicesResponseDocument("xml", stream)</c> answers
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs
+296
View File
@@ -0,0 +1,296 @@
using System.Globalization;
using System.Xml;
using System.Xml.Linq;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// <summary>
/// Turns an MTConnect Agent <c>MTConnectStreams</c> document — a <c>/current</c> snapshot or one
/// chunk of a <c>/sample</c> multipart stream — into the neutral
/// <see cref="MTConnectStreamsResult"/>. Deliberately a pure, socket-free static so the whole
/// parse is unit-testable against canned fixtures; chunk framing lives in
/// <see cref="MTConnectAgentClient"/> and hands this parser one chunk's XML at a time.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why hand-rolled rather than TrakHound.</b> Same finding as
/// <see cref="MTConnectProbeParser"/>: the packages pinned by Task 0 ship no XML document
/// formatter, and TrakHound's streaming client offers no socket-free entry point. Both
/// package references were dropped in Task 7; the driver depends only on the BCL.
/// </para>
/// <para>
/// <b>The Streams document is shaped nothing like the Devices document</b>, and three
/// differences drive this parser's design.
/// </para>
/// <list type="number">
/// <item>
/// <description>
/// <b>The observation element's name is the data item's type in PascalCase</b>
/// (<c>Position</c>, <c>PartCount</c>, <c>Execution</c>) — not the probe's
/// UPPER_SNAKE <c>POSITION</c> / <c>PART_COUNT</c>. The standard defines hundreds of
/// types and vendors add more, so matching a fixed set of element names would drop
/// observations silently. <b>Every</b> element child of <c>&lt;Samples&gt;</c>,
/// <c>&lt;Events&gt;</c> and <c>&lt;Condition&gt;</c> is therefore an observation,
/// keyed by its <c>dataItemId</c> attribute — the only identity the probe model and
/// the streams document actually share.
/// </description>
/// </item>
/// <item>
/// <description>
/// <b>A CONDITION observation's value is its ELEMENT NAME, not its text.</b>
/// <c>&lt;Normal/&gt;</c> / <c>&lt;Warning/&gt;</c> / <c>&lt;Fault/&gt;</c> /
/// <c>&lt;Unavailable/&gt;</c> carry the state in the name and are typically empty;
/// where text is present it is the operator-facing message, not the state. Reading
/// the text would report every condition as an empty string.
/// <c>&lt;Unavailable/&gt;</c> is normalized to the literal
/// <see cref="UnavailableSentinel"/> so a no-comms condition lands on exactly the
/// same sentinel as a Sample/Event whose text is <c>UNAVAILABLE</c> — the
/// observation index maps that one token to bad quality.
/// </description>
/// </item>
/// <item>
/// <description>
/// <b>Timestamps are normalized to UTC explicitly.</b> MTConnect timestamps are
/// ISO-8601 Zulu, but a bare <c>DateTime.Parse</c> yields <c>Local</c> or
/// <c>Unspecified</c> depending on machine settings — and this value becomes the
/// OPC UA SourceTimestamp, so a wrong <see cref="DateTimeKind"/> shifts every
/// timestamp by the host's UTC offset with nothing to show for it.
/// </description>
/// </item>
/// </list>
/// <para>
/// <b>Failure posture.</b> A malformed document, a non-<c>MTConnectStreams</c> root (an Agent
/// <c>MTConnectError</c> served under HTTP 200 included), an unusable header, or an
/// observation missing its <c>dataItemId</c>/<c>timestamp</c> all throw
/// <see cref="InvalidDataException"/>. A document with a valid header and <i>zero</i>
/// observations, by contrast, is entirely legal — <c>/sample</c> chunks are deltas and an
/// idle Agent's keep-alive is an observation-free document — so it parses to an empty
/// observation list rather than throwing.
/// </para>
/// </remarks>
internal static class MTConnectStreamsParser
{
/// <summary>
/// The single token that means "the Agent has no value for this data item". Sample/Event
/// observations carry it as text; a CONDITION carries it as the element name
/// <c>&lt;Unavailable/&gt;</c> and is normalized onto this spelling here.
/// </summary>
private const string UnavailableSentinel = "UNAVAILABLE";
private const string ConditionContainer = "Condition";
/// <summary>
/// The three elements whose element children are observations. A ComponentStream may carry
/// any subset of them, including none.
/// </summary>
private static readonly string[] ObservationContainers = ["Samples", "Events", ConditionContainer];
/// <summary>Parses a <c>/current</c> response, or one <c>/sample</c> chunk, held as a string.</summary>
/// <param name="xml">The raw <c>MTConnectStreams</c> XML document.</param>
/// <exception cref="InvalidDataException">
/// The payload is empty, not well-formed XML, not an <c>MTConnectStreams</c> document (an
/// Agent <c>MTConnectError</c> included — Agents answer a bad request with one under HTTP
/// 200), carries no usable <c>&lt;Header&gt;</c>, or carries a malformed observation.
/// </exception>
public static MTConnectStreamsResult Parse(string xml)
{
if (string.IsNullOrWhiteSpace(xml))
{
throw new InvalidDataException(
"MTConnect streams response was empty; expected an MTConnectStreams XML document.");
}
XDocument document;
try
{
// 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'));
}
catch (XmlException ex)
{
throw new InvalidDataException($"MTConnect streams response is not well-formed XML: {ex.Message}", ex);
}
return Build(document);
}
/// <summary>Parses a <c>/current</c> response read from a stream.</summary>
/// <param name="stream">A stream positioned at the start of the XML document.</param>
/// <exception cref="InvalidDataException">As for <see cref="Parse(string)"/>.</exception>
public static MTConnectStreamsResult Parse(Stream stream)
{
ArgumentNullException.ThrowIfNull(stream);
XDocument document;
try
{
document = XDocument.Load(stream);
}
catch (XmlException ex)
{
throw new InvalidDataException($"MTConnect streams response is not well-formed XML: {ex.Message}", ex);
}
return Build(document);
}
private static MTConnectStreamsResult Build(XDocument document)
{
var root = document.Root
?? throw new InvalidDataException("MTConnect streams response has no root element.");
if (!IsNamed(root, "MTConnectStreams"))
{
throw new InvalidDataException(DescribeUnexpectedRoot(root));
}
var header = 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"),
ReadObservations(root));
}
/// <summary>
/// Collects every observation in the document, in Agent-supplied order. Containers are found
/// by descending the whole <c>&lt;Streams&gt;</c> subtree rather than by walking a fixed
/// DeviceStream → ComponentStream chain, so a vendor's extra nesting level cannot silently
/// hide a device's observations.
/// </summary>
private static IReadOnlyList<MTConnectObservation> ReadObservations(XElement root)
{
var observations = new List<MTConnectObservation>();
foreach (var container in root.Descendants().Where(IsObservationContainer))
{
var isCondition = IsNamed(container, ConditionContainer);
foreach (var element in container.Elements())
{
observations.Add(ReadObservation(element, isCondition));
}
}
return observations;
}
private static MTConnectObservation ReadObservation(XElement element, bool isCondition)
{
var dataItemId = RequiredAttribute(element, "dataItemId", $"Observation <{element.Name.LocalName}>");
return new MTConnectObservation(
dataItemId,
isCondition ? ConditionState(element) : element.Value.Trim(),
ReadTimestampUtc(element, dataItemId));
}
/// <summary>
/// A condition's state is its element name. <c>Unavailable</c> is normalized to the
/// <see cref="UnavailableSentinel"/> spelling so downstream quality mapping has exactly one
/// token to recognize; every other state is reported as-named (<c>Normal</c>, <c>Warning</c>,
/// <c>Fault</c>, and any vendor state).
/// </summary>
private static string ConditionState(XElement element)
{
var state = element.Name.LocalName;
return string.Equals(state, "Unavailable", StringComparison.OrdinalIgnoreCase) ? UnavailableSentinel : state;
}
private static DateTime ReadTimestampUtc(XElement element, string dataItemId)
{
var raw = RequiredAttribute(element, "timestamp", $"Observation '{dataItemId}'");
if (!DateTime.TryParse(
raw,
CultureInfo.InvariantCulture,
DateTimeStyles.AdjustToUniversal | DateTimeStyles.AssumeUniversal,
out var parsed))
{
throw new InvalidDataException(
$"MTConnect streams response is malformed: observation '{dataItemId}' has an unparseable 'timestamp' ('{raw}').");
}
// AdjustToUniversal already yields Utc; stated explicitly so a future styles change cannot
// quietly hand the OPC UA layer a Local or Unspecified SourceTimestamp.
return DateTime.SpecifyKind(parsed, DateTimeKind.Utc);
}
private static bool IsObservationContainer(XElement element) =>
ObservationContainers.Any(name => 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/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.csproj
+6 -7
View File
@@ -17,13 +17,12 @@
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts\ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts.csproj"/>
</ItemGroup>
<ItemGroup>
<!-- Task 0 decision (verified 2026-07-24 against real nuget.org): TrakHound MTConnect.NET,
MIT, netstandard2.0 lib (resolves on net10 via the in-box compat mapping). Pulls
MTConnect.NET-TLS 6.9.0.2 transitively via -HTTP — expected, not a rogue dependency. -->
<PackageReference Include="MTConnect.NET-Common"/>
<PackageReference Include="MTConnect.NET-HTTP"/>
</ItemGroup>
<!-- No backend NuGet: the Agent surface is plain HTTP + XML, served entirely by the BCL
(HttpClient + System.Xml.Linq). The TrakHound MTConnect.NET-Common/-HTTP references Task 0
added were removed in Task 7 — they can neither parse an MTConnect document (no XML
formatter ships in either package) nor frame the /sample stream socket-free (their client
type dials its own URL). See the Task 0 CORRECTION block in
docs/plans/2026-07-24-mtconnect-driver.md. -->
<ItemGroup>
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests"/>