/* ======================================================================== * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Collections.Generic; using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; namespace Opc.Ua.Client.Subscriptions.MonitoredItems { /// /// A managed set of monitored items inside a subscription. /// internal sealed class MonitoredItemManager : IMonitoredItemCollection, IMonitoredItemContext, IAsyncDisposable { /// public IEnumerable Items { get { lock (m_monitoredItemsLock) { return [.. m_monitoredItems.Values]; } } } /// public uint Count { get { lock (m_monitoredItemsLock) { return (uint)m_monitoredItems.Count; } } } /// /// Create monitored item manager /// /// /// public MonitoredItemManager(IMonitoredItemManagerContext context, ITelemetryContext telemetry) { m_logger = telemetry.LoggerFactory.CreateLogger(); m_context = context; } /// public async ValueTask DisposeAsync() { try { foreach (MonitoredItem? monitoredItem in m_monitoredItems.Values.ToList()) { await monitoredItem.DisposeAsync().ConfigureAwait(false); } } finally { m_monitoredItems.Clear(); m_monitoredItemsByName.Clear(); } } /// public bool TryAdd(string name, IOptionsMonitor options, out IMonitoredItem? monitoredItem) { lock (m_monitoredItemsLock) { if (m_monitoredItemsByName.TryGetValue(name, out MonitoredItem? item)) { monitoredItem = item; return false; } item = m_context.CreateMonitoredItem(name, options, this); m_monitoredItems.Add(item.ClientHandle, item); m_monitoredItemsByName.Add(name, item); monitoredItem = item; } m_context.Update(); return true; } /// /// Construct + register a monitored item that was loaded from a /// snapshot (see ). The /// freshly-constructed item is bound to the saved /// / /// via and any pending /// create-request queued during construction is abandoned, so /// the V2 state machine does not issue a redundant /// CreateMonitoredItems. The owning subscription drives /// the take-over via the standard transfer flow /// (). /// internal bool AddLoaded(MonitoredItemLoadState state) { lock (m_monitoredItemsLock) { if (m_monitoredItemsByName.ContainsKey(state.Name)) { return false; } MonitoredItem item = m_context.CreateMonitoredItem( state.Name, state.Options, this); item.ApplyLoadState(state); m_monitoredItems.Add(item.ClientHandle, item); m_monitoredItemsByName.Add(state.Name, item); return true; } // Intentionally NOT calling m_context.Update() — Update() // signals the subscription's state-control auto-reset event // which would prematurely wake StateManagerAsync and have it // try to create what we just loaded. } /// public bool TryGetMonitoredItemByClientHandle(uint clientHandle, [MaybeNullWhen(false)] out IMonitoredItem? monitoredItem) { lock (m_monitoredItemsLock) { bool result = m_monitoredItems.TryGetValue(clientHandle, out MonitoredItem? item); monitoredItem = item; return result; } } /// public bool TryGetMonitoredItemByName(string name, [MaybeNullWhen(false)] out IMonitoredItem? monitoredItem) { lock (m_monitoredItemsLock) { bool result = m_monitoredItemsByName.TryGetValue(name, out MonitoredItem? item); monitoredItem = item; return result; } } /// public bool TryRemove(uint clientHandle) { lock (m_monitoredItemsLock) { if (!m_monitoredItems.Remove(clientHandle, out MonitoredItem? monitoredItem)) { return false; } m_monitoredItemsByName.Remove(monitoredItem.Name); m_deletedItems.Add(monitoredItem); } m_context.Update(); return true; } /// public IReadOnlyList Update(IReadOnlyList<( string Name, IOptionsMonitor Options)> state) { var monitoredItems = new List(state.Count); lock (m_monitoredItemsLock) { // Capture current state IDictionary remove = m_monitoredItems.ToDictionary(); // Generate items foreach ((string? name, IOptionsMonitor? options) in state) { if (!m_monitoredItemsByName.TryGetValue(name, out MonitoredItem? item)) { // Create new item item = m_context.CreateMonitoredItem(name, options, this); m_monitoredItems.Add(item.ClientHandle, item); m_monitoredItemsByName.Add(name, item); } else { item.Options = options; remove.Remove(item.ClientHandle); } monitoredItems.Add(item); } // Remove all items not in state foreach (uint clientHandle in remove.Keys) { if (m_monitoredItems.Remove(clientHandle, out MonitoredItem? monitoredItem)) { m_monitoredItemsByName.Remove(monitoredItem.Name); m_deletedItems.Add(monitoredItem); } } } m_context.Update(); return monitoredItems; } /// public void NotifyItemChange(MonitoredItem monitoredItem, bool itemDisposed) { Debug.Assert(monitoredItem != null); if (!itemDisposed) { m_context.Update(); } } /// public bool NotifyItemChangeResult( MonitoredItem monitoredItem, int retryCount, MonitoredItemOptions source, ServiceResult serviceResult, bool final, MonitoringFilterResult? filterResult) { return final || retryCount > 5; // TODO: Resiliency policy } /// public async ValueTask ConditionRefreshAsync( uint monitoredItemServerId, CancellationToken ct = default) { ArrayOf methodsToCall = [ new CallMethodRequest { ObjectId = ObjectTypeIds.ConditionType, MethodId = MethodIds.ConditionType_ConditionRefresh2, InputArguments = [ new Variant(m_context.Id), new Variant(monitoredItemServerId) ] } ]; CallResponse response = await m_context.MethodServiceSet .CallAsync(null, methodsToCall, ct).ConfigureAwait(false); ArrayOf results = response.Results; ArrayOf diagnosticInfos = response.DiagnosticInfos; ClientBase.ValidateResponse(results, methodsToCall); ClientBase.ValidateDiagnosticInfos(diagnosticInfos, methodsToCall); if (StatusCode.IsBad(results[0].StatusCode)) { throw new ServiceResultException(ClientBase.GetResult( results[0].StatusCode, 0, diagnosticInfos, response.ResponseHeader)); } } /// /// Create notifications for monitored items /// /// /// internal ReadOnlyMemory CreateNotification( DataChangeNotification notification) { var memory = new DataValueChange[notification.MonitoredItems.Count]; lock (m_monitoredItemsLock) { for (int i = 0; i < notification.MonitoredItems.Count; i++) { MonitoredItemNotification item = notification.MonitoredItems[i]; m_monitoredItems.TryGetValue(item.ClientHandle, out MonitoredItem? monitored); memory[i] = new DataValueChange(monitored, item.Value, item.DiagnosticInfo); } } // TODO: Sort on order of monitored items // memory.AsSpan().Sort((x, y) => x.MonitoredItem!.ClientHandle.CompareTo( // y.MonitoredItem!.ClientHandle)); return memory; } /// /// Create notifications for monitored items /// /// /// internal ReadOnlyMemory CreateNotification( EventNotificationList notification) { var memory = new EventNotification[notification.Events.Count]; lock (m_monitoredItemsLock) { for (int i = 0; i < notification.Events.Count; i++) { EventFieldList item = notification.Events[i]; m_monitoredItems.TryGetValue(item.ClientHandle, out MonitoredItem? monitored); memory[i] = new EventNotification(monitored, item.EventFields); } } // memory.AsSpan().Sort((x, y) => x.MonitoredItem!.ClientHandle.CompareTo( // y.MonitoredItem!.ClientHandle)); return memory; } /// /// Apply changes to monitored items /// /// /// /// /// internal async Task ApplyChangesAsync(bool once, bool resetAll, CancellationToken ct) { bool modified = false; while (!ct.IsCancellationRequested && TryGetMonitoredItemChanges( out List? itemsToDelete, out List? itemsToModify, resetAll)) { await ApplyMonitoredItemChangesAsync(itemsToDelete, itemsToModify, ct).ConfigureAwait(false); // While there are changes pending to be applied apply them modified = true; if (once) { break; } } return modified; } /// /// Notify monitored items of the state changes in the subscription /// /// /// internal void OnSubscriptionStateChange(SubscriptionState state, TimeSpan currentPublishingInterval) { lock (m_monitoredItemsLock) { foreach (MonitoredItem item in m_monitoredItems.Values) { item.OnSubscriptionStateChange(state, currentPublishingInterval); } } } /// /// Notify monitored items that the subscription manager is paused /// /// internal void NotifySubscriptionManagerPaused(bool paused) { lock (m_monitoredItemsLock) { foreach (MonitoredItem item in m_monitoredItems.Values) { item.NotifySubscriptionManagerPaused(paused); } } } /// /// Apply monitored item changes /// /// /// /// /// internal async ValueTask ApplyMonitoredItemChangesAsync(List itemsToDelete, List itemsToModify, CancellationToken ct) { if (itemsToDelete.Count != 0) { foreach (MonitoredItem? dispose in itemsToDelete.Where(c => !c.Created)) { await dispose.DisposeAsync().ConfigureAwait(false); } itemsToDelete.RemoveAll(c => !c.Created); } if (itemsToDelete.Count != 0) { try { var monitoredItemIds = new ArrayOf(itemsToDelete .Select(m => m.ServerId).ToArray()); DeleteMonitoredItemsResponse response = await m_context.MonitoredItemServiceSet.DeleteMonitoredItemsAsync(null, m_context.Id, monitoredItemIds, ct).ConfigureAwait(false); ClientBase.ValidateResponse(response.Results, monitoredItemIds); ClientBase.ValidateDiagnosticInfos(response.DiagnosticInfos, monitoredItemIds); for (int index = 0; index < response.Results.Count; index++) { if (StatusCode.IsGood(response.Results[index]) || response.Results[index] == StatusCodes.BadMonitoredItemIdInvalid || !itemsToDelete[index].Created) { await itemsToDelete[index].DisposeAsync().ConfigureAwait(false); continue; } m_deletedItems.Add(itemsToDelete[index]); // Retry this // TODO: Give up after a while } } catch (Exception ex) { m_deletedItems.AddRange(itemsToDelete); m_logger.LogInformation(ex, "Failed to delete monitored items."); } } // To force recreate if item is created therefore, we need to delete the item first. var deletes = itemsToModify .Where(c => c.Item.Created && c.ForceRecreate) .ToList(); if (deletes.Count > 0) { var monitoredItemIds = new ArrayOf(deletes.Select(c => c.Item.ServerId).ToArray()); DeleteMonitoredItemsResponse response = await m_context.MonitoredItemServiceSet.DeleteMonitoredItemsAsync(null, m_context.Id, monitoredItemIds, ct).ConfigureAwait(false); ClientBase.ValidateResponse(response.Results, monitoredItemIds); ClientBase.ValidateDiagnosticInfos(response.DiagnosticInfos, monitoredItemIds); for (int index = 0; index < response.Results.Count; index++) { deletes[index].SetDeleteResult( response.Results[index], index, response.DiagnosticInfos, response.ResponseHeader); } } // Modify all items that need to be modified. var modifications = itemsToModify .Where(c => c.Item.Created) .ToList(); foreach (IGrouping group in modifications .Where(c => c.Modify != null) .GroupBy(c => c.Timestamps)) { var monitoredItems = group.ToList(); var requests = new ArrayOf(group.Select(c => c.Modify!).ToArray()); if (requests.Count > 0) { ModifyMonitoredItemsResponse response = await m_context.MonitoredItemServiceSet.ModifyMonitoredItemsAsync(null, m_context.Id, group.Key, requests, ct).ConfigureAwait(false); ClientBase.ValidateResponse(response.Results, requests); ClientBase.ValidateDiagnosticInfos(response.DiagnosticInfos, requests); // update results. for (int index = 0; index < response.Results.Count; index++) { monitoredItems[index].SetModifyResult(requests[index], response.Results[index], index, response.DiagnosticInfos, response.ResponseHeader); } } } // Finalize by updating the monitoring mode where needed foreach (IGrouping mode in modifications .Where(c => c.MonitoringModeChange != null) .GroupBy(c => c.MonitoringModeChange!.Value)) { var monitoredItems = mode .Where(c => c.Item.Created && c.Item.CurrentMonitoringMode != mode.Key) .ToList(); var requests = new ArrayOf(monitoredItems.Select(c => c.Item.ServerId).ToArray()); if (requests.Count > 0) { SetMonitoringModeResponse response = await m_context.MonitoredItemServiceSet.SetMonitoringModeAsync(null, m_context.Id, mode.Key, requests, ct).ConfigureAwait(false); ClientBase.ValidateResponse(response.Results, requests); ClientBase.ValidateDiagnosticInfos(response.DiagnosticInfos, requests); // update results. for (int index = 0; index < response.Results.Count; index++) { monitoredItems[index].SetMonitoringModeResult( mode.Key, response.Results[index], index, response.DiagnosticInfos, response.ResponseHeader); } } } // Create all items var creations = itemsToModify .Where(c => !c.Item.Created) .ToList(); foreach (IGrouping group in creations .Where(c => c.Create != null) .GroupBy(c => c.Timestamps)) { var monitoredItems = group.ToList(); var requests = new ArrayOf(group.Select(c => c.Create!).ToArray()); if (requests.Count > 0) { CreateMonitoredItemsResponse response = await m_context.MonitoredItemServiceSet.CreateMonitoredItemsAsync(null, m_context.Id, group.Key, requests, ct).ConfigureAwait(false); ClientBase.ValidateResponse(response.Results, requests); ClientBase.ValidateDiagnosticInfos(response.DiagnosticInfos, requests); // update results. for (int index = 0; index < response.Results.Count; index++) { monitoredItems[index].SetCreateResult(requests[index], response.Results[index], index, response.DiagnosticInfos, response.ResponseHeader); } } } } /// /// Get monitored item changes /// /// /// /// /// internal bool TryGetMonitoredItemChanges(out List itemsToDelete, out List itemsToModify, bool resetAll = false) { // modify the subscription. itemsToModify = []; lock (m_monitoredItemsLock) { itemsToDelete = [.. m_deletedItems]; m_deletedItems.Clear(); foreach (MonitoredItem monitoredItem in m_monitoredItems.Values) { if (resetAll) { monitoredItem.Reset(); } // build item request. if (monitoredItem.TryGetPendingChange(out MonitoredItem.Change? change)) { itemsToModify.Add(change); } } } return itemsToDelete.Count != 0 || itemsToModify.Count != 0; } /// /// /// If a Client called CreateMonitoredItems during the network interruption /// and the call succeeded in the Server but did not return to the Client, /// then the Client does not know if the call succeeded. The Client may /// receive data changes for these monitored items but is not able to remove /// them since it does not know the Server handle. /// /// /// There is also no way for the Client to detect if the create succeeded. /// To delete and recreate the Subscription is also not an option since /// there may be several monitored items operating normally that should /// not be interrupted. /// /// /// /// internal async ValueTask TrySynchronizeHandlesAsync( CancellationToken ct) { (bool success, IReadOnlyList<(uint serverHandle, uint clientHandle)>? serverHandleStateMap) = await GetMonitoredItemsAsync( ct).ConfigureAwait(false); ArrayOf itemsToDelete; lock (m_monitoredItemsLock) { if (!success) { // Reset all items foreach (MonitoredItem monitoredItem in m_monitoredItems.Values) { monitoredItem.Reset(); } return false; } IDictionary monitoredItems = m_monitoredItems.ToDictionary(); m_monitoredItems.Clear(); // // Assign the server id to the item with the matching client handle. This // handles the case where the CreateMonitoredItems call succeeded on the // server side, but the response was not provided back. // var clientServerHandleMap = serverHandleStateMap .ToDictionary(m => m.clientHandle, m => m.serverHandle); foreach (KeyValuePair monitoredItem in monitoredItems.ToList()) { // // Adjust any items where the server handles does not map to the // handle the server has assigned. // uint clientHandle = monitoredItem.Value.ClientHandle; if (clientServerHandleMap.Remove(clientHandle, out uint serverHandle)) { monitoredItem.Value.SetTransferResult(clientHandle, serverHandle); monitoredItems.Remove(monitoredItem.Key); m_monitoredItems.Add(clientHandle, monitoredItem.Value); } } // // Assign the server side client handle to the item with the same server id // This handles the case where we are recreating the subscription from a // previously stored state. // var serverClientHandleMap = clientServerHandleMap .ToDictionary(m => m.Value, m => m.Key); foreach (KeyValuePair monitoredItem in monitoredItems.ToList()) { uint serverHandle = monitoredItem.Value.ServerId; if (serverClientHandleMap.Remove(serverHandle, out uint clientHandle)) { // // There should not be any more item with the same client handle // in this subscription, they were updated before. TODO: Assert // monitoredItem.Value.SetTransferResult(clientHandle, serverHandle); monitoredItems.Remove(monitoredItem.Key); m_monitoredItems.Add(clientHandle, monitoredItem.Value); } } m_deletedItems.Clear(); itemsToDelete = new ArrayOf(serverClientHandleMap.Keys.ToArray()); // Remaining items do not exist anymore on the server and need to be recreated foreach (MonitoredItem? missingOnServer in monitoredItems.Values) { if (missingOnServer.Created) { m_logger.LogDebug("{Subscription}: Recreate client item {Item}.", this, missingOnServer); missingOnServer.Reset(); } m_monitoredItems.Add(missingOnServer.ClientHandle, missingOnServer); } } if (itemsToDelete.Count == 0) { // Completed return true; } // Ensure all items on the server that are not in the subscription are deleted try { await m_context.MonitoredItemServiceSet.DeleteMonitoredItemsAsync(null, m_context.Id, itemsToDelete, ct).ConfigureAwait(false); return true; } catch (Exception ex) { m_logger.LogError(ex, "{Subscription}: Failed to delete missing items.", this); return false; } } private record struct MonitoredItemsHandles(bool Success, IReadOnlyList<(uint serverHandle, uint clientHandle)> Handles); /// /// Call the GetMonitoredItems method on the server. /// /// private async ValueTask GetMonitoredItemsAsync( CancellationToken ct) { try { var requests = new ArrayOf(new[] { new CallMethodRequest { ObjectId = ObjectIds.Server, MethodId = MethodIds.Server_GetMonitoredItems, InputArguments = [new Variant(m_context.Id)] } }); CallResponse response = await m_context.MethodServiceSet.CallAsync(null, requests, ct).ConfigureAwait(false); ArrayOf results = response.Results; ArrayOf diagnosticInfos = response.DiagnosticInfos; ClientBase.ValidateResponse(results, requests); ClientBase.ValidateDiagnosticInfos(diagnosticInfos, requests); if (StatusCode.IsBad(results[0].StatusCode)) { throw ServiceResultException.Create(results[0].StatusCode, 0, diagnosticInfos, response.ResponseHeader.StringTable); } ArrayOf outputArguments = results[0].OutputArguments; if (outputArguments.Count != 2 || !outputArguments[0].TryGetValue(out ArrayOf serverHandles) || !outputArguments[1].TryGetValue(out ArrayOf clientHandles) || serverHandles.IsNull || clientHandles.IsNull || serverHandles.Count != clientHandles.Count) { throw ServiceResultException.Create(StatusCodes.BadUnexpectedError, "Output arguments incorrect"); } return new MonitoredItemsHandles( true, serverHandles.ToList().Zip(clientHandles.ToList()).ToList()); } catch (ServiceResultException sre) { m_logger.LogError(sre, "{Subscription}: Failed to call GetMonitoredItems on server", this); return new MonitoredItemsHandles(false, []); } } private readonly IMonitoredItemManagerContext m_context; private readonly List m_deletedItems = []; private readonly Lock m_monitoredItemsLock = new(); private readonly Dictionary m_monitoredItems = []; private readonly Dictionary m_monitoredItemsByName = []; private readonly ILogger m_logger; } }