lmxopcua/src/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared/FrameWriter.cs

using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using MessagePack;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared;

/// <summary>
///     Writes length-prefixed, kind-tagged MessagePack frames to a stream. Thread-safe via
///     <see cref="SemaphoreSlim"/> — multiple producers (e.g. heartbeat + data-plane sharing a stream)
///     get serialized writes.
/// </summary>
public sealed class FrameWriter : IDisposable
{
    private readonly Stream _stream;
    private readonly SemaphoreSlim _gate = new(1, 1);
    private readonly bool _leaveOpen;

    public FrameWriter(Stream stream, bool leaveOpen = false)
    {
        _stream = stream ?? throw new ArgumentNullException(nameof(stream));
        _leaveOpen = leaveOpen;
    }

    public async Task WriteAsync<T>(MessageKind kind, T message, CancellationToken ct)
    {
        var body = MessagePackSerializer.Serialize(message, cancellationToken: ct);
        if (body.Length > Framing.MaxFrameBodyBytes)
            throw new InvalidOperationException(
                $"IPC frame body {body.Length} exceeds {Framing.MaxFrameBodyBytes} byte cap.");

        var lengthPrefix = new byte[Framing.LengthPrefixSize];
        // Big-endian — easy to read in hex dumps.
        lengthPrefix[0] = (byte)((body.Length >> 24) & 0xFF);
        lengthPrefix[1] = (byte)((body.Length >> 16) & 0xFF);
        lengthPrefix[2] = (byte)((body.Length >> 8)  & 0xFF);
        lengthPrefix[3] = (byte)( body.Length        & 0xFF);

        await _gate.WaitAsync(ct).ConfigureAwait(false);
        try
        {
            await _stream.WriteAsync(lengthPrefix, 0, lengthPrefix.Length, ct).ConfigureAwait(false);
            _stream.WriteByte((byte)kind);
            await _stream.WriteAsync(body, 0, body.Length, ct).ConfigureAwait(false);
            await _stream.FlushAsync(ct).ConfigureAwait(false);
        }
        finally { _gate.Release(); }
    }

    public void Dispose()
    {
        _gate.Dispose();
        if (!_leaveOpen) _stream.Dispose();
    }
}