/* ========================================================================
* 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.Threading.Tasks;
using Microsoft.Extensions.Logging.Abstractions;
using Moq;
using NUnit.Framework;
using Opc.Ua.Tests;
namespace Opc.Ua.Server.Tests
{
///
/// Verifies that returned and mirrored notifications retain their payloads independently of queue recycling.
///
[TestFixture]
[Category("Subscription")]
[NonParallelizable]
public sealed class NotificationEncodingLifetimeRegressionTests
{
///
/// Verifies that delayed encoding preserves notifications after acknowledgement, eviction, clearing, and pool
/// reuse.
///
[Test]
public async Task ReturnedNotificationSurvivesPoolReuseUntilEncodingCompletesAsync(
[Values("publish", "queued", "republish", "fullMirror", "deltaMirror")] string surface,
[Values("acknowledge", "evict", "clear")] string mutation,
[Values(false, true)] bool events)
{
NotificationMessage? mirrored = null;
var fullStore = new Mock();
fullStore.Setup(store => store.StoreRetransmissionState(
1, It.IsAny(), It.IsAny>()))
.Callback>((_, _, values) => mirrored ??= values[0]);
var deltaStore = new Mock();
deltaStore.Setup(store => store.StoreRetransmissionStateDelta(
1, It.IsAny(), It.IsAny>(), It.IsAny>()))
.Callback, ArrayOf>(
(_, _, values, _) => mirrored ??= values[0]);
ISubscriptionRetransmissionStore? store = surface switch
{
"fullMirror" => fullStore.Object,
"deltaMirror" => deltaStore.Object,
_ => null
};
var queue = new SentMessageQueue(() => 1, 2, store, NullLogger.Instance);
var source = (NotificationMessage)NotificationMessageActivator.Instance.CreateInstance();
source.SequenceNumber = surface == "queued" ? 2u : 1u;
var item = (MonitoredItemNotification)MonitoredItemNotificationActivator.Instance.CreateInstance();
item.ClientHandle = 77;
item.Value = new DataValue(42);
var eventItem = (EventFieldList)EventFieldListActivator.Instance.CreateInstance();
eventItem.ClientHandle = 77;
eventItem.EventFields = [42];
var data = new DataChangeNotification { MonitoredItems = [item] };
var eventList = new EventNotificationList { Events = [eventItem] };
source.NotificationData = [new ExtensionObject(events ? eventList : data)];
List messages = surface == "queued"
? [new NotificationMessage { SequenceNumber = 1 }, source]
: [source];
NotificationMessage published = queue.Enqueue(messages, [], out _, out _);
NotificationMessage selected = surface switch
{
"queued" => queue.TryDequeueQueued([], false, out _)!,
"republish" => queue.FindForRepublish(source.SequenceNumber)!,
"fullMirror" or "deltaMirror" => mirrored!,
_ => published
};
uint expectedSequence = source.SequenceNumber;
var releaseEncoder = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var response = new PublishResponse { NotificationMessage = selected };
Task encoding = EncodeAfterReleaseAsync(response, releaseEncoder.Task);
try
{
if (mutation == "acknowledge")
{
Assert.That(queue.TryAcknowledge(expectedSequence), Is.True);
}
else if (mutation == "evict")
{
queue.Enqueue(
[new NotificationMessage { SequenceNumber = 3 }, new NotificationMessage { SequenceNumber = 4 }],
[], out _, out _);
}
else
{
queue.Clear();
}
Assert.That(source.IsEmpty, Is.True);
}
finally
{
releaseEncoder.TrySetResult(true);
}
var reused = (NotificationMessage)NotificationMessageActivator.Instance.CreateInstance();
reused.SequenceNumber = 999;
PublishResponse decoded;
try
{
decoded = await encoding.ConfigureAwait(false);
}
finally
{
reused.Reuse();
}
Assert.That(decoded.NotificationMessage.SequenceNumber, Is.EqualTo(expectedSequence));
Assert.That(decoded.NotificationMessage.NotificationData, Has.Count.EqualTo(1));
if (events)
{
Assert.That(decoded.NotificationMessage.NotificationData[0].TryGetValue(out EventNotificationList? value),
Is.True);
Assert.That(value!.Events[0].ClientHandle, Is.EqualTo(77u));
Assert.That(value.Events[0].EventFields[0].GetInt32(), Is.EqualTo(42));
Assert.That(eventItem.ClientHandle, Is.Zero);
item.Reuse();
}
else
{
Assert.That(decoded.NotificationMessage.NotificationData[0].TryGetValue(out DataChangeNotification? value),
Is.True);
Assert.That(value!.MonitoredItems[0].ClientHandle, Is.EqualTo(77u));
Assert.That(value.MonitoredItems[0].Value.WrappedValue.GetInt32(), Is.EqualTo(42));
Assert.That(item.ClientHandle, Is.Zero);
eventItem.Reuse();
}
queue.Clear();
}
///
/// Waits for queue mutation before round-tripping the retained publish response through binary encoding.
///
private static async Task EncodeAfterReleaseAsync(PublishResponse response, Task release)
{
await release.ConfigureAwait(false);
ServiceMessageContext context = ServiceMessageContext.Create(NUnitTelemetryContext.Create());
using var encoder = new BinaryEncoder(context);
response.Encode(encoder);
using var decoder = new BinaryDecoder(encoder.CloseAndReturnBuffer()!, context);
var decoded = new PublishResponse();
decoded.Decode(decoder);
return decoded;
}
}
}