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