/* ========================================================================
* 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.Linq;
using NUnit.Framework;
using Opc.Ua.Tests;
using Quickstarts.Servers;
namespace Opc.Ua.Server.Tests
{
///
/// Round-trip tests for the BinaryEncoder/BinaryDecoder based
/// serialization in and
/// .
/// Tests call the internal encode/decode methods directly.
/// Persistor cleanup clears the process-wide batch directory.
///
[TestFixture]
[Category("Server")]
[SetCulture("en-us")]
[SetUICulture("en-us")]
[NonParallelizable]
public class DurableSubscriptionSerializationTests
{
private ServiceMessageContext m_context;
///
/// Creates the serialization message context and registers the two fixture namespaces.
///
[OneTimeSetUp]
public void OneTimeSetUp()
{
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
m_context = ServiceMessageContext.Create(telemetry);
m_context.NamespaceUris.GetIndexOrAppend(
"urn:test:namespace1");
m_context.NamespaceUris.GetIndexOrAppend(
"urn:test:namespace2");
}
///
/// Verifies that an empty durable subscription preserves its settings and empty item and message collections.
///
[Test]
public void RoundTripEmptySubscription()
{
StoredSubscription result =
RoundTripSubscription(CreateMinimalSubscription(id: 42));
Assert.That(result.Id, Is.EqualTo(42u));
Assert.That(result.IsDurable, Is.True);
Assert.That(result.PublishingInterval, Is.EqualTo(1000.0));
Assert.That(result.Priority, Is.EqualTo((byte)5));
Assert.That(result.MonitoredItems.Count(), Is.Zero);
Assert.That(result.SentMessages, Has.Count.EqualTo(0));
}
///
/// Verifies that the current sample format preserves publishing state and anonymous owner application URI.
///
[TestCase(false)]
[TestCase(true)]
public void RoundTripSubscriptionState(bool publishingEnabled)
{
StoredSubscription original = CreateMinimalSubscription(id: 43);
original.PublishingEnabled = publishingEnabled;
original.OwnerClientApplicationUri = "urn:test:client";
StoredSubscription result = RoundTripSubscription(original);
Assert.Multiple(() =>
{
Assert.That(result.PublishingEnabled, Is.EqualTo(publishingEnabled));
Assert.That(result.OwnerClientApplicationUri, Is.EqualTo("urn:test:client"));
});
}
///
/// Verifies that the current sample format preserves the triggering links between
/// monitored items (OPC 10000-4 ยง5.13.1.6).
///
[Test]
public void RoundTripSubscriptionTriggeringLinks()
{
StoredSubscription original = CreateMinimalSubscription(id: 45);
original.TriggeringLinks = new Dictionary>
{
[1] = [2u, 3u],
[4] = [5u]
};
StoredSubscription result = RoundTripSubscription(original);
Assert.That(result.TriggeringLinks, Is.Not.Null);
Assert.Multiple(() =>
{
Assert.That(result.TriggeringLinks.Keys, Is.EquivalentTo(new uint[] { 1, 4 }));
Assert.That(result.TriggeringLinks[1], Is.EqualTo(new uint[] { 2, 3 }));
Assert.That(result.TriggeringLinks[4], Is.EqualTo(new uint[] { 5 }));
});
}
///
/// Verifies that version-one sample records migrate to the safe default publishing state.
///
[Test]
public void DecodeVersionOneSubscriptionDefaultsState()
{
using var encoder = new BinaryEncoder(m_context);
StoredSubscription original = CreateMinimalSubscription(id: 44);
WriteLegacySubscription(encoder, original);
using var decoder = new BinaryDecoder(encoder.CloseAndReturnBuffer()!, m_context);
StoredSubscription result = SubscriptionStore.DecodeSubscription(decoder, version: 1);
Assert.Multiple(() =>
{
Assert.That(result.PublishingEnabled, Is.True);
Assert.That(result.OwnerClientApplicationUri, Is.Null);
});
}
///
/// Verifies that durable subscription serialization preserves its monitored items and their identifiers.
///
[Test]
public void RoundTripSubscriptionWithMonitoredItems()
{
StoredSubscription original = CreateMinimalSubscription(id: 100);
original.MonitoredItems =
[
CreateMonitoredItem(id: 1, subscriptionId: 100),
CreateMonitoredItem(id: 2, subscriptionId: 100),
CreateMonitoredItem(id: 3, subscriptionId: 100)
];
StoredSubscription result = RoundTripSubscription(original);
var items = result.MonitoredItems.ToList();
Assert.That(items, Has.Count.EqualTo(3));
Assert.That(items[0].Id, Is.EqualTo(1u));
Assert.That(items[1].Id, Is.EqualTo(2u));
Assert.That(items[2].Id, Is.EqualTo(3u));
Assert.That(items[0].SubscriptionId, Is.EqualTo(100u));
}
///
/// Verifies that durable subscription serialization preserves monitored-item properties.
///
[Test]
public void RoundTripMonitoredItemProperties()
{
StoredSubscription original = CreateMinimalSubscription(id: 1);
StoredMonitoredItem mi = CreateMonitoredItem(id: 7, subscriptionId: 1);
mi.NodeId = new NodeId("TestNode", 2);
mi.AttributeId = Attributes.Value;
mi.QueueSize = 10;
mi.SamplingInterval = 500.0;
mi.DiscardOldest = true;
mi.MonitoringMode = MonitoringMode.Reporting;
mi.IsDurable = true;
mi.LastValue = new DataValue(
new Variant(42), StatusCodes.Good, DateTime.UtcNow);
original.MonitoredItems =
[mi];
var restored = RoundTripSubscription(original)
.MonitoredItems.First() as StoredMonitoredItem;
Assert.That(restored, Is.Not.Null);
Assert.That(restored.Id, Is.EqualTo(7u));
Assert.That(restored.NodeId,
Is.EqualTo(new NodeId("TestNode", 2)));
Assert.That(restored.AttributeId,
Is.EqualTo(Attributes.Value));
Assert.That(restored.QueueSize, Is.EqualTo(10u));
Assert.That(restored.SamplingInterval, Is.EqualTo(500.0));
Assert.That(restored.DiscardOldest, Is.True);
Assert.That(restored.MonitoringMode,
Is.EqualTo(MonitoringMode.Reporting));
Assert.That(restored.IsDurable, Is.True);
Assert.That(restored.LastValue.IsNull, Is.False);
Assert.That((int)restored.LastValue.WrappedValue,
Is.EqualTo(42));
Assert.That(restored.LastValue, Is.EqualTo(mi.LastValue));
}
///
/// Verifies that the retired interim notification-state format is rejected before decoding a subscription.
///
[Test]
public void DecodeRejectsInterimVersionTwoSubscriptions()
{
StoredMonitoredItem first = CreateMonitoredItem(id: 7, subscriptionId: 1);
first.LastValue = new DataValue(Variant.From(42), StatusCodes.Good);
StoredMonitoredItem second = CreateMonitoredItem(id: 8, subscriptionId: 2);
second.LastValue = new DataValue(Variant.From(43), StatusCodes.Good);
using var encoder = new BinaryEncoder(m_context);
foreach (StoredMonitoredItem item in new[] { first, second })
{
StoredSubscription subscription = CreateMinimalSubscription(item.SubscriptionId);
subscription.MonitoredItems = [item];
SubscriptionStore.EncodeSubscription(encoder, subscription);
encoder.WriteBoolean(null, true);
encoder.WriteDataValue(null, new DataValue(Variant.Null, StatusCodes.BadCommunicationError));
encoder.WriteStatusCode(null, StatusCodes.BadCommunicationError);
}
byte[] bytes = encoder.CloseAndReturnBuffer()!;
using var decoder = new BinaryDecoder(bytes!, m_context);
Assert.That(
() => SubscriptionStore.DecodeSubscription(decoder, version: 2),
Throws.TypeOf()
.With.Message.Contains("Unsupported durable subscription store version 2"));
Assert.That(decoder.Position, Is.Zero);
}
///
/// Verifies that new subscription records contain raw monitored-item state rather than transient notifications.
///
[Test]
public void NewSubscriptionRecordContainsOnlyRawMonitoredItemState()
{
StoredSubscription subscription = CreateMinimalSubscription(id: 1);
StoredMonitoredItem item = CreateMonitoredItem(id: 7, subscriptionId: 1);
item.LastValue = new DataValue(Variant.From(42), StatusCodes.Good);
subscription.MonitoredItems = [item];
using var encoder = new BinaryEncoder(m_context);
SubscriptionStore.EncodeSubscription(encoder, subscription);
byte[] bytes = encoder.CloseAndReturnBuffer()!;
using var decoder = new BinaryDecoder(bytes!, m_context);
IStoredMonitoredItem restored = SubscriptionStore.DecodeSubscription(decoder, version: 4)
.MonitoredItems.Single();
Assert.That(restored.Id, Is.EqualTo(item.Id));
Assert.That(restored.LastValue, Is.EqualTo(item.LastValue));
Assert.That(decoder.Position, Is.EqualTo(bytes.Length));
}
///
/// Verifies that durable subscription serialization preserves the user identity token.
///
[Test]
public void RoundTripSubscriptionWithUserIdentityToken()
{
StoredSubscription original = CreateMinimalSubscription(id: 55);
original.UserIdentityToken = new UserNameIdentityToken
{
PolicyId = "username_policy",
UserName = "testuser"
};
StoredSubscription result = RoundTripSubscription(original);
Assert.That(result.UserIdentityToken, Is.Not.Null);
Assert.That(result.UserIdentityToken,
Is.InstanceOf());
var token = (UserNameIdentityToken)result.UserIdentityToken;
Assert.That(token.UserName, Is.EqualTo("testuser"));
Assert.That(token.PolicyId, Is.EqualTo("username_policy"));
}
///
/// Verifies that multiple durable subscriptions round-trip independently.
///
[Test]
public void RoundTripMultipleSubscriptions()
{
var subs = new List
{
CreateMinimalSubscription(id: 1),
CreateMinimalSubscription(id: 2),
CreateMinimalSubscription(id: 3)
};
subs[0].MonitoredItems =
[
CreateMonitoredItem(id: 10, subscriptionId: 1),
CreateMonitoredItem(id: 11, subscriptionId: 1)
];
subs[1].MonitoredItems =
[
CreateMonitoredItem(id: 20, subscriptionId: 2)
];
List results = RoundTripSubscriptions(subs);
Assert.That(results, Has.Count.EqualTo(3));
Assert.That(results[0].Id, Is.EqualTo(1u));
Assert.That(results[1].Id, Is.EqualTo(2u));
Assert.That(results[2].Id, Is.EqualTo(3u));
Assert.That(results[0].MonitoredItems.Count(),
Is.EqualTo(2));
Assert.That(results[1].MonitoredItems.Count(),
Is.EqualTo(1));
Assert.That(results[2].MonitoredItems.Count(),
Is.Zero);
}
///
/// Verifies that a storable data-change queue round-trips through serialization.
///
[Test]
public void RoundTripStorableDataChangeQueue()
{
StorableDataChangeQueue result = RoundTripDataChangeQueue(
new StorableDataChangeQueue
{
IsDurable = true,
MonitoredItemId = 42,
ItemsInQueue = 5,
QueueSize = 100
});
Assert.That(result.IsDurable, Is.True);
Assert.That(result.MonitoredItemId, Is.EqualTo(42u));
Assert.That(result.ItemsInQueue, Is.EqualTo(5));
Assert.That(result.QueueSize, Is.EqualTo(100u));
Assert.That(result.EnqueueBatch, Is.Null);
Assert.That(result.DequeueBatch, Is.Null);
}
///
/// Verifies that a storable data-change queue preserves its batches during serialization.
///
[Test]
public void RoundTripStorableDataChangeQueueWithBatches()
{
var values = new List<(DataValue, ServiceResult?)>
{
(new DataValue(new Variant(1), StatusCodes.Good), null),
(new DataValue(new Variant("hello"), StatusCodes.Good),
new ServiceResult(StatusCodes.BadTimeout))
};
var batch = new DataChangeBatch(values, 10, 42);
StorableDataChangeQueue result = RoundTripDataChangeQueue(
new StorableDataChangeQueue
{
IsDurable = true,
MonitoredItemId = 42,
ItemsInQueue = 2,
QueueSize = 100,
EnqueueBatch = batch,
DataChangeBatches =
[batch]
});
Assert.That(result.EnqueueBatch, Is.Not.Null);
Assert.That(result.EnqueueBatch.Values,
Has.Count.EqualTo(2));
Assert.That(
(int)result.EnqueueBatch.Values[0].Item1.WrappedValue,
Is.EqualTo(1));
Assert.That(result.EnqueueBatch.Values[1].Item2,
Is.Not.Null);
Assert.That(result.DataChangeBatches,
Has.Count.EqualTo(1));
}
///
/// Verifies that a storable event queue round-trips through serialization.
///
[Test]
public void RoundTripStorableEventQueue()
{
StorableEventQueue result = RoundTripEventQueue(
new StorableEventQueue
{
IsDurable = true,
MonitoredItemId = 99,
QueueSize = 50
});
Assert.That(result.IsDurable, Is.True);
Assert.That(result.MonitoredItemId, Is.EqualTo(99u));
Assert.That(result.QueueSize, Is.EqualTo(50u));
Assert.That(result.EnqueueBatch, Is.Null);
Assert.That(result.DequeueBatch, Is.Null);
}
private static StoredSubscription CreateMinimalSubscription(
uint id)
{
return new StoredSubscription
{
Id = id,
IsDurable = true,
LifetimeCounter = 10,
MaxLifetimeCount = 100,
MaxKeepaliveCount = 10,
MaxMessageCount = 5,
MaxNotificationsPerPublish = 1000,
PublishingInterval = 1000.0,
Priority = 5,
LastSentMessage = 0,
SequenceNumber = 1,
SentMessages = [],
MonitoredItems = []
};
}
private static StoredMonitoredItem CreateMonitoredItem(
uint id, uint subscriptionId)
{
return new StoredMonitoredItem
{
Id = id,
SubscriptionId = subscriptionId,
NodeId = new NodeId(id, 0),
AttributeId = Attributes.Value,
QueueSize = 1,
SamplingInterval = 1000.0,
MonitoringMode = MonitoringMode.Reporting,
TimestampsToReturn = TimestampsToReturn.Both,
DiagnosticsMasks = DiagnosticsMasks.None
};
}
private static void WriteLegacySubscription(
BinaryEncoder encoder,
StoredSubscription subscription)
{
encoder.WriteUInt32(null, subscription.Id);
encoder.WriteBoolean(null, subscription.IsDurable);
encoder.WriteUInt32(null, subscription.LifetimeCounter);
encoder.WriteUInt32(null, subscription.MaxLifetimeCount);
encoder.WriteUInt32(null, subscription.MaxKeepaliveCount);
encoder.WriteUInt32(null, subscription.MaxMessageCount);
encoder.WriteUInt32(null, subscription.MaxNotificationsPerPublish);
encoder.WriteDouble(null, subscription.PublishingInterval);
encoder.WriteByte(null, subscription.Priority);
encoder.WriteInt32(null, subscription.LastSentMessage);
encoder.WriteUInt32(null, subscription.SequenceNumber);
encoder.WriteExtensionObject(null, ExtensionObject.Null);
encoder.WriteExtensionObjectArray(null, []);
encoder.WriteInt32(null, 0);
}
private StoredSubscription RoundTripSubscription(
StoredSubscription original)
{
return RoundTripSubscriptions([original]).Single();
}
///
/// Encodes and decodes a list of subscriptions using
/// the internal SubscriptionStore methods.
///
private List RoundTripSubscriptions(
List originals)
{
using var stream = new MemoryStream();
using (var encoder = new BinaryEncoder(
stream, m_context, true))
{
encoder.WriteStringArray(
null, m_context.NamespaceUris.ToArrayOf());
encoder.WriteStringArray(
null, m_context.ServerUris.ToArrayOf());
encoder.WriteInt32(null, originals.Count);
foreach (StoredSubscription sub in originals)
{
SubscriptionStore.EncodeSubscription(
encoder, sub);
}
}
stream.Position = 0;
using var decoder = new BinaryDecoder(
stream, m_context, true);
ArrayOf nsUris = decoder.ReadStringArray(null);
ArrayOf srvUris = decoder.ReadStringArray(null);
decoder.SetMappingTables(
new NamespaceTable(nsUris.Memory.ToArray().Select(uri => uri!)),
new StringTable(srvUris.Memory.ToArray().Select(uri => uri!)));
int count = decoder.ReadInt32(null);
var results = new List(count);
for (int i = 0; i < count; i++)
{
results.Add(
SubscriptionStore.DecodeSubscription(decoder));
}
return results;
}
///
/// Encodes and decodes a StorableDataChangeQueue using
/// the internal DurableMonitoredItemQueueFactory methods.
///
private StorableDataChangeQueue RoundTripDataChangeQueue(
StorableDataChangeQueue original)
{
using var stream = new MemoryStream();
using (var encoder = new BinaryEncoder(
stream, m_context, true))
{
encoder.WriteStringArray(
null, m_context.NamespaceUris.ToArrayOf());
encoder.WriteStringArray(
null, m_context.ServerUris.ToArrayOf());
DurableMonitoredItemQueueFactory
.EncodeDataChangeQueue(encoder, original);
}
stream.Position = 0;
using var decoder = new BinaryDecoder(
stream, m_context, true);
ArrayOf nsUris = decoder.ReadStringArray(null);
ArrayOf srvUris = decoder.ReadStringArray(null);
decoder.SetMappingTables(
new NamespaceTable(nsUris.Memory.ToArray().Select(uri => uri!)),
new StringTable(srvUris.Memory.ToArray().Select(uri => uri!)));
return DurableMonitoredItemQueueFactory
.DecodeDataChangeQueue(decoder);
}
///
/// Encodes and decodes a StorableEventQueue using
/// the internal DurableMonitoredItemQueueFactory methods.
///
private StorableEventQueue RoundTripEventQueue(
StorableEventQueue original)
{
using var stream = new MemoryStream();
using (var encoder = new BinaryEncoder(
stream, m_context, true))
{
encoder.WriteStringArray(
null, m_context.NamespaceUris.ToArrayOf());
encoder.WriteStringArray(
null, m_context.ServerUris.ToArrayOf());
DurableMonitoredItemQueueFactory
.EncodeEventQueue(encoder, original);
}
stream.Position = 0;
using var decoder = new BinaryDecoder(
stream, m_context, true);
ArrayOf nsUris = decoder.ReadStringArray(null);
ArrayOf srvUris = decoder.ReadStringArray(null);
decoder.SetMappingTables(
new NamespaceTable(nsUris.Memory.ToArray().Select(uri => uri!)),
new StringTable(srvUris.Memory.ToArray().Select(uri => uri!)));
return DurableMonitoredItemQueueFactory
.DecodeEventQueue(decoder);
}
///
/// Verifies that a data-change batch round-trips through the queue persistor.
///
[Test]
public void RoundTripDataChangeBatchViaPersistor()
{
using IDisposable scope =
AmbientMessageContext.SetScopedContext(m_context);
var values = new List<(DataValue, ServiceResult?)>
{
(new DataValue(new Variant(42), StatusCodes.Good), null),
(new DataValue(new Variant("test"), StatusCodes.Good),
new ServiceResult(StatusCodes.BadTimeout))
};
var batch = new DataChangeBatch(values, 10, 99);
var persistor = new BatchPersistor(
NUnitTelemetryContext.Create());
// Persist
persistor.PersistSynchronously(batch);
Assert.That(batch.IsPersisted, Is.True);
Assert.That(batch.Values, Is.Null);
// Restore
persistor.RestoreSynchronously(batch);
Assert.That(batch.IsPersisted, Is.False);
Assert.That(batch.Values, Is.Not.Null);
Assert.That(batch.Values, Has.Count.EqualTo(2));
Assert.That(
(int)batch.Values[0].Item1.WrappedValue,
Is.EqualTo(42));
Assert.That(batch.Values[1].Item2, Is.Not.Null);
Assert.That(
batch.Values[1].Item2.StatusCode,
Is.EqualTo(StatusCodes.BadTimeout));
// Clean up
persistor.DeleteBatches([]);
}
///
/// Verifies that an event batch round-trips through the queue persistor.
///
[Test]
public void RoundTripEventBatchViaPersistor()
{
using IDisposable scope =
AmbientMessageContext.SetScopedContext(m_context);
var events = new List
{
new() { ClientHandle = 1 },
new() { ClientHandle = 2 }
};
var batch = new EventBatch(events, 5, 77);
var persistor = new BatchPersistor(
NUnitTelemetryContext.Create());
// Persist
persistor.PersistSynchronously(batch);
Assert.That(batch.IsPersisted, Is.True);
// Restore
persistor.RestoreSynchronously(batch);
Assert.That(batch.IsPersisted, Is.False);
Assert.That(batch.Events, Is.Not.Null);
Assert.That(batch.Events, Has.Count.EqualTo(2));
// Clean up
persistor.DeleteBatches([]);
}
///
/// Verifies that deleting a persisted queue batch removes its backing file.
///
[Test]
public void PersistorDeleteBatchRemovesFile()
{
using IDisposable scope =
AmbientMessageContext.SetScopedContext(m_context);
var values = new List<(DataValue, ServiceResult?)>
{
(new DataValue(new Variant(1), StatusCodes.Good), null)
};
var batch = new DataChangeBatch(values, 5, 50);
var persistor = new BatchPersistor(
NUnitTelemetryContext.Create());
persistor.PersistSynchronously(batch);
Assert.That(batch.IsPersisted, Is.True);
// Delete all batches
persistor.DeleteBatches([]);
// Restore should not crash (file gone)
batch.Restore(values);
batch.SetPersisted();
persistor.RestoreSynchronously(batch);
// RestoreInProgress is reset even when file is missing
Assert.That(batch.RestoreInProgress, Is.False);
}
}
}