/* ========================================================================
* 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.Globalization;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
namespace Opc.Ua.Client.Subscriptions
{
///
/// Save / Load support for the V2 .
///
///
///
/// The on-wire format is the OPC UA format.
/// Each subscription is captured as a
/// — an
/// emitted by the
/// source generator — and written
/// directly via .
/// The reader instantiates
/// statically by type (no lookup is
/// required because the wire schema is implicit in the call site).
/// Future schema evolution is handled by adding new optional fields
/// to the snapshot record — they are encoded only when set and
/// recognized via by the
/// reader.
///
///
/// The stream still preserves the source session's namespace and
/// server URI tables so the snapshot is portable across sessions
/// whose tables index URIs at different positions; the loader
/// remaps those indices into the target session's tables before
/// decoding each snapshot's NodeId fields.
///
///
internal static class SubscriptionManagerSerializer
{
#pragma warning disable RCS1229 // Use async/await when necessary - this path is synchronous; ValueTask wraps work for future async I/O
public static ValueTask SaveAsync(
SubscriptionManager manager,
Stream stream,
IServiceMessageContext messageContext,
IEnumerable? subscriptions,
CancellationToken ct = default)
#pragma warning restore RCS1229
{
ct.ThrowIfCancellationRequested();
if (manager == null)
{
throw new ArgumentNullException(nameof(manager));
}
if (stream == null)
{
throw new ArgumentNullException(nameof(stream));
}
if (messageContext == null)
{
throw new ArgumentNullException(nameof(messageContext));
}
// Capture a snapshot per selected subscription. Default =
// all subscriptions managed by this instance. Snapshot is
// taken from the concrete subscription type.
IEnumerable selected =
subscriptions ?? manager.Items;
var snapshots = new List();
foreach (ISubscription s in selected)
{
if (s is Subscription concrete)
{
snapshots.Add(concrete.Snapshot());
}
}
using var encoder = new BinaryEncoder(stream, messageContext, true);
encoder.WriteStringArray(null, messageContext.NamespaceUris.ToArrayOf());
encoder.WriteStringArray(null, messageContext.ServerUris.ToArrayOf());
encoder.WriteInt32(null, snapshots.Count);
foreach (SubscriptionStateSnapshot snapshot in snapshots)
{
// Write the snapshot directly as an IEncodeable.
// Schema identity is statically encoded by the call site
// (we always read back a SubscriptionStateSnapshot); schema
// evolution is handled by adding new optional fields with
// CanOmitFields-aware encoding.
encoder.WriteEncodeable(null, snapshot);
}
return default;
}
public static async ValueTask> LoadAsync(
SubscriptionManager manager,
Stream stream,
IServiceMessageContext messageContext,
Func handlerFactory,
bool transferSubscriptions,
CancellationToken ct)
{
if (manager == null)
{
throw new ArgumentNullException(nameof(manager));
}
if (stream == null)
{
throw new ArgumentNullException(nameof(stream));
}
if (messageContext == null)
{
throw new ArgumentNullException(nameof(messageContext));
}
if (handlerFactory == null)
{
throw new ArgumentNullException(nameof(handlerFactory));
}
using var decoder = new BinaryDecoder(stream, messageContext, true);
ArrayOf nsUris = decoder.ReadStringArray(null);
ArrayOf serverUris = decoder.ReadStringArray(null);
decoder.SetMappingTables(
nsUris.IsNull
? new NamespaceTable()
: new NamespaceTable(nsUris.Memory.ToArray()!),
serverUris.IsNull
? new StringTable()
: new StringTable(serverUris.Memory.ToArray()!));
int count = decoder.ReadInt32(null);
if (count <= 0)
{
return [];
}
var restored = new List(count);
for (int i = 0; i < count; i++)
{
ct.ThrowIfCancellationRequested();
// Schema identity is implicit in the call: we always
// read a SubscriptionStateSnapshot. Future schema evolution
// happens by adding optional fields recognized via
// HasField on the decoder side.
SubscriptionStateSnapshot state =
decoder.ReadEncodeable(null);
string syntheticName = i.ToString(CultureInfo.InvariantCulture);
ISubscriptionNotificationHandler handler =
handlerFactory(syntheticName);
ISubscription subscription = await manager.RestoreAsync(
handler,
state,
transferSubscriptions,
ct).ConfigureAwait(false);
restored.Add(subscription);
}
return restored;
}
}
}