using System; using System.IO; using System.Threading; using System.Threading.Tasks; using MessagePack; using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Shared.Contracts; namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Shared; /// /// Reads length-prefixed, kind-tagged frames from a stream. Single-consumer — do not call /// from multiple threads against the same instance. /// public sealed class FrameReader : IDisposable { private readonly Stream _stream; private readonly bool _leaveOpen; public FrameReader(Stream stream, bool leaveOpen = false) { _stream = stream ?? throw new ArgumentNullException(nameof(stream)); _leaveOpen = leaveOpen; } public async Task<(FocasMessageKind Kind, byte[] Body)?> ReadFrameAsync(CancellationToken ct) { var lengthPrefix = new byte[Framing.LengthPrefixSize]; if (!await ReadExactAsync(lengthPrefix, ct).ConfigureAwait(false)) return null; var length = (lengthPrefix[0] << 24) | (lengthPrefix[1] << 16) | (lengthPrefix[2] << 8) | lengthPrefix[3]; if (length < 0 || length > Framing.MaxFrameBodyBytes) throw new InvalidDataException($"IPC frame length {length} out of range."); var kindByte = _stream.ReadByte(); if (kindByte < 0) throw new EndOfStreamException("EOF after length prefix, before kind byte."); var body = new byte[length]; if (!await ReadExactAsync(body, ct).ConfigureAwait(false)) throw new EndOfStreamException("EOF mid-frame."); return ((FocasMessageKind)(byte)kindByte, body); } public static T Deserialize(byte[] body) => MessagePackSerializer.Deserialize(body); private async Task ReadExactAsync(byte[] buffer, CancellationToken ct) { var offset = 0; while (offset < buffer.Length) { var read = await _stream.ReadAsync(buffer, offset, buffer.Length - offset, ct).ConfigureAwait(false); if (read == 0) { if (offset == 0) return false; throw new EndOfStreamException($"Stream ended after reading {offset} of {buffer.Length} bytes."); } offset += read; } return true; } public void Dispose() { if (!_leaveOpen) _stream.Dispose(); } }