Files
UA-.NETStandard/tests/Opc.Ua.Server.Tests/DurableSubscriptionSerializationTests.cs
T
romanett efad0ddeff Server subscriptions: fix spec defects in Subscription/MonitoredItem handling found by audit (#4629)
# Description

Fixes defects found in a spec-grounded audit of the server
**Subscription / MonitoredItem** implementation
(`src/Opc.Ua.Server/Subscription/**`, `NodeManager/MonitoredItem/**`,
`EventManager`, monitored-item service dispatch and filter validation).
Each behaviour change was checked against OPC 10000-4 (§5.13, §5.14,
§7.21, §7.22), Part 5 (subscription diagnostics) and Part 9
(ConditionRefresh). Every fix has a regression test.

The audit found 56 defects (1 Critical, 3 High, 20 Medium, 32 Low). A
follow-up max-effort code review of the fix branch found 27 more issues,
mostly introduced by the first round of fixes. All but one are fixed
here.

### Most important fixes
- **Critical — remote hang:** an `AggregateFilter` with
`processingInterval = NaN` was not revised, and `AggregateCalculator`
looped forever while holding the node-manager lock, so one
CreateMonitoredItems request blocked the node manager. Non-finite, ≤0
and sub-tick intervals are now revised (Part 4 §5.13.2.1), and the
calculator rejects them as a backstop.
- **Monitored-item limits:** new
`ServerConfiguration.MaxMonitoredItemCount` /
`MaxMonitoredItemsPerSubscription`; Bad_TooManyMonitoredItems is
returned when a limit is hit, and ServerCapabilities now report the
limits. Both default to `0` (no limit), so existing behaviour is
unchanged.
- **Sampling groups:** a client could get one dedicated thread per
distinct sampling interval above 1 h; intervals above the last rate
group are now rounded, and the loop is async. An item could also end up
in two sampling groups and keep being sampled after it was deleted.
- **Subscription state machine** (Part 4 §5.14.1.2):
- the lifetime counter resets while a Publish is queued and counts
otherwise;
  - the lifetime and keep-alive stay within `MaxSubscriptionLifetime`;
  - LatePublishRequestCount is counted once per entry into LATE;
- messages held while publishing is disabled are no longer sent, and are
not offered by Transfer/Republish either;
- ConditionRefresh returns Bad_RefreshInProgress for any refresh already
queued or running on the subscription.
- **Triggering** (Part 4 §5.13.1.6):
- linked items are triggered by an item still draining a backlog, by the
first value after re-enable, and by an item created Disabled;
  - links are persisted for durable subscriptions;
- Disabled/Sampling items follow the publish rules in the shared
pipeline, so custom `IMonitoredItem` implementations follow them too.
- **Queues** (Part 4 §5.13.1.5, §7.21):
  - no duplicate value after a queue-size increase;
- EventQueueOverflow events are placed where the events were lost,
including on queue shrink;
  - the duplicate-event check scans the newest entries;
- EventQueueOverflowCount and MonitoringQueueOverflowCount are
maintained correctly;
- changing select clauses rebuilds queued events in the new layout, and
is safe on durable queues and on failure.
- **Filters:**
- per-clause EventFilterResult instead of rejecting the whole item (Part
4 §7.22.3);
- corrected status codes: Bad_DeadbandFilterInvalid,
Bad_MonitoredItemFilterInvalid, Bad_EventFilterInvalid,
Bad_MonitoringModeInvalid, Bad_TimestampsToReturnInvalid;
- aggregate filter in subscriptions: processing interval at least 2× the
revised sampling interval, revised StartTime on the requested grid,
ServerTimestamp = end of interval (Part 4 §7.22.4).
- **Lifecycle and races:**
- expired subscriptions are always cleaned up, even when the session
closes during cleanup;
  - Create/Transfer racing CloseSession returns Bad_SessionClosed;
  - session diagnostics counts no longer drift;
  - a large but valid Publish TimeoutHint no longer fails the request;
- a ready subscription is served by queued Publish requests instead of
going Late;
  - restored persisted state is validated.
- **Access:** a value change notification is checked against each
subscriber's UserAccessLevel, not the writer's.
- **Redundancy / durable store:**
  - the first-unsent sequence number is mirrored;
  - sequence numbers are no longer reused after a second failover;
- triggering links are stored (definition format 6, written only when
links exist, so older replicas can still read link-less subscriptions).

### Behaviour changes reviewers should know
- With the default config, publishing intervals above
`MaxSubscriptionLifetime / 3` (1,200,000 ms) are revised down, so
lifetime ≥ 3×keep-alive fits the configured maximum.
- The revised keep-alive is capped at a third of the max lifetime.
- Idle subscriptions expire after exactly MaxLifetimeCount cycles
without a queued Publish.
- A partly invalid EventFilter select list is accepted with per-clause
errors. The public `EventFilter.Validate` keeps Status Good if at least
one clause is valid; callers that cannot return per-clause results must
also check the new `EventFilter.Result.HasSelectClauseErrors`. There is
a migration note in `docs/migrate/2.0.x/sessions-subscriptions.md`.
- Aggregate processing intervals may be revised higher (≥ 2× sampling
interval); aggregate ServerTimestamp is now the interval end.
- Sampling intervals above 3,600,000 ms are rounded to multiples of
900,000 ms (default rate groups).

### Public API
Additive only:
- `ServerConfiguration.MaxMonitoredItemCount` and
`MaxMonitoredItemsPerSubscription`;
- `EventFilter.Result.HasSelectClauseErrors`;
- `IStoredSubscriptionTriggering`,
`ISubscriptionRetransmissionSendStateStore`,
`SubscriptionRetransmissionState.FirstUnsentSequenceNumber` (optional
store capabilities, detected with `as`);
- new `EventQueueHandler` constructor overloads.

### Known open items (not in this PR)
- **Part 4 §5.13.2.1:** CreateMonitoredItems should create an item even
when the user lacks Read permission and report the error in Publish.
This is not done: the create-time RolePermissions check is the only role
check on the initial value, so following the spec would leak values
until every node manager re-checks per sample.
- **Sampling loop:** it runs on thread-pool threads; node managers with
blocking synchronous reads should not use sampling groups heavily. The
alternative, a dedicated thread per group, was the thread-exhaustion
issue fixed here.
- **Restarts:** a pending QueueSize-1 value and a pending overflow flag
are not kept across a restart; that needs persistence interface
extensions.
- **Limit defaults:** decide whether the new limits should get finite
defaults (currently 0 = unlimited).

## Related Issues

- No tracking issue; the findings come from an internal audit of the
server subscription code.

## Checklist

- [ ] I have signed the
[CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf)
and read the
[CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md)
doc.
- [x] I have added tests that prove my fix is effective or that my
feature works and increased code coverage.
- [x] I have added all necessary documentation.
- [ ] I have verified that my changes do not introduce (new) build or
analyzer warnings.
- [ ] I ran **all** tests locally using the **UA.slnx** solution against
at least .net **framework** and .net **10**, and all passed.
- [ ] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings.
- [ ] I have addressed **all** PR feedback received.

### Local verification (net10.0, all green)
| Suite | Passed | Failed |
|---|---|---|
| Opc.Ua.Server.Tests | 7963 | 0 |
| Opc.Ua.Subscriptions.Tests | 643 | 0 |
| Opc.Ua.Subscriptions.Classic.Tests | 57 | 0 |
| Opc.Ua.Subscriptions.Durable.Tests | 36 | 0 |
| Opc.Ua.Redundancy.Server.Tests | 847 | 0 |
| Opc.Ua.Core.Tests (filter tests) | 283 | 0 |

net48 build of `Opc.Ua.Server.Tests`: 0 errors. net48 tests and the full
UA.slnx were not run locally.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Marc Schier <marcschier@users.noreply.github.com>
2026-10-03 05:43:16 +02:00

684 lines
27 KiB
C#

/* ========================================================================
* 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
{
/// <summary>
/// Round-trip tests for the BinaryEncoder/BinaryDecoder based
/// serialization in <see cref="SubscriptionStore"/> and
/// <see cref="DurableMonitoredItemQueueFactory"/>.
/// Tests call the internal encode/decode methods directly.
/// Persistor cleanup clears the process-wide batch directory.
/// </summary>
[TestFixture]
[Category("Server")]
[SetCulture("en-us")]
[SetUICulture("en-us")]
[NonParallelizable]
public class DurableSubscriptionSerializationTests
{
private ServiceMessageContext m_context;
/// <summary>
/// Creates the serialization message context and registers the two fixture namespaces.
/// </summary>
[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");
}
/// <summary>
/// Verifies that an empty durable subscription preserves its settings and empty item and message collections.
/// </summary>
[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));
}
/// <summary>
/// Verifies that the current sample format preserves publishing state and anonymous owner application URI.
/// </summary>
[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"));
});
}
/// <summary>
/// Verifies that the current sample format preserves the triggering links between
/// monitored items (OPC 10000-4 §5.13.1.6).
/// </summary>
[Test]
public void RoundTripSubscriptionTriggeringLinks()
{
StoredSubscription original = CreateMinimalSubscription(id: 45);
original.TriggeringLinks = new Dictionary<uint, IReadOnlyList<uint>>
{
[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 }));
});
}
/// <summary>
/// Verifies that version-one sample records migrate to the safe default publishing state.
/// </summary>
[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);
});
}
/// <summary>
/// Verifies that durable subscription serialization preserves its monitored items and their identifiers.
/// </summary>
[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));
}
/// <summary>
/// Verifies that durable subscription serialization preserves monitored-item properties.
/// </summary>
[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));
}
/// <summary>
/// Verifies that the retired interim notification-state format is rejected before decoding a subscription.
/// </summary>
[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<InvalidDataException>()
.With.Message.Contains("Unsupported durable subscription store version 2"));
Assert.That(decoder.Position, Is.Zero);
}
/// <summary>
/// Verifies that new subscription records contain raw monitored-item state rather than transient notifications.
/// </summary>
[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));
}
/// <summary>
/// Verifies that durable subscription serialization preserves the user identity token.
/// </summary>
[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<UserNameIdentityToken>());
var token = (UserNameIdentityToken)result.UserIdentityToken;
Assert.That(token.UserName, Is.EqualTo("testuser"));
Assert.That(token.PolicyId, Is.EqualTo("username_policy"));
}
/// <summary>
/// Verifies that multiple durable subscriptions round-trip independently.
/// </summary>
[Test]
public void RoundTripMultipleSubscriptions()
{
var subs = new List<StoredSubscription>
{
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<StoredSubscription> 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);
}
/// <summary>
/// Verifies that a storable data-change queue round-trips through serialization.
/// </summary>
[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);
}
/// <summary>
/// Verifies that a storable data-change queue preserves its batches during serialization.
/// </summary>
[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));
}
/// <summary>
/// Verifies that a storable event queue round-trips through serialization.
/// </summary>
[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();
}
/// <summary>
/// Encodes and decodes a list of subscriptions using
/// the internal SubscriptionStore methods.
/// </summary>
private List<StoredSubscription> RoundTripSubscriptions(
List<StoredSubscription> 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<string> nsUris = decoder.ReadStringArray(null);
ArrayOf<string> srvUris = decoder.ReadStringArray(null);
decoder.SetMappingTables(
new NamespaceTable(nsUris.Memory.ToArray()),
new StringTable(srvUris.Memory.ToArray()));
int count = decoder.ReadInt32(null);
var results = new List<StoredSubscription>(count);
for (int i = 0; i < count; i++)
{
results.Add(
SubscriptionStore.DecodeSubscription(decoder));
}
return results;
}
/// <summary>
/// Encodes and decodes a StorableDataChangeQueue using
/// the internal DurableMonitoredItemQueueFactory methods.
/// </summary>
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<string> nsUris = decoder.ReadStringArray(null);
ArrayOf<string> srvUris = decoder.ReadStringArray(null);
decoder.SetMappingTables(
new NamespaceTable(nsUris.Memory.ToArray()),
new StringTable(srvUris.Memory.ToArray()));
return DurableMonitoredItemQueueFactory
.DecodeDataChangeQueue(decoder);
}
/// <summary>
/// Encodes and decodes a StorableEventQueue using
/// the internal DurableMonitoredItemQueueFactory methods.
/// </summary>
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<string> nsUris = decoder.ReadStringArray(null);
ArrayOf<string> srvUris = decoder.ReadStringArray(null);
decoder.SetMappingTables(
new NamespaceTable(nsUris.Memory.ToArray()),
new StringTable(srvUris.Memory.ToArray()));
return DurableMonitoredItemQueueFactory
.DecodeEventQueue(decoder);
}
/// <summary>
/// Verifies that a data-change batch round-trips through the queue persistor.
/// </summary>
[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([]);
}
/// <summary>
/// Verifies that an event batch round-trips through the queue persistor.
/// </summary>
[Test]
public void RoundTripEventBatchViaPersistor()
{
using IDisposable scope =
AmbientMessageContext.SetScopedContext(m_context);
var events = new List<EventFieldList>
{
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([]);
}
/// <summary>
/// Verifies that deleting a persisted queue batch removes its backing file.
/// </summary>
[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);
}
}
}