mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-24 02:58:36 +08:00
efad0ddeff
# 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>
684 lines
27 KiB
C#
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);
|
|
}
|
|
}
|
|
}
|