/* ========================================================================
* 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;
}
}