/* ========================================================================
* 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.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Opc.Ua.Client.ModelChange;
using Opc.Ua.Client.Subscriptions.Streaming;
namespace Opc.Ua.Client
{
///
/// Streaming subscription + model change tracking integration for
/// . Implements lazy access to a shared
/// streaming subscription and an optional model change tracker that
/// invalidates the session's when the server
/// reports address-space changes.
///
public partial class ManagedSession
{
private IStreamingSubscription? m_defaultStreaming;
private IModelChangeTracker? m_modelChangeTracker;
private readonly object m_streamingLock = new();
private bool m_modelChangeTrackingEnabled;
///
/// A shared lazy over this
/// session's . Backs both
/// (when enabled) and ad-hoc
/// -based subscriptions.
///
///
/// Created on first access and disposed with the session. The
/// underlying OPC UA subscription is itself created lazily, on the
/// first call to a SubscribeXxxAsync method.
///
public IStreamingSubscription DefaultStreaming
{
get
{
if (m_defaultStreaming != null)
{
return m_defaultStreaming;
}
lock (m_streamingLock)
{
if (m_defaultStreaming != null)
{
return m_defaultStreaming;
}
m_defaultStreaming = new StreamingSubscription(SubscriptionManager);
return m_defaultStreaming;
}
}
}
///
/// Optional model change tracker. Created on first call to
/// . Subscribes to
/// the server's GeneralModelChangeEventType notifier and
/// invalidates the session's on change.
/// Returns null if tracking has never been enabled.
///
public IModelChangeTracker? ModelChange => m_modelChangeTracker;
///
/// Enables address-space model change tracking on this session.
/// Creates the tracker on first call and starts pumping changes.
/// Idempotent — subsequent calls are no-ops once enabled.
///
/// Cancellation token.
public async ValueTask EnableModelChangeTrackingAsync(CancellationToken ct = default)
{
if (m_modelChangeTrackingEnabled)
{
return;
}
IModelChangeTracker tracker;
lock (m_streamingLock)
{
if (m_modelChangeTrackingEnabled)
{
return;
}
m_modelChangeTracker ??= new ModelChangeTracker(
DefaultStreaming,
NodeCache,
m_logger);
m_modelChangeTrackingEnabled = true;
tracker = m_modelChangeTracker;
}
await tracker.StartTrackingAsync(ct).ConfigureAwait(false);
}
///
/// Disables address-space model change tracking. The tracker is
/// stopped but the streaming subscription is preserved.
///
/// Cancellation token.
public async ValueTask DisableModelChangeTrackingAsync(CancellationToken ct = default)
{
IModelChangeTracker? tracker;
lock (m_streamingLock)
{
if (!m_modelChangeTrackingEnabled)
{
return;
}
m_modelChangeTrackingEnabled = false;
tracker = m_modelChangeTracker;
}
if (tracker != null)
{
await tracker.StopTrackingAsync(ct).ConfigureAwait(false);
}
}
///
/// Disposes the streaming subscription and model change tracker
/// owned by this session. Called from
/// ManagedSession.DisposeAsync.
///
private async ValueTask DisposeStreamingAsync()
{
IModelChangeTracker? tracker;
IStreamingSubscription? streaming;
lock (m_streamingLock)
{
tracker = m_modelChangeTracker;
streaming = m_defaultStreaming;
m_modelChangeTracker = null;
m_defaultStreaming = null;
m_modelChangeTrackingEnabled = false;
}
if (tracker != null)
{
try
{
await tracker.DisposeAsync().ConfigureAwait(false);
}
catch (Exception ex)
{
m_logger.LogDebug(ex,
"ManagedSession: ModelChangeTracker dispose failed.");
}
}
if (streaming != null)
{
try
{
await streaming.DisposeAsync().ConfigureAwait(false);
}
catch (Exception ex)
{
m_logger.LogDebug(ex,
"ManagedSession: DefaultStreaming dispose failed.");
}
}
}
}
}