diff --git a/docs/GatewayDashboardDesign.md b/docs/GatewayDashboardDesign.md index 5267fdc..d2327fa 100644 --- a/docs/GatewayDashboardDesign.md +++ b/docs/GatewayDashboardDesign.md @@ -165,7 +165,7 @@ bearer). Each hub class is `[Authorize(Policy = HubClientsPolicy)]`. | Hub | Path | Producer | Payload | Routing | |---|---|---|---|---| -| `DashboardSnapshotHub` | `/hubs/snapshot` | `DashboardSnapshotPublisher` (BackgroundService consuming `IDashboardSnapshotService.WatchSnapshotsAsync`) | `DashboardSnapshot` | Sent to all connected clients on every snapshot tick; new connections receive the current snapshot synchronously in `OnConnectedAsync`. | +| `DashboardSnapshotHub` | `/hubs/snapshot` | `DashboardSnapshotPublisher` (BackgroundService consuming `IDashboardSnapshotService.WatchSnapshotsAsync`) | `DashboardSnapshot` | Sent to all connected clients on every snapshot tick, but only while at least one client is connected (see "Idle gating" below); new connections receive the current snapshot synchronously in `OnConnectedAsync`. | | `AlarmsHub` | `/hubs/alarms` | `AlarmsHubPublisher` (BackgroundService consuming `IGatewayAlarmService.StreamAsync(filter: null)`) | `AlarmFeedMessage` (`active_alarm` / `snapshot_complete` / `transition`) | Connected clients auto-join `__alarms__`; all clients receive every message. Publisher auto-reconnects every 5s on stream faults. | | `EventsHub` | `/hubs/events` | `DashboardEventBroadcaster` invoked by each session's internal dashboard-mirror subscriber on its `SessionEventDistributor` (registered when the session becomes Ready) | `MxEvent` | Clients call `SubscribeSession(sessionId)` to join `session:{id}`, which also registers them in `EventsHubViewerRegistry` — the mirror is gated on that registry (see "Mirror gating" below). The dashboard is a first-class distributor subscriber, so it receives the session's events whether or not a gRPC client is streaming. It sees RAW session events — not the per-gRPC-subscriber `AfterWorkerSequence` filtering that `EventStreamService` applies at its own boundary — because the dashboard is a separate LDAP-authenticated monitoring view meant to show the session's full event activity. Tag values are stripped from the mirrored `MxEvent` copy by `DashboardEventBroadcaster` when `Dashboard:ShowTagValues` is false (the default) — event metadata (tag reference, quality, status, timestamps) still renders, but the value fields are blanked, so no value leaks through this seam. The per-session hub ACL that would scope a Viewer to specific sessions is still outstanding (SEC-25 / remediation roadmap item 12); the value redaction is the near-term hardening that closes the value-leak seam independently of that ACL. | @@ -187,6 +187,38 @@ Default cadences: - event publisher emits per event fanned by the session's `SessionEventDistributor` to its internal dashboard-mirror subscriber (independent of any gRPC `StreamEvents`). +### Idle gating and snapshot cost + +A snapshot is not free: each one takes a session-registry snapshot and sorts it, +copies the metrics dictionaries under the global metrics lock, and projects +sessions, workers, faults, and the Galaxy summary. Without gating that work ran +once a second for the life of the process even when no browser was connected. + +`DashboardSnapshotHub` counts live connections into the singleton +`DashboardSnapshotHubConnectionCounter` (`OnConnectedAsync` / `OnDisconnectedAsync`, +clamped at zero). `DashboardSnapshotPublisher` reads that count before advancing the +snapshot enumerator: while it is zero the publisher does not call `MoveNextAsync` at +all, so the producing iterator stays suspended at its `yield` and builds nothing — +the gate removes the snapshot *build*, not just the broadcast. The publisher +re-checks once a second while idle, so the first viewer to connect resumes the tick +within roughly one snapshot interval. That viewer does not wait for it either: +`DashboardPageBase` seeds its first render synchronously from +`IDashboardSnapshotService.GetSnapshot()`, and `OnConnectedAsync` pushes a snapshot +to the new connection immediately. + +Two per-tick costs inside the snapshot itself are bounded independently of the gate: + +- the effective configuration (`EffectiveGatewayConfiguration`) is built once and + cached. It is a projection of `IOptions`, which the gateway binds + at startup and never reloads, so rebuilding the whole option tree every tick + produced an identical object; +- the API key summaries are refreshed at most once every 15 seconds + (`ApiKeySummaryRefreshInterval`) instead of on every tick. The list is a SQLite + read whose content changes only when an operator creates, rotates, or revokes a + key, so a key change reaches the dashboard within that interval. Only a + *successful* refresh restarts the interval, so a failed or timed-out read is + retried on the next tick and the previous summaries stay on screen. + Avoid pushing every MXAccess data-change event into a wider broadcast group. The current design routes events strictly through `session:{id}` groups; the snapshot hub continues to carry aggregate event counters and rates. @@ -362,6 +394,18 @@ its lease expires. One session means one worker process backs every dashboard circuit; all access is serialised so the worker sees one in-flight command at a time. Tag reads go through `GatewaySession.SubscribeBulkAsync` / `ReadBulkAsync`. +The advise set that backs those reads is capped at 256 tags (one browse page plus +headroom) and evicted least-recently-read-first. Without the cap every tag any +viewer ever inspected stayed advised on the single dashboard worker until the +session faulted, so browsing a large galaxy accreted unbounded live MXAccess +subscriptions — and the event churn they feed — on one x86 process. Reading a tag +already in the set marks it most-recently-read; subscribing past the cap unadvises +the oldest entries with `GatewaySession.UnsubscribeBulkAsync` in one batch before +the new ones are advised. Tags read in the same call are never evicted to make +room for each other. A failed unadvise does not fail the read: the tags are +dropped from tracking anyway (they re-subscribe if read again), because the +session-invalidation path already handles gateway/worker drift. + The Alarms page does **not** use the dashboard session: alarm data comes from the gateway's always-on central monitor. `QueryAlarmsAsync` reads `IGatewayAlarmService.CurrentAlarms` — the monitor's in-process cache — so the diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs index 55e9c56..d28a44c 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs @@ -15,13 +15,26 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync { private const string BackendName = "Galaxy"; private const string ClientName = "mxgateway-dashboard"; + + // One browse page of tags plus headroom. Bounds the standing advise load the + // single dashboard worker carries — and the event churn that advise set feeds — + // however much of a galaxy an operator browses through in one sitting. + private const int MaxSubscribedTags = 256; + private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(5); private readonly ISessionManager _sessionManager; private readonly IGatewayAlarmService _alarmService; private readonly ILogger _logger; private readonly SemaphoreSlim _gate = new(1, 1); - private readonly HashSet _subscribed = new(StringComparer.OrdinalIgnoreCase); + + // Least-recently-read-last advise set: the list holds every currently advised + // tag ordered most- to least-recently read, the dictionary indexes into it. + // Both are only ever touched under _gate, which already serialises all viewers. + private readonly Dictionary> _subscribed = + new(StringComparer.OrdinalIgnoreCase); + + private readonly LinkedList _recency = new(); private GatewaySession? _session; private int _serverHandle; @@ -58,15 +71,15 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync (GatewaySession session, int serverHandle) = await EnsureReadyAsync(cancellationToken) .ConfigureAwait(false); - string[] toSubscribe = tagAddresses.Where(tag => !_subscribed.Contains(tag)).ToArray(); + string[] toSubscribe = TouchAndCollectNewTags(tagAddresses, out int justReadCount); if (toSubscribe.Length > 0) { - await session.SubscribeBulkAsync(serverHandle, toSubscribe, cancellationToken) + await EvictForAsync(session, serverHandle, toSubscribe.Length, justReadCount, cancellationToken) .ConfigureAwait(false); - foreach (string tag in toSubscribe) - { - _subscribed.Add(tag); - } + IReadOnlyList subscribeResults = await session + .SubscribeBulkAsync(serverHandle, toSubscribe, cancellationToken) + .ConfigureAwait(false); + TrackSubscribed(toSubscribe, subscribeResults); } IReadOnlyList results = await session @@ -107,6 +120,131 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync return Task.FromResult(new DashboardAlarmQueryResult(alarms, error, _alarmService.WorkerProcessId)); } + // Promotes every already-advised tag in this read to the front of the recency + // list and returns the tags that still need subscribing (distinct, in request + // order). `justReadCount` is how many distinct tags of this read were already + // advised — they now occupy the front of the list and must never be evicted to + // make room for the same read's new tags. Callers must hold _gate. + private string[] TouchAndCollectNewTags(IReadOnlyCollection tagAddresses, out int justReadCount) + { + int touched = 0; + List toSubscribe = []; + HashSet seen = new(StringComparer.OrdinalIgnoreCase); + + foreach (string tag in tagAddresses) + { + if (_subscribed.TryGetValue(tag, out LinkedListNode? node)) + { + if (!ReferenceEquals(node, _recency.First)) + { + _recency.Remove(node); + _recency.AddFirst(node); + } + + if (seen.Add(tag)) + { + touched++; + } + } + else if (seen.Add(tag)) + { + toSubscribe.Add(tag); + } + } + + justReadCount = touched; + return [.. toSubscribe]; + } + + // Drops least-recently-read tags off the back of the advise set until the + // incoming tags fit under MaxSubscribedTags, unadvising them on the worker in + // one batch. A failed unadvise must not fail the read: the tags are dropped + // from tracking regardless, and the session-invalidation path already handles + // gateway/worker drift. Callers must hold _gate. + private async Task EvictForAsync( + GatewaySession session, + int serverHandle, + int incomingCount, + int justReadCount, + CancellationToken cancellationToken) + { + int overflow = _subscribed.Count + incomingCount - MaxSubscribedTags; + int evictable = _subscribed.Count - justReadCount; + int evictCount = Math.Min(overflow, evictable); + if (evictCount <= 0) + { + return; + } + + List evictedHandles = new(evictCount); + for (int i = 0; i < evictCount && _recency.Last is { } oldest; i++) + { + _recency.RemoveLast(); + _subscribed.Remove(oldest.Value.TagAddress); + if (oldest.Value.ItemHandle != 0) + { + evictedHandles.Add(oldest.Value.ItemHandle); + } + } + + _logger.LogDebug( + "Dashboard advise set hit its cap of {Cap}; evicted {EvictedCount} least-recently-read tags.", + MaxSubscribedTags, + evictCount); + + if (evictedHandles.Count == 0) + { + return; + } + + try + { + await session.UnsubscribeBulkAsync(serverHandle, evictedHandles, cancellationToken) + .ConfigureAwait(false); + } + catch (Exception exception) when (exception is not OperationCanceledException) + { + _logger.LogDebug( + exception, + "Unadvising {EvictedCount} evicted dashboard tags failed; they stay dropped from tracking.", + evictedHandles.Count); + } + } + + // Records the freshly advised tags as the most recently read, keeping each + // tag's item handle so eviction can unadvise it. Tags the worker failed to + // advise are still tracked (matching the pre-cap behaviour of not retrying + // them on every read) but carry no handle, so eviction just forgets them. + // Callers must hold _gate. + private void TrackSubscribed(IReadOnlyList tagAddresses, IReadOnlyList results) + { + Dictionary handles = new(results.Count, StringComparer.OrdinalIgnoreCase); + foreach (SubscribeResult result in results) + { + if (result.WasSuccessful && !string.IsNullOrEmpty(result.TagAddress)) + { + handles[result.TagAddress] = result.ItemHandle; + } + } + + // Inserted back-to-front so the read's first tag ends up most recent. + for (int i = tagAddresses.Count - 1; i >= 0; i--) + { + string tag = tagAddresses[i]; + handles.TryGetValue(tag, out int itemHandle); + _subscribed[tag] = _recency.AddFirst(new SubscribedTag(tag, itemHandle)); + } + } + + // Forgets the whole advise set without unadvising: every call site is one where + // the backing session (and with it every item handle) is already gone. + // Callers must hold _gate. + private void ClearSubscriptions() + { + _subscribed.Clear(); + _recency.Clear(); + } + // Returns a Ready session + its Register server handle, opening a fresh // session when none exists or the current one is no longer usable. Callers // must hold _gate. @@ -132,7 +270,7 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync await CloseQuietlyAsync(existing.SessionId).ConfigureAwait(false); } - _subscribed.Clear(); + ClearSubscriptions(); _session = null; GatewaySession session = await _sessionManager.OpenSessionAsync( @@ -178,7 +316,7 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync { _session = null; _serverHandle = 0; - _subscribed.Clear(); + ClearSubscriptions(); } private async Task CloseQuietlyAsync(string sessionId) @@ -212,4 +350,8 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync _gate.Dispose(); } + + // One entry of the advise set. ItemHandle is the handle the worker bound for + // the tag, or 0 when the subscribe failed and there is nothing to unadvise. + private readonly record struct SubscribedTag(string TagAddress, int ItemHandle); } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs new file mode 100644 index 0000000..dcb12ca --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs @@ -0,0 +1,344 @@ +using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Server.Dashboard; +using ZB.MOM.WW.MxGateway.Server.Sessions; +using ZB.MOM.WW.MxGateway.Server.Workers; +using ZB.MOM.WW.MxGateway.Tests.TestSupport; + +namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard; + +public sealed class DashboardLiveDataServiceTests +{ + // Mirrors DashboardLiveDataService.MaxSubscribedTags — the cap is private, so the + // tests drive it through the public read surface at exactly its documented size. + private const int MaxSubscribedTags = 256; + + /// + /// Verifies a tag already in the advise set is not subscribed again on a later read. + /// + [Fact] + public async Task ReadAsync_WhenTagAlreadySubscribed_DoesNotResubscribe() + { + RecordingWorkerClient worker = new(); + await using FakeSessionManager sessionManager = new(worker); + await using DashboardLiveDataService service = CreateService(sessionManager); + + await service.ReadAsync(["Tank_001.PV", "Tank_002.PV"], CancellationToken.None); + DashboardLiveReadResult second = await service.ReadAsync( + ["Tank_001.PV", "Tank_002.PV"], + CancellationToken.None); + + Assert.Null(second.Error); + Assert.Equal(["Tank_001.PV", "Tank_002.PV"], worker.SubscribedTags); + Assert.Empty(worker.UnsubscribedHandles); + } + + /// + /// Verifies the advise set is capped: subscribing past the cap unadvises the + /// least-recently-read tag on the worker and leaves the re-read tag advised. + /// + [Fact] + public async Task ReadAsync_PastCap_EvictsLeastRecentlyReadTag() + { + RecordingWorkerClient worker = new(); + await using FakeSessionManager sessionManager = new(worker); + await using DashboardLiveDataService service = CreateService(sessionManager); + + string[] filler = CreateTagAddresses(MaxSubscribedTags); + await service.ReadAsync(filler, CancellationToken.None); + + // Re-read the tag that is currently least recent, making it most recent: the + // tag read before it is now the eviction candidate. + await service.ReadAsync([filler[^1]], CancellationToken.None); + Assert.Equal(MaxSubscribedTags, worker.SubscribedTags.Count); + + DashboardLiveReadResult overflow = await service.ReadAsync( + ["Overflow.PV"], + CancellationToken.None); + + Assert.Null(overflow.Error); + Assert.Equal([worker.HandleFor(filler[^2])], worker.UnsubscribedHandles); + Assert.Equal("Overflow.PV", worker.SubscribedTags[^1]); + Assert.Equal(MaxSubscribedTags + 1, worker.SubscribedTags.Count); + + // The evicted tag is no longer tracked and re-subscribes; the re-read one does not. + await service.ReadAsync([filler[^1], filler[^2]], CancellationToken.None); + Assert.Equal(filler[^2], worker.SubscribedTags[^1]); + Assert.Equal(MaxSubscribedTags + 2, worker.SubscribedTags.Count); + } + + /// + /// Verifies a failed unadvise of an evicted tag does not fail the read, and the + /// evicted tag is dropped from tracking anyway. + /// + [Fact] + public async Task ReadAsync_WhenEvictionUnsubscribeFails_StillCompletesRead() + { + RecordingWorkerClient worker = new() { FailUnsubscribe = true }; + await using FakeSessionManager sessionManager = new(worker); + await using DashboardLiveDataService service = CreateService(sessionManager); + + string[] filler = CreateTagAddresses(MaxSubscribedTags); + await service.ReadAsync(filler, CancellationToken.None); + + DashboardLiveReadResult overflow = await service.ReadAsync( + ["Overflow.PV"], + CancellationToken.None); + + Assert.Null(overflow.Error); + Assert.Equal("Overflow.PV", Assert.Single(overflow.Values).TagAddress); + Assert.Equal(1, sessionManager.OpenCount); + + // The evicted tag was dropped from tracking despite the failed unadvise. + await service.ReadAsync([filler[^1]], CancellationToken.None); + Assert.Equal(filler[^1], worker.SubscribedTags[^1]); + } + + private static DashboardLiveDataService CreateService(ISessionManager sessionManager) + { + return new DashboardLiveDataService( + sessionManager, + new FakeGatewayAlarmService(), + NullLogger.Instance); + } + + private static string[] CreateTagAddresses(int count) + { + string[] addresses = new string[count]; + for (int i = 0; i < count; i++) + { + addresses[i] = $"Tank_{i:D4}.PV"; + } + + return addresses; + } + + // Serves the dashboard service a single Ready session backed by the recording + // worker, so reads exercise the real GatewaySession bulk command path. + private sealed class FakeSessionManager(RecordingWorkerClient workerClient) : ISessionManager, IAsyncDisposable + { + private readonly List _sessions = []; + + /// Gets the number of sessions the dashboard service opened. + public int OpenCount { get; private set; } + + /// + public Task OpenSessionAsync( + SessionOpenRequest request, + string? clientIdentity, + string? ownerKeyId, + CancellationToken cancellationToken) + { + OpenCount++; + string sessionId = $"dashboard-session-{OpenCount}"; + GatewaySession session = new( + sessionId, + "Galaxy", + $"mxgw-1-{sessionId}", + "nonce", + clientIdentity, + request.ClientSessionName, + request.ClientCorrelationId, + TimeSpan.FromSeconds(30), + TimeSpan.FromSeconds(5), + TimeSpan.FromSeconds(5), + DateTimeOffset.UnixEpoch); + session.AttachWorkerClient(workerClient); + session.MarkReady(); + _sessions.Add(session); + return Task.FromResult(session); + } + + /// + public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) + { + session = _sessions.Find(candidate => candidate.SessionId == sessionId); + return session is not null; + } + + /// + public Task CloseSessionAsync(string sessionId, CancellationToken cancellationToken) + { + return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false)); + } + + /// + public Task InvokeAsync( + string sessionId, + WorkerCommand command, + CancellationToken cancellationToken) => + throw new NotSupportedException(); + + /// + public IAsyncEnumerable ReadEventsAsync(string sessionId, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + /// + public Task KillWorkerAsync( + string sessionId, + string reason, + CancellationToken cancellationToken) => + throw new NotSupportedException(); + + /// + public Task CloseExpiredLeasesAsync(DateTimeOffset now, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + /// + public Task ShutdownAsync(CancellationToken cancellationToken) => throw new NotSupportedException(); + + /// Disposes every session handed to the dashboard service. + /// A task that represents the asynchronous operation. + public async ValueTask DisposeAsync() + { + foreach (GatewaySession session in _sessions) + { + await session.DisposeAsync().ConfigureAwait(false); + } + } + } + + // Answers Register / SubscribeBulk / UnsubscribeBulk / ReadBulk with successful + // replies and records what the dashboard advised and unadvised. + private sealed class RecordingWorkerClient : IWorkerClient + { + private const int RegisteredServerHandle = 77; + + private readonly Dictionary _itemHandles = new(StringComparer.OrdinalIgnoreCase); + private int _nextItemHandle = 1000; + + /// + public string SessionId => "dashboard-session-1"; + + /// + public int? ProcessId => 4242; + + /// + public WorkerClientState State => WorkerClientState.Ready; + + /// + public DateTimeOffset LastHeartbeatAt => DateTimeOffset.UnixEpoch; + + /// Gets the tag addresses subscribed, in the order the dashboard asked for them. + public List SubscribedTags { get; } = []; + + /// Gets the item handles the dashboard unsubscribed, in order. + public List UnsubscribedHandles { get; } = []; + + /// Gets or sets a value indicating whether unsubscribe commands throw. + public bool FailUnsubscribe { get; set; } + + /// Gets the item handle bound for a previously subscribed tag. + /// Tag address to look up. + /// The bound item handle. + public int HandleFor(string tagAddress) => _itemHandles[tagAddress]; + + /// + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task InvokeAsync( + WorkerCommand command, + TimeSpan timeout, + CancellationToken cancellationToken) + { + MxCommand mxCommand = command.Command + ?? throw new InvalidOperationException("The dashboard sent a command with no payload."); + + MxCommandReply reply = new() + { + Kind = mxCommand.Kind, + ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok }, + }; + + switch (mxCommand.Kind) + { + case MxCommandKind.Register: + reply.Register = new RegisterReply { ServerHandle = RegisteredServerHandle }; + break; + case MxCommandKind.SubscribeBulk: + reply.SubscribeBulk = Subscribe(mxCommand.SubscribeBulk.TagAddresses); + break; + case MxCommandKind.UnsubscribeBulk: + if (FailUnsubscribe) + { + throw new InvalidOperationException("Simulated worker unsubscribe failure."); + } + + UnsubscribedHandles.AddRange(mxCommand.UnsubscribeBulk.ItemHandles); + reply.UnsubscribeBulk = new BulkSubscribeReply(); + break; + case MxCommandKind.ReadBulk: + reply.ReadBulk = Read(mxCommand.ReadBulk.TagAddresses); + break; + default: + throw new NotSupportedException($"Unexpected dashboard command {mxCommand.Kind}."); + } + + return Task.FromResult(new WorkerCommandReply { Reply = reply }); + } + + /// + public async IAsyncEnumerable ReadEventsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.CompletedTask.ConfigureAwait(false); + yield break; + } + + /// + public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public void Kill(string reason) + { + } + + /// + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + + private BulkSubscribeReply Subscribe(IEnumerable tagAddresses) + { + BulkSubscribeReply subscribeReply = new(); + foreach (string tagAddress in tagAddresses) + { + SubscribedTags.Add(tagAddress); + if (!_itemHandles.TryGetValue(tagAddress, out int itemHandle)) + { + itemHandle = _nextItemHandle++; + _itemHandles[tagAddress] = itemHandle; + } + + subscribeReply.Results.Add(new SubscribeResult + { + ServerHandle = RegisteredServerHandle, + TagAddress = tagAddress, + ItemHandle = itemHandle, + WasSuccessful = true, + }); + } + + return subscribeReply; + } + + private BulkReadReply Read(IEnumerable tagAddresses) + { + BulkReadReply readReply = new(); + foreach (string tagAddress in tagAddresses) + { + readReply.Results.Add(new BulkReadResult + { + ServerHandle = RegisteredServerHandle, + TagAddress = tagAddress, + ItemHandle = _itemHandles.TryGetValue(tagAddress, out int itemHandle) ? itemHandle : 0, + WasSuccessful = true, + Quality = 192, + }); + } + + return readReply; + } + } +}