/* ======================================================================== * 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/ * ======================================================================*/ // CA2007: tests run without a SynchronizationContext; ConfigureAwait(false) // adds noise without a behavioural benefit. Disabled file-level for the suite. #pragma warning disable CA2007 using System; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; using Moq; using NUnit.Framework; using Opc.Ua.Client.Subscriptions.Fakes; using Opc.Ua.Tests; namespace Opc.Ua.Client.Subscriptions { [TestFixture] public sealed class MessageProcessorTests { [SetUp] public void SetUp() { m_completion = new FakeMessageAckQueue(); m_telemetry = NUnitTelemetryContext.Create(); m_mockServices = new Mock(); } [Test] public async Task DisposeAsyncShouldCompleteMessageWriterAndCancelTokenAsync() { // Arrange var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 3 }; // Act await sut.DisposeAsync().ConfigureAwait(false); // Assert Assert.That(sut.PublishState, Is.EqualTo(PublishState.Completed)); Assert.That(m_completion.CompletedSubscriptions, Is.EqualTo(new uint[] { 3 })); } [Test] public async Task OnPublishReceivedKeepAliveShouldDispatchKeepAliveAsync() { // Arrange var message = new NotificationMessage { SequenceNumber = 3 }; var availableSequenceNumbers = new List { 1, 2, 3 }; var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 3 }; await using (sut.ConfigureAwait(false)) { // Act await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.KeepAliveNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); // Assert Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True); Assert.That(sut.AvailableInRetransmissionQueue, Is.EqualTo(availableSequenceNumbers)); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(3)); Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.False); Assert.That(m_completion.QueuedAcks, Is.Empty); // Arrange sut.KeepAliveNotificationReceived.Reset(); sut.DataChangeNotificationReceived.Reset(); message = new NotificationMessage { SequenceNumber = 4, NotificationData = [ new ExtensionObject(new DataChangeNotification { MonitoredItems = [ new MonitoredItemNotification() ] }) ] }; // Act await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.DataChangeNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); // Assert Assert.That(sut.AvailableInRetransmissionQueue, Is.EqualTo(availableSequenceNumbers)); Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.True); Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.False); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(4)); Assert.That(m_completion.QueuedAcks, Has.Count.EqualTo(1)); Assert.That(m_completion.QueuedAcks[0].SubscriptionId, Is.EqualTo(3u)); Assert.That(m_completion.QueuedAcks[0].SequenceNumber, Is.EqualTo(4u)); } } [Test] public async Task ProcessMessageAsyncShouldRepublishMissingMessagesAsync() { // Arrange var availableSequenceNumbers = new List { 1, 2, 3 }; var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 2 }; await using (sut.ConfigureAwait(false)) { await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 1 }, availableSequenceNumbers, stringTable).ConfigureAwait(false); m_mockServices .Setup(c => c.RepublishAsync( It.IsAny(), It.Is(id => id == sut.Id), It.Is(s => s == 2), It.IsAny())) .ReturnsAsync(new RepublishResponse { NotificationMessage = new NotificationMessage { SequenceNumber = 2, NotificationData = [ new ExtensionObject(new DataChangeNotification { MonitoredItems = [ new MonitoredItemNotification() ] }) ] } }) .Verifiable(Times.Once); // Act await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 3, NotificationData = [ new ExtensionObject(new EventNotificationList { Events = [ new EventFieldList() ] }) ] }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.EventNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); // Assert Assert.That(sut.EventNotificationReceived.IsSet, Is.True); Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True); Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.True); m_mockServices.Verify(); } } [Test] public async Task ProcessReceivedMessagesAsyncShouldProcessMessagesInOrderAsync() { var availableSequenceNumbers = new List { 1, 2, 3 }; var stringTable = new List { "test" }; // Arrange NotificationMessage[] messages = [.. Enumerable.Range(2, 99).Select(i => new NotificationMessage { SequenceNumber = (uint)i })]; UnsecureRandom.Shared.Shuffle(messages); var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 3 }; await using (sut.ConfigureAwait(false)) { sut.Block.Wait(); await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 1u }, availableSequenceNumbers, stringTable).ConfigureAwait(false); foreach (NotificationMessage message in messages) { await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable).ConfigureAwait(false); } sut.Block.Release(); // Act await Task.Delay(10).ConfigureAwait(false); Assert.That(sut.ReceivedSequenceNumbers, Is.EqualTo( Enumerable.Range(1, 100).Select(i => (uint)i))); Assert.That(sut.AvailableInRetransmissionQueue, Is.EqualTo(availableSequenceNumbers)); Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.False); Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(100)); } } [Test] public async Task DuplicateSequenceNumberShouldNotRedispatchAsync() { var availableSequenceNumbers = new List { 1, 2, 3 }; var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 7 }; await using (sut.ConfigureAwait(false)) { var message = new NotificationMessage { SequenceNumber = 5, NotificationData = [ new ExtensionObject(new DataChangeNotification { MonitoredItems = [ new MonitoredItemNotification() ] }) ] }; // First arrival await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable) .ConfigureAwait(false); await sut.DataChangeNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)) .ConfigureAwait(false); int firstCount = sut.ReceivedSequenceNumbers.Count; Assert.That(firstCount, Is.GreaterThanOrEqualTo(1)); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(5)); // Reset signaling sut.DataChangeNotificationReceived.Reset(); // Same sequence number arriving again await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable) .ConfigureAwait(false); // Give the processor a moment in case it tries to dispatch await Task.Delay(50).ConfigureAwait(false); // The duplicate should not re-fire the data-change handler. Assert.That( sut.DataChangeNotificationReceived.IsSet, Is.False, "Duplicate sequence number must not re-dispatch the notification"); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(5)); } } [Test] public async Task KeepAliveInterleavedWithNotificationsAsync() { var availableSequenceNumbers = new List { 1, 2, 3 }; var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 5 }; await using (sut.ConfigureAwait(false)) { // 1: notification await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 1, NotificationData = [ new ExtensionObject(new DataChangeNotification { MonitoredItems = [new MonitoredItemNotification()] }) ] }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.DataChangeNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)) .ConfigureAwait(false); // 2: keep-alive (no NotificationData) sut.KeepAliveNotificationReceived.Reset(); await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 2 }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.KeepAliveNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)) .ConfigureAwait(false); // 3: another notification sut.DataChangeNotificationReceived.Reset(); await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 3, NotificationData = [ new ExtensionObject(new DataChangeNotification { MonitoredItems = [new MonitoredItemNotification()] }) ] }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.DataChangeNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)) .ConfigureAwait(false); // All three sequence numbers should appear in order. Assert.That(sut.ReceivedSequenceNumbers, Is.EqualTo(new uint[] { 1, 2, 3 })); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(3)); } } [Test] public async Task EmptyNotificationDataIsTreatedAsKeepAliveAsync() { var availableSequenceNumbers = new List { 1, 2, 3 }; var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 9 }; await using (sut.ConfigureAwait(false)) { // NotificationData is null/default. await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 7 }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.KeepAliveNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)) .ConfigureAwait(false); Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True); Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.False); Assert.That(sut.EventNotificationReceived.IsSet, Is.False); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(7)); } } [Test] public async Task RepublishFailureLogsButContinuesProcessingAsync() { var availableSequenceNumbers = new List { 1, 2, 3 }; var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 11 }; await using (sut.ConfigureAwait(false)) { // First message at sequence 1. await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 1 }, availableSequenceNumbers, stringTable).ConfigureAwait(false); // Republish for the gap (sequence 2) returns an error. m_mockServices .Setup(c => c.RepublishAsync( It.IsAny(), It.Is(id => id == sut.Id), It.Is(s => s == 2), It.IsAny())) .ThrowsAsync(new ServiceResultException( StatusCodes.BadMessageNotAvailable)) .Verifiable(Times.AtLeastOnce); // Skip to sequence 3 to force the gap. await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 3, NotificationData = [ new ExtensionObject(new EventNotificationList { Events = [new EventFieldList()] }) ] }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.EventNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)) .ConfigureAwait(false); // Even though republish failed, the next message must still // be dispatched. Assert.That(sut.EventNotificationReceived.IsSet, Is.True); m_mockServices.Verify(); } } [Test] public async Task SequenceNumberWraparoundAdvancesLastProcessedAsync() { // Per OPC UA Part 4 §7.30.5, sequence numbers wrap from // uint.MaxValue to 1 (skipping 0). The processor must accept // this wrap as forward progress, not silently drop the // post-wrap message as "old". var availableSequenceNumbers = new List(); var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 13 }; await using (sut.ConfigureAwait(false)) { // Republish for the first message's "missing" gap (we start // from LastSequenceNumberProcessed=0 so no gap is computed // for this first one) — but for the second message, the gap // is empty because we wrap directly from uint.MaxValue to 1. await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = uint.MaxValue }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.KeepAliveNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)) .ConfigureAwait(false); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(uint.MaxValue)); sut.KeepAliveNotificationReceived.Reset(); await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 1 }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.KeepAliveNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)) .ConfigureAwait(false); Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True, "Wrapped sequence number must be accepted as forward progress, " + "not dropped as duplicate/old."); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(1)); } } [Test] public async Task ReceivingTransferStatusUpdateShouldUpdatePublishStateAsync() { // Arrange var availableSequenceNumbers = new List { 1, 2, 3 }; var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 3 }; await using (sut.ConfigureAwait(false)) { // Act await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 3, NotificationData = [ new ExtensionObject(new StatusChangeNotification { Status = StatusCodes.GoodSubscriptionTransferred }) ] }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.StatusChangeNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); // Assert Assert.That(sut.StatusChangeNotificationReceived.IsSet, Is.True); Assert.That(sut.ReceivedSequenceNumbers, Does.Contain(3)); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(3)); Assert.That(sut.PublishState, Is.EqualTo(PublishState.Transferred)); sut.StatusChangeNotificationReceived.Reset(); // Act await sut.OnPublishReceivedAsync(new NotificationMessage { SequenceNumber = 4, NotificationData = [ new ExtensionObject(new StatusChangeNotification { Status = StatusCodes.BadTimeout }) ] }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.StatusChangeNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); // Assert Assert.That(sut.StatusChangeNotificationReceived.IsSet, Is.True); Assert.That(sut.ReceivedSequenceNumbers, Does.Contain(4)); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(4)); Assert.That(sut.PublishState, Is.EqualTo(PublishState.Timeout)); } } /// /// V2 equivalent of the classic-engine regression captured by /// OldMessagesAbandonedAfterRepublishTimeoutInSequentialModeAsync /// (currently [Explicit] in Classic/SubscriptionUnitTests.cs). /// /// /// Mirrors the classic test's exact arrival order — [3], /// [2], [4], [1], [5] — but queues all five /// while the worker is parked inside a sentinel keep-alive dispatch. /// The new uses /// /// keyed by , so any /// items present in the channel at a single read are sorted by /// sequence number. With the consumer held, the five data messages /// land together and drain as [1, 2, 3, 4, 5] with no /// republishes — the reliability guarantee the classic engine fails /// to provide. (Without the sentinel, a fast consumer could read the /// first arrival before the rest queue, which is what the priority /// channel cannot recover from on its own.) /// [Test] public async Task OutOfOrderArrivalsReorderedByPriorityChannelAsync() { var availableSequenceNumbers = new List { 1, 2, 3, 4, 5 }; var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 11 }; await using (sut.ConfigureAwait(false)) { // Arrange: drain the gate, then park the worker inside a // sentinel keep-alive dispatch so all five data messages // queue up in the priority channel before any are read. // The keep-alive does not advance the data dedup gate // (LastDataSequenceNumberProcessed) so the burst is still // processed as a fresh sequence starting at 1. sut.Block.Wait(); await sut.OnPublishReceivedAsync( new NotificationMessage { SequenceNumber = 1u }, availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.KeepAliveNotificationReceived.WaitAsync() .WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); foreach (uint seq in new uint[] { 3, 2, 4, 1, 5 }) { await sut.OnPublishReceivedAsync(BuildDataChangeMessage(seq), availableSequenceNumbers, stringTable).ConfigureAwait(false); } sut.Block.Release(); // Act await WaitForLastSeqNumberAsync(sut, 5).ConfigureAwait(false); // Assert: the first entry is the sentinel keep-alive; the // remaining entries must be the data burst in strict // sequence-number order. Assert.That(sut.ReceivedSequenceNumbers, Has.Count.EqualTo(6)); Assert.That(sut.ReceivedSequenceNumbers[0], Is.EqualTo(1u), "first entry should be the sentinel keep-alive"); Assert.That(sut.ReceivedSequenceNumbers.Skip(1), Is.EqualTo(new uint[] { 1, 2, 3, 4, 5 })); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(5)); Assert.That(sut.MissingMessageCount, Is.Zero); Assert.That(sut.RepublishMessageCount, Is.Zero); m_mockServices.Verify( c => c.RepublishAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); } } /// /// V2 equivalent of the classic-engine regression captured by /// OldMessagesAbandonedAfterRepublishTimeoutInSequentialModeAsync, /// stressed in serial timing instead of as a burst. /// /// /// Same arrival order — [3], [2], [4], [1], /// [5] — but each forward-progressing push waits for delivery /// before the next is queued. The V2 engine's monotonic dedup gate /// (LastDataSequenceNumberProcessed) silently discards the /// stale [2] and [1] instead of re-inserting them /// out of order; [3, 4, 5] are delivered in strict ascending /// order with no republishes. /// [Test] public async Task LateArrivalsDiscardedAfterDedupGateAdvancedAsync() { var availableSequenceNumbers = new List(); var stringTable = new List { "test" }; var sut = new TestMessageProcessor(m_mockServices.Object, m_completion, m_telemetry) { Id = 12 }; await using (sut.ConfigureAwait(false)) { // [3]: first data message after create. The empty // AvailableInRetransmissionQueue suppresses the // first-after-create speculative republish path, so [3] // goes straight to the handler. await sut.OnPublishReceivedAsync(BuildDataChangeMessage(3), availableSequenceNumbers, stringTable).ConfigureAwait(false); await WaitForLastSeqNumberAsync(sut, 3).ConfigureAwait(false); // [2] is strictly behind LastDataSeq=3. Queue [2] then [4] // back-to-back; the priority channel picks [2] first // (lowest SequenceNumber) so the worker reads [2], discards // it silently, then reads [4] and delivers it. Waiting for // LastSeq == 4 confirms both happened in that order. await sut.OnPublishReceivedAsync(BuildDataChangeMessage(2), availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.OnPublishReceivedAsync(BuildDataChangeMessage(4), availableSequenceNumbers, stringTable).ConfigureAwait(false); await WaitForLastSeqNumberAsync(sut, 4).ConfigureAwait(false); // Same shape for [1] (even staler) followed by [5]. await sut.OnPublishReceivedAsync(BuildDataChangeMessage(1), availableSequenceNumbers, stringTable).ConfigureAwait(false); await sut.OnPublishReceivedAsync(BuildDataChangeMessage(5), availableSequenceNumbers, stringTable).ConfigureAwait(false); await WaitForLastSeqNumberAsync(sut, 5).ConfigureAwait(false); // Assert: only forward-progressing messages were delivered, // in strict order; the late [2] and [1] were discarded. Assert.That(sut.ReceivedSequenceNumbers, Is.EqualTo(new uint[] { 3, 4, 5 })); Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(5)); Assert.That(sut.MissingMessageCount, Is.Zero); Assert.That(sut.RepublishMessageCount, Is.Zero); m_mockServices.Verify( c => c.RepublishAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); } } private static NotificationMessage BuildDataChangeMessage(uint sequenceNumber) { return new NotificationMessage { SequenceNumber = sequenceNumber, NotificationData = [ new ExtensionObject(new DataChangeNotification { MonitoredItems = [ new MonitoredItemNotification() ] }) ] }; } private static async Task WaitForLastSeqNumberAsync( TestMessageProcessor sut, uint expected, int timeoutSeconds = 5) { DateTimeOffset deadline = DateTimeOffset.UtcNow.AddSeconds(timeoutSeconds); while (sut.LastSequenceNumberProcessed < expected) { if (DateTimeOffset.UtcNow > deadline) { Assert.Fail( $"Timeout: LastSequenceNumberProcessed = {sut.LastSequenceNumberProcessed}, " + $"expected = {expected}, received = " + $"[{string.Join(",", sut.ReceivedSequenceNumbers)}]"); } await Task.Delay(10).ConfigureAwait(false); } } private sealed class TestMessageProcessor : MessageProcessor { public TestMessageProcessor(ISubscriptionServiceSetClientMethods session, IMessageAckQueue completion, ITelemetryContext telemetry) : base(session, completion, telemetry) { } public new IReadOnlyList AvailableInRetransmissionQueue { get => base.AvailableInRetransmissionQueue; set => base.AvailableInRetransmissionQueue = value; } public SemaphoreSlim Block { get; } = new(1, 1); public AsyncManualResetEvent DataChangeNotificationReceived { get; } = new(); public AsyncManualResetEvent EventNotificationReceived { get; } = new(); public AsyncManualResetEvent KeepAliveNotificationReceived { get; } = new(); public new uint LastSequenceNumberProcessed { get => base.LastSequenceNumberProcessed; set => base.LastSequenceNumberProcessed = value; } public PublishState PublishState { get; set; } public List ReceivedSequenceNumbers { get; } = []; public AsyncManualResetEvent StatusChangeNotificationReceived { get; } = new(); public async ValueTask WaitAsync() { await Block.WaitAsync().ConfigureAwait(false); Block.Release(); } protected override ValueTask OnDataChangeNotificationAsync(uint sequenceNumber, DateTime publishTime, DataChangeNotification notification, PublishState publishStateMask, IReadOnlyList stringTable) { ReceivedSequenceNumbers.Add(sequenceNumber); if (publishStateMask != PublishState.None) { PublishState = publishStateMask; } DataChangeNotificationReceived.Set(); return WaitAsync(); } protected override ValueTask OnEventDataNotificationAsync(uint sequenceNumber, DateTime publishTime, EventNotificationList notification, PublishState publishStateMask, IReadOnlyList stringTable) { ReceivedSequenceNumbers.Add(sequenceNumber); if (publishStateMask != PublishState.None) { PublishState = publishStateMask; } EventNotificationReceived.Set(); return WaitAsync(); } protected override ValueTask OnKeepAliveNotificationAsync(uint sequenceNumber, DateTime publishTime, PublishState publishStateMask) { ReceivedSequenceNumbers.Add(sequenceNumber); if (publishStateMask != PublishState.None) { PublishState = publishStateMask; } KeepAliveNotificationReceived.Set(); return WaitAsync(); } protected override void OnPublishStateChanged(PublishState stateMask) { PublishState = stateMask; base.OnPublishStateChanged(stateMask); } protected override async ValueTask OnStatusChangeNotificationAsync(uint sequenceNumber, DateTime publishTime, StatusChangeNotification notification, PublishState publishStateMask, IReadOnlyList stringTable) { ReceivedSequenceNumbers.Add(sequenceNumber); if (publishStateMask != PublishState.None) { PublishState = publishStateMask; } StatusChangeNotificationReceived.Set(); await WaitAsync().ConfigureAwait(false); } } private FakeMessageAckQueue m_completion; private ITelemetryContext m_telemetry; private Mock m_mockServices; } }