/* ========================================================================
* 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.IO;
using System.Threading;
using System.Threading.Tasks;
using Opc.Ua.Client.Subscriptions;
namespace Opc.Ua.Client
{
///
/// Convenience extensions on for working
/// with the new options-based subscription API. These wrap plain
/// /
/// snapshots into
///
/// adapters so callers don't need to set up a DI options pipeline for
/// one-off use.
///
public static class ManagedSessionExtensions
{
///
/// Add a new subscription to the session using the supplied options
/// snapshot. The subscription is registered with
/// and starts
/// asynchronously.
///
/// is null.
public static ISubscription AddSubscription(
this ManagedSession session,
ISubscriptionNotificationHandler handler,
Subscriptions.SubscriptionOptions options)
{
if (session == null)
{
throw new ArgumentNullException(nameof(session));
}
if (handler == null)
{
throw new ArgumentNullException(nameof(handler));
}
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
var monitor = new OptionsMonitor(options);
return session.SubscriptionManager.Add(handler, monitor);
}
///
/// Add a new subscription to the session, configuring options via
/// a callback over a fresh
/// record.
///
/// is null.
public static ISubscription AddSubscription(
this ManagedSession session,
ISubscriptionNotificationHandler handler,
Func configure)
{
if (session == null)
{
throw new ArgumentNullException(nameof(session));
}
if (handler == null)
{
throw new ArgumentNullException(nameof(handler));
}
if (configure == null)
{
throw new ArgumentNullException(nameof(configure));
}
return session.AddSubscription(handler, configure(new Subscriptions.SubscriptionOptions()));
}
///
/// Add a monitored item to the subscription using the supplied
/// options snapshot.
///
/// is null.
///
public static bool TryAddMonitoredItem(
this ISubscription subscription,
string name,
Subscriptions.MonitoredItems.MonitoredItemOptions options,
out Subscriptions.MonitoredItems.IMonitoredItem? monitoredItem)
{
if (subscription == null)
{
throw new ArgumentNullException(nameof(subscription));
}
if (string.IsNullOrEmpty(name))
{
throw new ArgumentException("name is required", nameof(name));
}
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
var monitor = new OptionsMonitor(options);
return subscription.MonitoredItems.TryAdd(name, monitor, out monitoredItem);
}
///
/// Add a monitored item to the subscription, configuring options
/// via a callback over a fresh
///
/// record initialized with the supplied node id.
///
/// is null.
///
public static bool TryAddMonitoredItem(
this ISubscription subscription,
string name,
NodeId nodeId,
Func configure,
out Subscriptions.MonitoredItems.IMonitoredItem? monitoredItem)
{
if (subscription == null)
{
throw new ArgumentNullException(nameof(subscription));
}
if (string.IsNullOrEmpty(name))
{
throw new ArgumentException("name is required", nameof(name));
}
if (configure == null)
{
throw new ArgumentNullException(nameof(configure));
}
return subscription.TryAddMonitoredItem(
name,
configure(new Subscriptions.MonitoredItems.MonitoredItemOptions { StartNodeId = nodeId }),
out monitoredItem);
}
///
/// Persist every subscription managed by
/// (or a caller-supplied subset) to
/// in OPC UA binary encoding. The format starts with the
/// session's namespace and server URI tables so the snapshot is
/// portable across sessions whose tables index URIs in different
/// positions.
///
/// Session whose
/// is being
/// snapshotted.
/// Writable destination stream.
/// Optional subset of subscriptions
/// to include. When null every subscription currently
/// managed by is included.
/// Cancellation token.
public static ValueTask SaveSubscriptionsAsync(
this ManagedSession session,
Stream destination,
IEnumerable? subscriptions = null,
CancellationToken ct = default)
{
if (session == null)
{
throw new ArgumentNullException(nameof(session));
}
return session.SubscriptionManager.SaveAsync(
destination, session.MessageContext, subscriptions, ct);
}
///
/// Restore subscriptions previously persisted by
/// . Each restored subscription is
/// re-registered with
/// .
///
/// Session that owns the V2 subscription
/// manager and supplies the active message context.
/// Readable source stream produced by
/// .
/// Factory invoked once per
/// restored subscription to construct the application's
/// . The factory
/// receives the per-subscription stable name captured in the
/// snapshot.
/// When true the
/// restored subscriptions take over the original server-side
/// state via TransferSubscriptions; if that fails for any
/// subscription the V2 manager falls back to recreate.
/// Cancellation token.
public static ValueTask> LoadSubscriptionsAsync(
this ManagedSession session, Stream source,
Func handlerFactory,
bool transferSubscriptions = false, CancellationToken ct = default)
{
if (session == null)
{
throw new ArgumentNullException(nameof(session));
}
return session.SubscriptionManager.LoadAsync(source,
session.MessageContext, handlerFactory,
transferSubscriptions, ct);
}
///
/// Capture an in-memory snapshot of every subscription managed
/// by . The returned list of
/// s can be persisted by
/// the caller in any format and later passed to
/// .
///
public static IReadOnlyList SnapshotSubscriptions(
this ManagedSession session)
{
if (session == null)
{
throw new ArgumentNullException(nameof(session));
}
var result = new List();
foreach (ISubscription s in session.SubscriptionManager.Items)
{
if (s is Subscriptions.Subscription concrete)
{
result.Add(concrete.Snapshot());
}
}
return result;
}
///
/// Restore a list of s
/// previously captured by .
///
/// Session that owns the V2 subscription
/// manager.
/// Snapshots to restore.
/// Factory invoked once per
/// snapshot to construct the application's notification
/// handler. The factory receives the snapshot itself so callers
/// can route by options or per-item metadata.
/// When true the
/// restored subscriptions take over the original server-side
/// state via TransferSubscriptions; if that fails for
/// any subscription the V2 manager falls back to recreate.
/// Cancellation token.
public static async ValueTask> RestoreSubscriptionsAsync(
this ManagedSession session,
IReadOnlyList states,
Func handlerFactory,
bool transferSubscriptions = false,
CancellationToken ct = default)
{
if (session == null)
{
throw new ArgumentNullException(nameof(session));
}
if (states == null)
{
throw new ArgumentNullException(nameof(states));
}
if (handlerFactory == null)
{
throw new ArgumentNullException(nameof(handlerFactory));
}
var result = new List(states.Count);
SubscriptionManager manager =
(SubscriptionManager)session.SubscriptionManager;
foreach (SubscriptionStateSnapshot state in states)
{
result.Add(await manager.RestoreAsync(
handlerFactory(state),
state,
transferSubscriptions,
ct).ConfigureAwait(false));
}
return result;
}
}
}