/* ======================================================================== * 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.Concurrent; using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Text; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; namespace Opc.Ua.Client.Subscriptions.MonitoredItems { /// /// A monitored item that can be extended to add extra /// information as context in the subscription. /// internal abstract class MonitoredItem : IMonitoredItem, IAsyncDisposable { /// public string Name { get; } /// public uint Order => m_currentOptions?.Order ?? 0u; /// public uint ServerId { get; private set; } /// public bool Created => ServerId != 0; /// public ServiceResult Error { get; private set; } /// public MonitoringFilterResult? FilterResult { get; private set; } /// public MonitoringMode CurrentMonitoringMode { get; internal set; } /// public TimeSpan CurrentSamplingInterval { get; private set; } /// public uint CurrentQueueSize { get; private set; } /// public uint ClientHandle { get; private set; } /// /// Client handle of the item that triggers this item, or /// 0 if no triggering relationship has been set. Used /// by the snapshot path and by the /// public projection which resolves the handle to the live /// sibling via . /// internal uint TriggeringItemClientHandle { get; set; } /// public IMonitoredItem? TriggeringItem { get { uint handle = TriggeringItemClientHandle; if (handle == 0) { return null; } return Context.TryGetMonitoredItemByClientHandle( handle, out IMonitoredItem? item) ? item : null; } } /// /// Apply server-side identifiers from a saved snapshot to this /// freshly-created monitored item. Used by /// when the caller /// requests transferSubscriptions so the loaded item can /// be matched to its server-side state via /// TransferSubscriptions. /// /// /// Re-assigning after construction /// matters because the V2 publish loop dispatches notifications /// by client handle. The snapshot's client handle is the one the /// server still uses; the post-construction one is freshly /// generated and would never match an incoming notification. /// internal void ApplyTransferState(uint clientHandle, uint serverId) { ClientHandle = clientHandle; ServerId = serverId; } /// /// Install fully-loaded state from a snapshot during V2 /// transfer-on-load ( /// with transferSubscriptions: true). Unlike /// , this also: /// /// Abandons the create-or-modify /// the ctor's /// just queued, so the V2 state machine does not issue a /// redundant CreateMonitoredItems request whose /// ClientHandle would mismatch the snapshot's /// (the source of the Debug.Assert failure that /// motivated this work). /// Re-installs the saved triggering links /// so a subsequent SetTriggering replay reflects the /// snapshot. /// /// The caller is responsible for binding the (possibly /// freshly-revised) server-side state via the /// TransferSubscriptions / GetMonitoredItems path /// that runs in . /// internal void ApplyLoadState(MonitoredItemLoadState state) { // Abandon any pending changes queued during the ctor. while (TryGetPendingChange(out Change? change)) { change.Abandon(); } // Snapshot the current options so the V2 state machine // treats the item as "configured to the loaded values" and // skips the create path until a real change arrives. m_currentOptions = m_options.CurrentValue; ClientHandle = state.ClientHandle; ServerId = state.ServerId; TriggeringItemClientHandle = state.TriggeringItemClientHandle; } /// /// The subscription that owns the monitored item. /// protected IMonitoredItemContext Context { get; } /// /// Current monitored item options /// internal IOptionsMonitor Options { get => m_options; set { if (m_options != value) { m_options = value; QueuePendingChanges(m_options.CurrentValue, m_currentOptions); m_changeTracking?.Dispose(); m_changeTracking = m_options.OnChange( (o, _) => OnOptionsChanged(o)); } } } /// /// Create monitored item /// /// /// /// /// protected MonitoredItem(IMonitoredItemContext context, string name, IOptionsMonitor options, ILogger logger) { Context = context; Name = name; Error = ServiceResult.Good; ClientHandle = Utils.IncrementIdentifier(ref GlobalClientHandleUint); m_logger = logger; m_options = Options = options; m_logger.LogDebug("{Item} CREATED.", this); } /// public ValueTask DisposeAsync() { GC.SuppressFinalize(this); return DisposeAsync(disposing: true); } /// /// Capture an immutable snapshot of this item's configuration /// + identifiers + triggering state. /// public MonitoredItemStateSnapshot Snapshot() { return MonitoredItemStateSnapshot.AsOptions( Name, m_options.CurrentValue, ClientHandle, ServerId, TriggeringItemClientHandle); } /// public ValueTask ConditionRefreshAsync(CancellationToken ct = default) { if (!Created) { throw ServiceResultException.Create( StatusCodes.BadMonitoredItemIdInvalid, "Monitored item has not been created on the server."); } return Context.ConditionRefreshAsync(ServerId, ct); } /// public override string? ToString() { StringBuilder sb = new StringBuilder() .Append(Context) .Append('#') .Append(ClientHandle) .Append('|') .Append(ServerId) .Append(" (") .Append(Name) .Append(')'); return sb.ToString(); } /// /// Dispose monitored item /// /// protected virtual ValueTask DisposeAsync(bool disposing) { if (disposing && !m_disposedValue) { while (TryGetPendingChange(out Change? change)) { change.Abandon(); } Context.NotifyItemChange(this, true); m_logger.LogDebug("{Item} REMOVED.", this); ServerId = 0; m_changeTracking?.Dispose(); m_disposedValue = true; } return default; } /// /// Called when the subscription state changed /// /// /// protected internal virtual void OnSubscriptionStateChange( SubscriptionState state, TimeSpan publishingInterval) { MonitoredItemOptions? options = m_currentOptions; if (options == null || (state != SubscriptionState.Created && state != SubscriptionState.Modified)) { return; } uint queueSize = options.QueueSize; if (!options.AutoSetQueueSize) { return; } if (publishingInterval == TimeSpan.Zero) { return; } TimeSpan samplingInterval = CurrentSamplingInterval; if (samplingInterval == TimeSpan.Zero) { samplingInterval = options.SamplingInterval; } if (samplingInterval <= TimeSpan.Zero) { return; } queueSize = Math.Max(queueSize, (uint)Math.Ceiling( publishingInterval.TotalMilliseconds / samplingInterval.TotalMilliseconds)) + 1; if (queueSize == options.QueueSize) { return; } OnOptionsChanged(options with { QueueSize = queueSize }); } /// /// Called when the options change /// /// protected virtual void OnOptionsChanged(MonitoredItemOptions options) { QueuePendingChanges(options, m_currentOptions); Context.NotifyItemChange(this); } /// /// Notify subscription that the subscription manager has paused or /// resumed operations. /// /// protected internal virtual void NotifySubscriptionManagerPaused(bool paused) { // empty } /// /// Get the current pending change in the change list. The change list /// collects the changes to be made to the monitored item while the /// subscription is applying state changes. /// /// /// internal bool TryGetPendingChange([NotNullWhen(true)] out Change? change) { return m_pendingChanges.TryPeek(out change); } /// /// Updates the object with the results of a transfer subscription request. /// /// /// /// internal void SetTransferResult(uint clientHandle, uint serverHandle) { if (m_disposedValue) { throw new ObjectDisposedException(GetType().FullName); } // ensure the global counter is not duplicating future handle ids if (clientHandle != ClientHandle) { m_logger.LogInformation("{Item}: UPDATE CLIENT ID from {Old} to {New}.", this, ClientHandle, clientHandle); ClientHandle = clientHandle; Utils.SetIdentifierToAtLeast(ref GlobalClientHandleUint, clientHandle); } if (serverHandle != ServerId) { m_logger.LogInformation("{Item}: UPDATE SERVER ID from {Old} to {New}.", this, ServerId, serverHandle); ServerId = serverHandle; } } /// /// Reset the monitored item to its initial state for recreation /// on server side. /// /// internal void Reset() { if (m_disposedValue) { throw new ObjectDisposedException(GetType().FullName); } m_logger.LogDebug("{Item}: RESET.", this); ServerId = 0; MonitoredItemOptions? options = m_currentOptions; while (TryGetPendingChange(out Change? change)) { change.Abandon(); options = change.Options; } m_currentOptions = null; if (options == null) { return; } QueuePendingChanges(options, null); } /// /// Queues changes into the change queue for the item /// /// /// private void QueuePendingChanges(MonitoredItemOptions options, MonitoredItemOptions? currentOptions) { if (currentOptions != null && currentOptions == options) { // No changes Context.NotifyItemChangeResult(this, 0, options, new ServiceResult( StatusCodes.BadNothingToDo), true, null); return; } if (options.StartNodeId.IsNull) { // Not valid Context.NotifyItemChangeResult(this, 0, options, new ServiceResult( StatusCodes.BadNodeIdInvalid), true, null); return; } m_currentOptions = options; m_pendingChanges.Enqueue(new Change(this, options, currentOptions)); } /// /// Complete a change /// /// /// internal bool CompleteChange(Change change) { return m_pendingChanges.TryDequeue(out Change? completed) && change == completed; } /// /// Change steps to apply to the monitored items in the /// subscription context. The change list is a queue of /// steps to perform inside the subscription. /// see Subscription.ApplyMonitoredItemChangesAsync for /// more information /// internal sealed class Change { /// /// The item on which the changes are applied /// public MonitoredItem Item { get; } /// /// Timestamps to return /// public TimestampsToReturn Timestamps => Options.TimestampsToReturn; /// /// Create request if the item is not yet created /// public MonitoredItemCreateRequest? Create { get; } /// /// Modification request if the item is already created /// public MonitoredItemModifyRequest? Modify { get; private set; } /// /// Monitoring mode change pending /// public MonitoringMode? MonitoringModeChange { get; private set; } /// /// Force recreating the item due to changes in the /// read item information. /// public bool ForceRecreate { get; private set; } /// /// Current retry count of this change /// public int RetryCount { get; private set; } /// /// Options that are the source of the change /// public MonitoredItemOptions Options { get; } /// /// Create change /// /// /// /// public Change(MonitoredItem item, MonitoredItemOptions options, MonitoredItemOptions? currentOptions) { Debug.Assert(!options.StartNodeId.IsNull); Options = options; Item = item; var parameters = new MonitoringParameters { ClientHandle = item.ClientHandle, SamplingInterval = (int)options.SamplingInterval.TotalMilliseconds, QueueSize = options.QueueSize, DiscardOldest = options.DiscardOldest, Filter = options.Filter != null ? new ExtensionObject(options.Filter) : default }; Create = new MonitoredItemCreateRequest { ItemToMonitor = new ReadValueId { NodeId = options.StartNodeId, AttributeId = options.AttributeId, IndexRange = options.IndexRange, DataEncoding = options.Encoding ?? QualifiedName.Null }, MonitoringMode = options.MonitoringMode, RequestedParameters = parameters }; if (currentOptions == null || currentOptions.StartNodeId != options.StartNodeId || currentOptions.AttributeId != options.AttributeId || currentOptions.IndexRange != options.IndexRange || currentOptions.Encoding != options.Encoding) { Modify = null; MonitoringModeChange = null; ForceRecreate = true; } else { Modify = new MonitoredItemModifyRequest { MonitoredItemId = item.ServerId, RequestedParameters = parameters }; if (currentOptions.MonitoringMode != options.MonitoringMode) { MonitoringModeChange = options.MonitoringMode; // If only monitoring mode changed, no need to modify if (currentOptions with { MonitoringMode = MonitoringModeChange.Value } == options) { Modify = null; } } else { MonitoringModeChange = null; } } } /// /// Updates the object with the results of a create monitored item request. /// /// /// /// /// /// internal void SetCreateResult(MonitoredItemCreateRequest request, MonitoredItemCreateResult result, int index, ArrayOf diagnosticInfos, ResponseHeader responseHeader) { Debug.Assert(request.RequestedParameters.ClientHandle == Item.ClientHandle); ServiceResult error = ServiceResult.Good; if (StatusCode.IsBad(result.StatusCode)) { error = ClientBase.GetResult(result.StatusCode, index, diagnosticInfos, responseHeader); } Item.CurrentMonitoringMode = request.MonitoringMode; Item.CurrentSamplingInterval = TimeSpan.FromMilliseconds( request.RequestedParameters.SamplingInterval); Item.CurrentQueueSize = request.RequestedParameters.QueueSize; if (ServiceResult.IsGood(error)) { Item.ServerId = result.MonitoredItemId; Item.CurrentSamplingInterval = TimeSpan.FromMilliseconds(result.RevisedSamplingInterval); Item.CurrentQueueSize = result.RevisedQueueSize; Item.LogRevisedSamplingRateAndQueueSize(Options, true); Notify(error, true, result.FilterResult); return; } // Declare final if not communication error which includes success RetryCount++; Notify(error, false, result.FilterResult); } /// /// Updates the object with the results of a modify monitored item request. /// /// /// /// /// /// internal void SetModifyResult(MonitoredItemModifyRequest request, MonitoredItemModifyResult result, int index, ArrayOf diagnosticInfos, ResponseHeader responseHeader) { Debug.Assert(request.RequestedParameters.ClientHandle == Item.ClientHandle); ServiceResult error = ServiceResult.Good; if (StatusCode.IsBad(result.StatusCode)) { error = ClientBase.GetResult(result.StatusCode, index, diagnosticInfos, responseHeader); } if (ServiceResult.IsGood(error)) { Item.CurrentSamplingInterval = TimeSpan.FromMilliseconds( request.RequestedParameters.SamplingInterval); Item.CurrentQueueSize = request.RequestedParameters.QueueSize; Item.CurrentSamplingInterval = TimeSpan.FromMilliseconds( result.RevisedSamplingInterval); Item.CurrentQueueSize = result.RevisedQueueSize; if (MonitoringModeChange == null) { Item.LogRevisedSamplingRateAndQueueSize(Options, false); } // Declare final Notify(error, MonitoringModeChange == null, result.FilterResult); return; } if (!IsCommunicationError(error)) { // Do not apply the mode change but force a recreate MonitoringModeChange = null; Modify = null; ForceRecreate = true; } if (MonitoringModeChange == null) { // Retry the modification request RetryCount++; } Notify(error, false, result.FilterResult); } /// /// Set monitoring mode result /// /// /// /// /// /// /// internal void SetMonitoringModeResult(MonitoringMode monitoringMode, StatusCode statusCode, int index, ArrayOf diagnosticInfos, ResponseHeader responseHeader) { ServiceResult error = ServiceResult.Good; if (StatusCode.IsBad(statusCode)) { error = ClientBase.GetResult(statusCode, index, diagnosticInfos, responseHeader); } if (ServiceResult.IsGood(error)) { Item.CurrentMonitoringMode = monitoringMode; Item.LogRevisedSamplingRateAndQueueSize(Options, false); Notify(error, true); return; } if (!IsCommunicationError(error)) { // Reapply the mode change Modify = null; ForceRecreate = true; } // Retry RetryCount++; Notify(error, false); } /// /// Set result of delete /// /// /// /// /// internal void SetDeleteResult(StatusCode statusCode, int index, ArrayOf diagnosticInfos, ResponseHeader responseHeader) { ServiceResult error = ServiceResult.Good; if (StatusCode.IsBad(statusCode)) { error = ClientBase.GetResult(statusCode, index, diagnosticInfos, responseHeader); } bool final = Create == null && Modify == null; if (ServiceResult.IsGood(error) || error.StatusCode == StatusCodes.BadMonitoredItemIdInvalid || final) { Item.ServerId = 0; ForceRecreate = false; // Now state is !Created so continue here to recreate } else { // Retry RetryCount++; } Notify(error, final); } /// /// Abandon the change /// internal void Abandon() { Notify(new ServiceResult(StatusCodes.BadOperationAbandoned), true); } /// /// Notify and handle retries /// /// /// /// private void Notify(ServiceResult error, bool final, ExtensionObject? filterResultExtensionObject = null) { MonitoringFilterResult? filterResult = null; if (filterResultExtensionObject.HasValue && filterResultExtensionObject.Value.TryGetValue(out MonitoringFilterResult? fr)) { filterResult = fr; } bool stop = Item.Context.NotifyItemChangeResult( Item, RetryCount, Options, error, final, filterResult); if (final || stop) { Item.CompleteChange(this); } Item.Error = error; Item.FilterResult = filterResult == null ? Item.FilterResult : (MonitoringFilterResult)filterResult.Clone(); } /// /// Returns true if communication issue and not an error /// in subscription or even success or uncertain states /// /// /// private static bool IsCommunicationError(ServiceResult error) { StatusCode code = error.StatusCode; return code == StatusCodes.BadCommunicationError || code == StatusCodes.BadNotConnected || code == StatusCodes.BadSecureChannelClosed; } } /// /// Log revised sampling rate and queue size /// /// /// public void LogRevisedSamplingRateAndQueueSize(MonitoredItemOptions options, bool created) { if (options.SamplingInterval != CurrentSamplingInterval && options.QueueSize != CurrentQueueSize && CurrentQueueSize != 0) { m_logger.LogInformation( "{Item}: {Action} SamplingInterval was " + "revised from {SamplingInterval} to {CurrentSamplingInterval} " + "and QueueSize from {QueueSize} to {CurrentQueueSize}.", this, created ? "CREATED" : "UPDATED", options.SamplingInterval, CurrentSamplingInterval, options.QueueSize, CurrentQueueSize); } else if (options.SamplingInterval != CurrentSamplingInterval) { m_logger.LogInformation( "{Item}: {Action} SamplingInterval was " + "revised from {SamplingInterval} to {CurrentSamplingInterval}.", this, created ? "CREATED" : "UPDATED", options.SamplingInterval, CurrentSamplingInterval); } else if (options.QueueSize != CurrentQueueSize && CurrentQueueSize != 0) { m_logger.LogInformation( "{Item}: {Action} QueueSize was " + "revised from {QueueSize} to {CurrentQueueSize}.", this, created ? "CREATED" : "UPDATED", options.QueueSize, CurrentQueueSize); } else { m_logger.LogDebug( "{Item}: {Action} with desired configuration.", this, created ? "CREATED" : "UPDATED"); } } private bool m_disposedValue; private MonitoredItemOptions? m_currentOptions; #pragma warning disable CA2213 // Disposed in DisposeAsync(bool) private IDisposable? m_changeTracking; #pragma warning restore CA2213 private readonly ConcurrentQueue m_pendingChanges = new(); private readonly ILogger m_logger; internal static uint GlobalClientHandleUint; private IOptionsMonitor m_options; } }