mxaccessgw/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs

using System;
using System.Collections.Generic;
using System.Threading;
using Google.Protobuf.Collections;
using Google.Protobuf.WellKnownTypes;
using ZB.MOM.WW.MxGateway.Contracts.Proto;

namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;

/// <summary>
///     Per-session cache of the most recent <c>OnDataChange</c> payload for
///     each (server handle, item handle) pair. Written by the MXAccess event
///     sink as new OnDataChange callbacks arrive; read by the ReadBulk command
///     executor so it can satisfy a "current value" request from a tag that is
///     already advised without modifying the existing subscription.
/// </summary>
/// <remarks>
///     Both writers and readers run on the worker's STA thread (COM dispatches
///     events on the apartment thread; commands also execute on the STA), so
///     no internal locking is required. The class is still nominally
///     thread-safe via a single sync root in case tests drive it from a
///     non-STA thread.
/// </remarks>
public sealed class MxAccessValueCache
{
    private readonly Dictionary<long, CachedValue> entries = new();
    private readonly object syncRoot = new();

    /// <summary>Records a fresh OnDataChange payload for the given handle pair.</summary>
    /// <param name="serverHandle">MXAccess server handle.</param>
    /// <param name="itemHandle">MXAccess item handle.</param>
    /// <param name="mxEvent">The protobuf MxEvent created by the event mapper.</param>
    public void Set(
        int serverHandle,
        int itemHandle,
        MxEvent mxEvent)
    {
        if (mxEvent is null)
        {
            throw new ArgumentNullException(nameof(mxEvent));
        }

        long key = CreateItemKey(serverHandle, itemHandle);
        lock (syncRoot)
        {
            ulong nextVersion = entries.TryGetValue(key, out CachedValue existing)
                ? existing.Version + 1
                : 1UL;

            entries[key] = new CachedValue(
                nextVersion,
                mxEvent.Value,
                mxEvent.Quality,
                mxEvent.SourceTimestamp,
                mxEvent.Statuses);
        }
    }

    /// <summary>Tries to read the most recent cached value for the handle pair.</summary>
    public bool TryGet(
        int serverHandle,
        int itemHandle,
        out CachedValue value)
    {
        long key = CreateItemKey(serverHandle, itemHandle);
        lock (syncRoot)
        {
            return entries.TryGetValue(key, out value);
        }
    }

    /// <summary>
    ///     Removes the cache slot for a handle pair. The session calls this
    ///     when an item is unregistered so stale values are not served to a
    ///     subsequent ReadBulk after a tag is removed and re-added.
    /// </summary>
    public void Remove(
        int serverHandle,
        int itemHandle)
    {
        long key = CreateItemKey(serverHandle, itemHandle);
        lock (syncRoot)
        {
            entries.Remove(key);
        }
    }

    /// <summary>
    ///     Waits until the cache entry's version exceeds <paramref name="sinceVersion"/>
    ///     or the deadline elapses, calling <paramref name="pumpStep"/> on every poll
    ///     iteration so the worker's STA can dispatch the inbound MXAccess message.
    /// </summary>
    /// <param name="serverHandle">MXAccess server handle.</param>
    /// <param name="itemHandle">MXAccess item handle.</param>
    /// <param name="sinceVersion">Version snapshot captured before the wait.</param>
    /// <param name="deadlineUtc">Absolute UTC deadline.</param>
    /// <param name="pumpStep">Action that pumps any pending Windows messages.</param>
    /// <param name="pollIntervalMs">How long to sleep between pump cycles. Default 5 ms.</param>
    public bool TryWaitForUpdate(
        int serverHandle,
        int itemHandle,
        ulong sinceVersion,
        DateTime deadlineUtc,
        Action pumpStep,
        out CachedValue value,
        int pollIntervalMs = 5)
    {
        if (pumpStep is null)
        {
            throw new ArgumentNullException(nameof(pumpStep));
        }

        while (true)
        {
            pumpStep();

            if (TryGet(serverHandle, itemHandle, out value) && value.Version > sinceVersion)
            {
                return true;
            }

            if (DateTime.UtcNow >= deadlineUtc)
            {
                return false;
            }

            Thread.Sleep(pollIntervalMs);
        }
    }

    /// <summary>Returns the current version for a handle pair, or 0 if no entry exists.</summary>
    public ulong CurrentVersion(
        int serverHandle,
        int itemHandle)
    {
        return TryGet(serverHandle, itemHandle, out CachedValue existing)
            ? existing.Version
            : 0UL;
    }

    private static long CreateItemKey(
        int serverHandle,
        int itemHandle)
    {
        return ((long)serverHandle << 32) | (uint)itemHandle;
    }

    /// <summary>
    ///     Snapshot of the most recent OnDataChange payload for a handle pair.
    ///     <see cref="Version"/> increments by one on every <see cref="Set"/>
    ///     call so the bulk read executor can detect "a new value arrived
    ///     since I started waiting".
    /// </summary>
    /// <remarks>
    ///     Plain readonly struct (not a record) so this compiles under the
    ///     worker's net48 target, which lacks <c>IsExternalInit</c>.
    /// </remarks>
    public readonly struct CachedValue
    {
        /// <summary>Initializes a new cached value snapshot.</summary>
        public CachedValue(
            ulong version,
            MxValue value,
            int quality,
            Timestamp sourceTimestamp,
            RepeatedField<MxStatusProxy> statuses)
        {
            Version = version;
            Value = value;
            Quality = quality;
            SourceTimestamp = sourceTimestamp;
            Statuses = statuses;
        }

        /// <summary>Monotonic per-handle version counter.</summary>
        public ulong Version { get; }

        /// <summary>The cached MxValue payload.</summary>
        public MxValue Value { get; }

        /// <summary>Quality code from the OnDataChange event.</summary>
        public int Quality { get; }

        /// <summary>Source timestamp from the OnDataChange event.</summary>
        public Timestamp SourceTimestamp { get; }

        /// <summary>MxStatusProxy entries from the OnDataChange event.</summary>
        public RepeatedField<MxStatusProxy> Statuses { get; }
    }
}