/* ========================================================================
* Copyright (c) 2005-2026 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/
* ======================================================================*/
#pragma warning disable CA2007
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Moq;
using NUnit.Framework;
namespace Opc.Ua.Server.Tests
{
[TestFixture]
[Category("Subscription")]
[Parallelizable]
public class SentMessageQueueTests
{
///
/// Verifies that restoration preserves sequence state and dequeues an independent copy of the queued message.
///
[Test]
public void CreateRestoredPreservesQueueStateAndDequeuesExistingMessages()
{
List messages =
[
CreateMessage(10),
CreateMessage(11)
];
var queue = SentMessageQueue.CreateRestored(
() => 7,
maxMessageCount: 5,
retransmissionStore: null,
Mock.Of(),
messages,
nextSequenceNumber: 42,
lastSentMessage: 1);
var availableSequenceNumbers = new List();
NotificationMessage? result = queue.TryDequeueQueued(
availableSequenceNumbers,
hasItemsToPublish: false,
out bool moreNotifications);
Assert.Multiple(() =>
{
Assert.That(queue.NextSequenceNumber, Is.EqualTo(42u));
Assert.That(queue.LastSentMessage, Is.EqualTo(2));
Assert.That(result, Is.Not.SameAs(messages[1]));
Assert.That(result!.IsEqual(messages[1]), Is.True);
Assert.That(availableSequenceNumbers, Has.Count.EqualTo(2));
Assert.That(moreNotifications, Is.False);
});
}
[Test]
public void EnqueueUsesDeltaStoreWhenAvailableAndReportsRemovedSequenceNumbers()
{
var deltaStore = new Mock();
var queue = SentMessageQueue.CreateRestored(
() => 9,
maxMessageCount: 2,
deltaStore.Object,
Mock.Of(),
[CreateMessage(1), CreateMessage(2)],
nextSequenceNumber: 3,
lastSentMessage: 2);
var availableSequenceNumbers = new List();
NotificationMessage published = queue.Enqueue(
[CreateMessage(3)],
availableSequenceNumbers,
out bool moreNotifications,
out uint newlyUnacknowledgedCount);
Assert.Multiple(() =>
{
Assert.That(published.SequenceNumber, Is.EqualTo(3u));
Assert.That(queue.SentCount, Is.EqualTo(2));
Assert.That(newlyUnacknowledgedCount, Is.EqualTo(1u));
Assert.That(moreNotifications, Is.False);
Assert.That(availableSequenceNumbers, Has.Count.EqualTo(2));
});
deltaStore.Verify(s => s.StoreRetransmissionStateDelta(
9,
3,
It.Is>(m => m.Count == 1 && m[0].SequenceNumber == 3),
It.Is>(r => r.Count == 1 && r[0] == 1)),
Times.Once);
}
[Test]
public void EnqueueTrimsOversizedBatchBeforeStoringSnapshot()
{
var store = new Mock();
var logger = new Mock();
var queue = new SentMessageQueue(
() => 11,
maxMessageCount: 2,
store.Object,
logger.Object);
var availableSequenceNumbers = new List();
NotificationMessage published = queue.Enqueue(
[CreateMessage(1), CreateMessage(2), CreateMessage(3)],
availableSequenceNumbers,
out bool moreNotifications,
out uint newlyUnacknowledgedCount);
Assert.Multiple(() =>
{
Assert.That(published.SequenceNumber, Is.EqualTo(2u));
Assert.That(queue.SentCount, Is.EqualTo(2));
// the trimmed unsent message counts as discarded before acknowledgement.
Assert.That(newlyUnacknowledgedCount, Is.EqualTo(1u));
Assert.That(moreNotifications, Is.True);
Assert.That(availableSequenceNumbers, Has.Count.EqualTo(1));
});
store.Verify(s => s.StoreRetransmissionState(
11,
1,
It.Is>(m => m.Count == 2 && m[0].SequenceNumber == 2)),
Times.Once);
}
[Test]
public async Task LoadRetransmissionStateAsyncRestoresStateFromStoreAsync()
{
var state = new SubscriptionRetransmissionState
{
NextSequenceNumber = 100,
SentMessages = [CreateMessage(20), CreateMessage(21)]
};
var store = new Mock();
store.Setup(s => s.LoadRetransmissionStateAsync(12, It.IsAny()))
.Returns(new ValueTask(state));
var queue = new SentMessageQueue(
() => 12,
maxMessageCount: 5,
store.Object,
Mock.Of());
await queue.LoadRetransmissionStateAsync(CancellationToken.None).ConfigureAwait(false);
Assert.Multiple(() =>
{
Assert.That(queue.NextSequenceNumber, Is.EqualTo(100u));
Assert.That(queue.LastSentMessage, Is.EqualTo(2));
Assert.That(queue.AvailableSequenceNumbersForRetransmission(), Has.Count.EqualTo(2));
});
}
[Test]
public void EnqueueWithZeroCapacityStillReturnsNotification()
{
var queue = new SentMessageQueue(
() => 14,
maxMessageCount: 0,
retransmissionStore: null,
Mock.Of());
var availableSequenceNumbers = new List();
NotificationMessage published = queue.Enqueue(
[CreateMessage(1)],
availableSequenceNumbers,
out bool moreNotifications,
out uint newlyUnacknowledgedCount);
Assert.Multiple(() =>
{
Assert.That(published.SequenceNumber, Is.EqualTo(1u));
Assert.That(queue.SentCount, Is.EqualTo(1));
Assert.That(moreNotifications, Is.False);
Assert.That(newlyUnacknowledgedCount, Is.Zero);
Assert.That(availableSequenceNumbers, Has.Count.EqualTo(1));
});
}
[Test]
public void TryAcknowledgeRemovesMessageAndMirrorsAcknowledgement()
{
var store = new Mock();
var queue = SentMessageQueue.CreateRestored(
() => 13,
maxMessageCount: 5,
store.Object,
Mock.Of(),
[CreateMessage(31), CreateMessage(32)],
nextSequenceNumber: 33,
lastSentMessage: 2);
bool removed = queue.TryAcknowledge(31);
bool missing = queue.TryAcknowledge(99);
Assert.Multiple(() =>
{
Assert.That(removed, Is.True);
Assert.That(missing, Is.False);
Assert.That(queue.SentCount, Is.EqualTo(1));
});
store.Verify(s => s.AcknowledgeNotification(13, 31), Times.Once);
}
[Test]
public void CreateRestoredWithNullEntryKeepsUnsentMessageQueued()
{
var queue = SentMessageQueue.CreateRestored(
() => 15,
maxMessageCount: 5,
null,
Mock.Of(),
[CreateMessage(7), null!, CreateMessage(8)],
nextSequenceNumber: 9,
lastSentMessage: 2);
// 7 and the null entry were in front of the cursor; 8 was not sent yet.
NotificationMessage? next = queue.TryDequeueQueued([], hasItemsToPublish: false, out _);
Assert.Multiple(() =>
{
Assert.That(next!.SequenceNumber, Is.EqualTo(8u));
Assert.That(queue.FindForRepublish(7), Is.Not.Null);
});
}
[TestCase(13u, 11u, new uint[] { 10, 12 }, 12u)]
[TestCase(3u, uint.MaxValue, new uint[] { uint.MaxValue - 1, 1, 2 }, 1u)]
public async Task LoadRetransmissionStateAsyncFindsBoundaryWhenFirstUnsentMessageIsMissingAsync(
uint nextSequenceNumber,
uint firstUnsentSequenceNumber,
uint[] retained,
uint expectedFirstUnsent)
{
var state = new SubscriptionRetransmissionState
{
NextSequenceNumber = nextSequenceNumber,
SentMessages = [.. retained.Select(CreateMessage)],
FirstUnsentSequenceNumber = firstUnsentSequenceNumber
};
var store = new Mock();
store.Setup(s => s.LoadRetransmissionStateAsync(21, It.IsAny()))
.Returns(new ValueTask(state));
var queue = new SentMessageQueue(
() => 21,
maxMessageCount: 5,
store.Object,
Mock.Of());
await queue.LoadRetransmissionStateAsync(CancellationToken.None).ConfigureAwait(false);
NotificationMessage? first = queue.TryDequeueQueued([], hasItemsToPublish: false, out _);
Assert.That(first!.SequenceNumber, Is.EqualTo(expectedFirstUnsent));
}
[Test]
public void TryAcknowledgeRejectsMessageNotSentYet()
{
var store = new Mock();
var queue = SentMessageQueue.CreateRestored(
() => 14,
maxMessageCount: 5,
store.Object,
Mock.Of(),
[CreateMessage(31), CreateMessage(32)],
nextSequenceNumber: 33,
lastSentMessage: 1);
// 32 is still queued for Publish: the client cannot know it, so acknowledging it
// fails (Bad_SequenceNumberUnknown) and it is still delivered.
bool acknowledged = queue.TryAcknowledge(32);
NotificationMessage? next = queue.TryDequeueQueued([], hasItemsToPublish: false, out _);
Assert.Multiple(() =>
{
Assert.That(acknowledged, Is.False);
Assert.That(next!.SequenceNumber, Is.EqualTo(32u));
});
store.Verify(s => s.AcknowledgeNotification(14, 32), Times.Never);
}
[Test]
public void CreateRestoredWithMissingStateStartsEmptyAtSequenceNumberOne()
{
var queue = SentMessageQueue.CreateRestored(
() => 15,
maxMessageCount: 5,
retransmissionStore: null,
Mock.Of(),
sentMessages: null,
nextSequenceNumber: 0,
lastSentMessage: -3);
Assert.Multiple(() =>
{
Assert.That(queue.SentCount, Is.Zero);
Assert.That(queue.LastSentMessage, Is.Zero);
// OPC 10000-4, 5.14.1.1: sequence number 0 is never used.
Assert.That(queue.NextSequenceNumber, Is.EqualTo(1u));
Assert.That(queue.AssignSequenceNumber(), Is.EqualTo(1u));
});
}
[Test]
public void CreateRestoredClampsSentIndexAndContinuesAfterNewestMessage()
{
var tooLarge = SentMessageQueue.CreateRestored(
() => 16,
maxMessageCount: 5,
retransmissionStore: null,
Mock.Of(),
[CreateMessage(5), CreateMessage(6)],
nextSequenceNumber: 0,
lastSentMessage: 10);
var negative = SentMessageQueue.CreateRestored(
() => 17,
maxMessageCount: 5,
retransmissionStore: null,
Mock.Of(),
[CreateMessage(5), CreateMessage(6)],
nextSequenceNumber: 7,
lastSentMessage: -1);
NotificationMessage? queued = negative.TryDequeueQueued(
[],
hasItemsToPublish: false,
out bool moreNotifications);
Assert.Multiple(() =>
{
Assert.That(tooLarge.LastSentMessage, Is.EqualTo(2));
Assert.That(tooLarge.NextSequenceNumber, Is.EqualTo(7u));
Assert.That(queued!.SequenceNumber, Is.EqualTo(5u));
Assert.That(moreNotifications, Is.True);
});
}
[Test]
public async Task LoadRetransmissionStateAsyncMapsSequenceNumberZeroAsync()
{
var state = new SubscriptionRetransmissionState
{
NextSequenceNumber = 0,
SentMessages = [CreateMessage(20), CreateMessage(21)]
};
var store = new Mock();
store.Setup(s => s.LoadRetransmissionStateAsync(18, It.IsAny()))
.Returns(new ValueTask(state));
var queue = new SentMessageQueue(
() => 18,
maxMessageCount: 5,
store.Object,
Mock.Of());
await queue.LoadRetransmissionStateAsync(CancellationToken.None).ConfigureAwait(false);
Assert.That(queue.NextSequenceNumber, Is.EqualTo(22u));
}
[Test]
public void SendStateStoreMirrorsFirstUnsentSequenceNumber()
{
var store = new Mock();
var queue = new SentMessageQueue(
() => 19,
maxMessageCount: 5,
store.Object,
Mock.Of());
queue.Enqueue(
[CreateMessage(10), CreateMessage(11), CreateMessage(12)],
[],
out bool moreNotifications,
out _);
Assert.That(moreNotifications, Is.True);
store.Verify(s => s.StoreFirstUnsentSequenceNumber(19, It.IsAny(), 11), Times.Once);
queue.TryDequeueQueued([], hasItemsToPublish: false, out _);
store.Verify(s => s.StoreFirstUnsentSequenceNumber(19, It.IsAny(), 12), Times.Once);
queue.TryDequeueQueued([], hasItemsToPublish: false, out _);
store.Verify(s => s.StoreFirstUnsentSequenceNumber(19, It.IsAny(), 0), Times.Once);
// Nothing changes: no further mirror writes.
queue.TryAcknowledge(10);
queue.Enqueue([CreateMessage(13)], [], out _, out _);
store.Verify(
s => s.StoreFirstUnsentSequenceNumber(It.IsAny(), It.IsAny(), It.IsAny()),
Times.Exactly(3));
}
[Test]
public async Task LoadRetransmissionStateAsyncKeepsUnsentMessagesQueuedForPublishAsync()
{
var state = new SubscriptionRetransmissionState
{
NextSequenceNumber = 13,
SentMessages = [CreateMessage(10), CreateMessage(11), CreateMessage(12)],
FirstUnsentSequenceNumber = 11
};
var store = new Mock();
store.Setup(s => s.LoadRetransmissionStateAsync(20, It.IsAny()))
.Returns(new ValueTask(state));
var queue = new SentMessageQueue(
() => 20,
maxMessageCount: 5,
store.Object,
Mock.Of());
await queue.LoadRetransmissionStateAsync(CancellationToken.None).ConfigureAwait(false);
// OPC 10000-4, 5.14.1.1: the backlog is delivered by Publish, not only by Republish.
NotificationMessage? first = queue.TryDequeueQueued([], hasItemsToPublish: false, out bool more);
NotificationMessage? second = queue.TryDequeueQueued([], hasItemsToPublish: false, out _);
NotificationMessage? none = queue.TryDequeueQueued([], hasItemsToPublish: false, out _);
Assert.Multiple(() =>
{
Assert.That(first!.SequenceNumber, Is.EqualTo(11u));
Assert.That(more, Is.True);
Assert.That(second!.SequenceNumber, Is.EqualTo(12u));
Assert.That(none, Is.Null);
});
}
[Test]
public async Task SendStateAfterRestoreCarriesNextSequenceNumberAsync()
{
var state = new SubscriptionRetransmissionState
{
NextSequenceNumber = 13,
SentMessages = [CreateMessage(10), CreateMessage(11), CreateMessage(12)],
FirstUnsentSequenceNumber = 11
};
var store = new Mock();
store.Setup(s => s.LoadRetransmissionStateAsync(21, It.IsAny()))
.Returns(new ValueTask(state));
var queue = new SentMessageQueue(
() => 21,
maxMessageCount: 5,
store.Object,
Mock.Of());
await queue.LoadRetransmissionStateAsync(CancellationToken.None).ConfigureAwait(false);
// The first send-state write after a restore is not preceded by a retransmission
// state write, so it must carry the restored next sequence number itself.
queue.TryDequeueQueued([], hasItemsToPublish: false, out _);
store.Verify(s => s.StoreFirstUnsentSequenceNumber(21, 13, 12), Times.Once);
}
private static NotificationMessage CreateMessage(uint sequenceNumber)
{
return new NotificationMessage
{
SequenceNumber = sequenceNumber,
NotificationData = []
};
}
}
}