diff --git a/src/Opc.Ua.Core.Types/State/BaseEventState.cs b/src/Opc.Ua.Core.Types/State/BaseEventState.cs index 6286b2a44c..1e9f9f8ffe 100644 --- a/src/Opc.Ua.Core.Types/State/BaseEventState.cs +++ b/src/Opc.Ua.Core.Types/State/BaseEventState.cs @@ -44,8 +44,13 @@ namespace Opc.Ua EventSeverity severity, LocalizedText message) { - m_eventId = PropertyState.With(this, Uuid.NewUuid().ToByteString()); - m_eventType = PropertyState.With(this, GetDefaultTypeDefinitionId(context.NamespaceUris)); + // the fields carry the event time as their source timestamp: a field without + // one is stamped with the current time on every read, and an event's fields are + // read once per select clause and monitored item it is delivered to. + DateTimeUtc time = DateTimeUtc.Now; + + m_eventId = WithTimestamp(PropertyState.With(this, Uuid.NewUuid().ToByteString()), time); + m_eventType = WithTimestamp(PropertyState.With(this, GetDefaultTypeDefinitionId(context.NamespaceUris)), time); TypeDefinitionId = m_eventType.Value; if (source != null) @@ -55,6 +60,7 @@ namespace Opc.Ua m_sourceNode = new PropertyState.Implementation(this) { Value = source.NodeId, + Timestamp = time, RolePermissions = source.RolePermissions, UserRolePermissions = source.UserRolePermissions, NodeId = source.NodeId @@ -63,14 +69,28 @@ namespace Opc.Ua if (!source.BrowseName.IsNull) { - m_sourceName = PropertyState.With(this, source.BrowseName.Name!); // QualifiedName.IsNull is false here, so Name is non-null + m_sourceName = WithTimestamp(PropertyState.With(this, source.BrowseName.Name!), time); // QualifiedName.IsNull is false here, so Name is non-null } } - m_time = PropertyState.With(this, DateTimeUtc.Now); - m_receiveTime = PropertyState.With(this, DateTimeUtc.Now); - m_severity = PropertyState.With(this, (ushort)severity); - m_message = PropertyState.With(this, message); + m_time = WithTimestamp(PropertyState.With(this, time), time); + DateTimeUtc receiveTime = DateTimeUtc.Now; + m_receiveTime = WithTimestamp(PropertyState.With(this, receiveTime), receiveTime); + m_severity = WithTimestamp(PropertyState.With(this, (ushort)severity), time); + m_message = WithTimestamp(PropertyState.With(this, message), time); + } + + /// + /// Sets the source timestamp of an event field. + /// + /// The type of the field. + /// The field. + /// The source timestamp. + private static T WithTimestamp(T field, DateTimeUtc timestamp) + where T : BaseVariableState + { + field.Timestamp = timestamp; + return field; } } diff --git a/src/Opc.Ua.Core/Stack/Types/FilterEvaluator.cs b/src/Opc.Ua.Core/Stack/Types/FilterEvaluator.cs index 366bfedbb2..1724364bf0 100644 --- a/src/Opc.Ua.Core/Stack/Types/FilterEvaluator.cs +++ b/src/Opc.Ua.Core/Stack/Types/FilterEvaluator.cs @@ -50,7 +50,6 @@ namespace Opc.Ua m_filter = filter; m_context = context; m_target = target; - m_logger = context.Telemetry.CreateLogger(); } /// @@ -977,7 +976,7 @@ namespace Opc.Ua ArgumentException or NullReferenceException) { - m_logger.ConversionFailed(ex, targetType); + Logger.ConversionFailed(ex, targetType); return default; } } @@ -1131,7 +1130,7 @@ namespace Opc.Ua ex is not OutOfMemoryException and not StackOverflowException and not AccessViolationException and not OperationCanceledException) { - m_logger.TargetEvaluationFailed(ex); + Logger.TargetEvaluationFailed(ex); return false; } finally @@ -1140,10 +1139,16 @@ namespace Opc.Ua } } + /// + /// The logger, created when a failure is logged: a filter is evaluated for every + /// event and monitored item, and creating a logger each time is not free. + /// + private ILogger Logger => m_logger ??= m_context.Telemetry.CreateLogger(); + private readonly ContentFilter m_filter; private readonly IFilterContext m_context; private readonly IFilterTarget m_target; - private readonly ILogger m_logger; + private ILogger? m_logger; private Variant[]? m_results; private bool[]? m_evaluated; private Variant[]? m_leftValues; @@ -1174,7 +1179,12 @@ namespace Opc.Ua IFilterContext context, IFilterTarget target) { - // check if nothing to do. + // check if nothing to do: an empty filter passes every target. + if (filter.Elements.Count == 0) + { + return true; + } + var evaluator = new FilterEvaluator(filter, context, target); return evaluator.Result; } diff --git a/src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs b/src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs index 7b39fd2d06..4211966306 100644 --- a/src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs +++ b/src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs @@ -59,6 +59,10 @@ namespace Opc.Ua.Redundancy.Server m_sink = sink; } + /// + /// The queue changes its events only through the base members. + internal override bool IndexesRecentEvents => true; + /// public override void Enqueue(EventFieldList value) { @@ -167,7 +171,7 @@ namespace Opc.Ua.Redundancy.Server return new EventQueueSnapshot { QueueSize = QueueSize, - Events = [.. m_events] + Events = CopyQueuedEvents() }; } diff --git a/src/Opc.Ua.Server/Subscription/MonitoredItem/MonitoredItem.cs b/src/Opc.Ua.Server/Subscription/MonitoredItem/MonitoredItem.cs index 56e96a9a3a..29d7ce6bff 100644 --- a/src/Opc.Ua.Server/Subscription/MonitoredItem/MonitoredItem.cs +++ b/src/Opc.Ua.Server/Subscription/MonitoredItem/MonitoredItem.cs @@ -1638,20 +1638,19 @@ namespace Opc.Ua.Server /// Fetches the event fields from the event. /// private EventFieldList GetEventFields( - IFilterContext context, + FilterContext context, EventFilter filter, IFilterTarget instance) { // fetch the event fields. bool[] rejected = GetRejectedSelectClauses(context, filter); - var eventFieldValues = new List(rejected.Length); + var eventFieldValues = new Variant[rejected.Length]; for (int ii = 0; ii < rejected.Length; ii++) { // a select clause rejected in the EventFilterResult returns a null field // (Part 4 ยง7.22.3). if (rejected[ii]) { - eventFieldValues.Add(Variant.Null); continue; } @@ -1665,22 +1664,16 @@ namespace Opc.Ua.Server clause.AttributeId, clause.ParsedIndexRange); - // add the value to the list of event fields. + // a missing value stays a null field. if (!value.IsNull) { // translate any localized text (scalar or array). if (value.TypeInfo.BuiltInType == BuiltInType.LocalizedText) { - value = m_server.ResourceManager.TranslateValue(Session?.PreferredLocales!, value); + value = m_server.ResourceManager.TranslateValue(context.PreferredLocales, value); } - // add value. - eventFieldValues.Add(value); - } - // add a dummy entry for missing values. - else - { - eventFieldValues.Add(Variant.Null); + eventFieldValues[ii] = value; } } var result = (EventFieldList)EventFieldListActivator.Instance.CreateInstance(); @@ -1690,6 +1683,45 @@ namespace Opc.Ua.Server return result; } + /// + /// Returns the context to evaluate the event filter with. The context only carries + /// the server tables and the preferred locales of the session, so it is created again + /// only when the session replaces its locales (or the item moves to another session). + /// + private FilterContext GetEventFilterContext() + { + string[]? preferredLocales = Session?.PreferredLocales; + EventFilterContext? cached = m_eventFilterContext; + if (cached == null || !ReferenceEquals(cached.PreferredLocales, preferredLocales)) + { + cached = new EventFilterContext( + preferredLocales, + new FilterContext( + m_server.NamespaceUris, + m_server.TypeTree, + preferredLocales!, + m_server.Telemetry)); + m_eventFilterContext = cached; + } + return cached.Context; + } + + /// + /// The event filter context and the locales it was created for. + /// + private sealed class EventFilterContext + { + public EventFilterContext(string[]? preferredLocales, FilterContext context) + { + PreferredLocales = preferredLocales; + Context = context; + } + + public string[]? PreferredLocales { get; } + + public FilterContext Context { get; } + } + /// /// Adds an event to the queue. /// @@ -1730,12 +1762,7 @@ namespace Opc.Ua.Server return; } - // construct the context to use for the event filter. - var context = new FilterContext( - m_server.NamespaceUris, - m_server.TypeTree, - Session?.PreferredLocales!, - m_server.Telemetry); + FilterContext context = GetEventFilterContext(); // event filter must be specified. if (FilterToUse is not EventFilter filter) @@ -2971,6 +2998,7 @@ namespace Opc.Ua.Server private bool m_resendData; private HashSet? m_filteredRetainConditionIds; private RejectedSelectClauses? m_rejectedSelectClauses; + private EventFilterContext? m_eventFilterContext; private bool m_isDetached; private bool m_isDeleted; private bool m_countedAsRecoverable; diff --git a/src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs b/src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs index ce99eca294..4853b206ed 100644 --- a/src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs +++ b/src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs @@ -29,6 +29,7 @@ using System; using System.Collections.Generic; +using System.Runtime.CompilerServices; using Microsoft.Extensions.Logging; namespace Opc.Ua.Server @@ -40,6 +41,12 @@ namespace Opc.Ua.Server { private const uint kMaxNoOfEntriesCheckedForDuplicateEvents = 1000; + /// + /// The number of dequeued slots at the front of from which the + /// list is compacted once they are half of it. + /// + private const int kMinDequeuedSlotsToCompact = 32; + /// /// Creates an empty queue. /// @@ -68,16 +75,32 @@ namespace Opc.Ua.Server public uint QueueSize { get; protected set; } /// - public int ItemsInQueue => m_events.Count; + public int ItemsInQueue => m_events.Count - m_head; /// public virtual bool Dequeue(out EventFieldList value) { value = null!; - if (m_events.Count != 0) + if (m_head < m_events.Count) { - value = m_events[0]; - m_events.RemoveAt(0); + // advance the head instead of shifting the list: removing the first entry + // moved every queued event for each dequeued one. + value = m_events[m_head]; + if (IndexesRecentEvents && ItemsInQueue <= kMaxNoOfEntriesCheckedForDuplicateEvents) + { + RemoveRecentHandle(value); + } + m_events[m_head++] = null!; + if (m_head == m_events.Count) + { + m_events.Clear(); + m_head = 0; + } + else if (m_head >= kMinDequeuedSlotsToCompact && m_head * 2 >= m_events.Count) + { + // moves fewer events than were dequeued since the last compaction. + CompactEvents(); + } return true; } return false; @@ -108,28 +131,46 @@ namespace Opc.Ua.Server } //Discard oldest - if (m_events.Count == QueueSize) + if (ItemsInQueue == QueueSize) { Dequeue(out EventFieldList _); } m_events.Add(value); + + if (IndexesRecentEvents) + { + AddRecentHandle(value); + if (ItemsInQueue > kMaxNoOfEntriesCheckedForDuplicateEvents) + { + // the entry that fell out of the newest entries is no longer checked. + RemoveRecentHandle( + m_events[m_events.Count - 1 - (int)kMaxNoOfEntriesCheckedForDuplicateEvents]); + } + } } /// public bool IsEventContainedInQueue(IFilterTarget instance) { // a duplicate is the same instance reported again through another notifier - // path, so it sits among the most recently queued events: scan from the newest. + // path, so it sits among the most recently queued events. The handles of the + // newest entries are counted in an index, so the check does not visit every + // queued event for each event that is not a duplicate. + if (IndexesRecentEvents) + { + return instance != null && m_recentHandles != null && m_recentHandles.ContainsKey(instance); + } + + // a derived queue may change m_events directly, which the index cannot follow. int lowestIndex = - m_events.Count > kMaxNoOfEntriesCheckedForDuplicateEvents + ItemsInQueue > kMaxNoOfEntriesCheckedForDuplicateEvents ? m_events.Count - (int)kMaxNoOfEntriesCheckedForDuplicateEvents - : 0; + : m_head; for (int i = m_events.Count - 1; i >= lowestIndex; i--) { - if (m_events[i] is EventFieldList processedEvent && - IsEventRaisedFor(processedEvent, instance)) + if (IsEventRaisedFor(m_events[i], instance)) { return true; } @@ -137,6 +178,13 @@ namespace Opc.Ua.Server return false; } + /// + /// True when the duplicate check uses the index of the newest queued events. Only + /// queues that change solely through the members of this class + /// may use it; any other derived queue scans the newest entries instead. + /// + internal virtual bool IndexesRecentEvents => GetType() == typeof(EventMonitoredItemQueue); + /// /// Returns true when the queued event was raised for , for /// the duplicate check of @@ -161,6 +209,7 @@ namespace Opc.Ua.Server public virtual void SetQueueSize(uint queueSize, bool discardOldest) { QueueSize = queueSize; + CompactEvents(); if (m_events.Count > QueueSize) { @@ -173,13 +222,128 @@ namespace Opc.Ua.Server m_events.RemoveRange((int)queueSize, m_events.Count - (int)queueSize); } } + + if (IndexesRecentEvents) + { + RebuildRecentHandles(); + } } /// - /// the contained in the queue + /// Copies the queued events, ordered oldest to newest. + /// + internal EventFieldList[] CopyQueuedEvents() + { + var events = new EventFieldList[ItemsInQueue]; + m_events.CopyTo(m_head, events, 0, events.Length); + return events; + } + + /// + /// Removes the dequeued slots at the front of . + /// + private void CompactEvents() + { + if (m_head > 0) + { + m_events.RemoveRange(0, m_head); + m_head = 0; + } + } + + /// + /// Counts the handles of the newest queued events again after the queue was truncated, + /// and drops the handles of the events that left the queue. + /// + private void RebuildRecentHandles() + { + m_recentHandles?.Clear(); + int count = m_events.Count; + int lowestIndex = Math.Max(m_head, count - (int)kMaxNoOfEntriesCheckedForDuplicateEvents); + for (int i = lowestIndex; i < count; i++) + { + AddRecentHandle(m_events[i]); + } + } + + /// + /// Counts the event instance that was raised for. + /// + private void AddRecentHandle(EventFieldList queuedEvent) + { + object? key = GetRaisedFor(queuedEvent); + if (key != null) + { + m_recentHandles ??= new Dictionary(ReferenceComparer.Instance); + m_recentHandles.TryGetValue(key, out int count); + m_recentHandles[key] = count + 1; + } + } + + /// + /// Uncounts the event instance that was raised for. + /// + private void RemoveRecentHandle(EventFieldList queuedEvent) + { + object? key = GetRaisedFor(queuedEvent); + if (key != null && m_recentHandles != null && m_recentHandles.TryGetValue(key, out int count)) + { + if (count > 1) + { + m_recentHandles[key] = count - 1; + } + else + { + m_recentHandles.Remove(key); + } + } + } + + /// + /// The event instance a queued event was raised for, see . + /// + private static object? GetRaisedFor(EventFieldList queuedEvent) + { + object? handle = queuedEvent?.Handle; + return handle is FilteredRetainTarget wrapper ? wrapper.Target : handle; + } + + /// + /// Compares event instances by reference. + /// + private sealed class ReferenceComparer : IEqualityComparer + { + public static readonly ReferenceComparer Instance = new(); + + public new bool Equals(object? x, object? y) + { + return ReferenceEquals(x, y); + } + + public int GetHashCode(object obj) + { + return RuntimeHelpers.GetHashCode(obj); + } + } + + /// + /// The events in the queue, oldest first. A dequeued event leaves a null slot at the + /// front of the list until the list is compacted, so the queued events are the last + /// entries. /// protected List m_events; + /// + /// The index of the oldest queued event in . + /// + private int m_head; + + /// + /// How many times each event instance is the handle of one of the newest + /// queued events. + /// + private Dictionary? m_recentHandles; + /// /// The logger to use here and in derived classes /// diff --git a/tests/Opc.Ua.Core.Tests/Stack/State/NodeStateTests.cs b/tests/Opc.Ua.Core.Tests/Stack/State/NodeStateTests.cs index 08d53f5b6a..26bba2f914 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/State/NodeStateTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/State/NodeStateTests.cs @@ -395,6 +395,41 @@ namespace Opc.Ua.Core.Tests.Stack.State Assert.That(clone.Parent, Is.SameAs(parent)); } + /// + /// Verify that the fields of an initialized event carry the event time as their + /// source timestamp, so reading them does not stamp them with the current time. + /// + [Test] + public void InitializedEventFieldsCarryTheEventTime() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + var context = new SystemContext(telemetry) { NamespaceUris = new NamespaceTable() }; + var source = new BaseObjectState(null) + { + NodeId = new NodeId(1000), + BrowseName = QualifiedName.From("Source") + }; + var eventState = new BaseEventState(null); + eventState.Initialize(context, source, EventSeverity.Medium, new LocalizedText("message")); + + DateTimeUtc time = eventState.Time!.Value; + Assert.That(time, Is.Not.EqualTo(DateTimeUtc.MinValue)); + Assert.That(eventState.EventId!.Timestamp, Is.EqualTo(time)); + Assert.That(eventState.EventType!.Timestamp, Is.EqualTo(time)); + Assert.That(eventState.SourceNode!.Timestamp, Is.EqualTo(time)); + Assert.That(eventState.SourceName!.Timestamp, Is.EqualTo(time)); + Assert.That(eventState.Time.Timestamp, Is.EqualTo(time)); + Assert.That(eventState.Severity!.Timestamp, Is.EqualTo(time)); + Assert.That(eventState.Message!.Timestamp, Is.EqualTo(time)); + Assert.That(eventState.ReceiveTime!.Timestamp, Is.EqualTo(eventState.ReceiveTime.Value)); + + var dataValue = new DataValue(); + ServiceResult result = eventState.Message.ReadAttribute( + context, Attributes.Value, default, default, ref dataValue); + Assert.That(ServiceResult.IsGood(result), Is.True); + Assert.That(dataValue.SourceTimestamp, Is.EqualTo(time)); + } + /// /// Verify that Clone works correctly for BaseEventState with null parent. /// diff --git a/tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs b/tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs index a625677afc..f6bb7c8949 100644 --- a/tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs +++ b/tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs @@ -27,8 +27,10 @@ * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ +using System; using System.Collections.Generic; using System.Linq; +using System.Runtime.CompilerServices; using Moq; using NUnit.Framework; using Opc.Ua.Tests; @@ -72,6 +74,205 @@ namespace Opc.Ua.Server.Tests Is.False); } + [Test] + public void EventQueueDuplicateCheckFollowsTheNewestEntries() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + using var queue = new EventMonitoredItemQueue(false, 1, telemetry); + queue.SetQueueSize(2000, true); + var handles = new List(); + for (int i = 0; i < 1200; i++) + { + var handle = new AuditSessionEventState(null); + handles.Add(handle); + // a trailing filtered retain event is queued with a wrapper of its instance. + object queuedHandle = i % 3 == 0 ? new FilteredRetainTarget(handle) : handle; + queue.Enqueue(new EventFieldList { EventFields = [new Variant(i)], Handle = queuedHandle }); + } + + // only the newest 1000 entries are checked. + Assert.That(queue.IsEventContainedInQueue(handles[199]), Is.False); + Assert.That(queue.IsEventContainedInQueue(handles[200]), Is.True); + Assert.That(queue.IsEventContainedInQueue(handles[201]), Is.True); + Assert.That(queue.IsEventContainedInQueue(handles[1199]), Is.True); + + // an instance queued twice stays a duplicate until both entries are gone. + queue.Enqueue(new EventFieldList { EventFields = [new Variant(-1)], Handle = handles[1199] }); + for (int i = 0; i < 1000; i++) + { + Assert.That(queue.Dequeue(out _), Is.True); + } + Assert.That(queue.ItemsInQueue, Is.EqualTo(201)); + Assert.That(queue.IsEventContainedInQueue(handles[999]), Is.False); + Assert.That(queue.IsEventContainedInQueue(handles[1000]), Is.True); + Assert.That(queue.Dequeue(out EventFieldList last), Is.True); + Assert.That(last.EventFields[0].GetInt32(), Is.EqualTo(1000)); + for (int i = 0; i < 199; i++) + { + Assert.That(queue.Dequeue(out _), Is.True); + } + Assert.That(queue.IsEventContainedInQueue(handles[1199]), Is.True); + Assert.That(queue.Dequeue(out _), Is.True); + Assert.That(queue.ItemsInQueue, Is.Zero); + Assert.That(queue.IsEventContainedInQueue(handles[1199]), Is.False); + + // a shrink counts the remaining entries again. + for (int i = 0; i < 10; i++) + { + queue.Enqueue(new EventFieldList { EventFields = [new Variant(i)], Handle = handles[i] }); + } + queue.SetQueueSize(4, false); + Assert.That(queue.IsEventContainedInQueue(handles[3]), Is.True); + Assert.That(queue.IsEventContainedInQueue(handles[4]), Is.False); + queue.Enqueue(new EventFieldList { EventFields = [new Variant(20)], Handle = handles[20] }); + Assert.That(queue.IsEventContainedInQueue(handles[0]), Is.False); + Assert.That(queue.IsEventContainedInQueue(handles[20]), Is.True); + } + + [Test] + public void EventQueueDuplicateCheckSeesEntriesAddedByDerivedQueue() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + using var queue = new DirectlyFilledEventQueue(telemetry); + queue.SetQueueSize(10, true); + var queued = new AuditSessionEventState(null); + queue.Enqueue(new EventFieldList { EventFields = [new Variant(1)], Handle = queued }); + + var added = new AuditSessionEventState(null); + queue.AddDirectly(new EventFieldList { EventFields = [new Variant(2)], Handle = added }); + Assert.That(queue.IsEventContainedInQueue(added), Is.True); + Assert.That(queue.IsEventContainedInQueue(queued), Is.True); + + // a queue changed directly stays consistent through further queue operations. + queue.AddDirectly(new EventFieldList { EventFields = [new Variant(3)], Handle = added }); + queue.Enqueue(new EventFieldList { EventFields = [new Variant(4)], Handle = queued }); + Assert.That(queue.Dequeue(out _), Is.True); + Assert.That(queue.Dequeue(out _), Is.True); + Assert.That(queue.IsEventContainedInQueue(added), Is.True); + Assert.That(queue.IsEventContainedInQueue(queued), Is.True); + Assert.That(queue.Dequeue(out _), Is.True); + Assert.That(queue.IsEventContainedInQueue(added), Is.False); + + // an entry replaced in place leaves the count unchanged. + var replacement = new AuditSessionEventState(null); + queue.ReplaceNewest(new EventFieldList { EventFields = [new Variant(5)], Handle = replacement }); + Assert.That(queue.IsEventContainedInQueue(replacement), Is.True); + Assert.That(queue.IsEventContainedInQueue(queued), Is.False); + } + + [Test] + public void EventQueueReleasesEventsAfterResizeAndDrain() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + using var queue = new EventMonitoredItemQueue(false, 1, telemetry); + queue.SetQueueSize(10, true); + WeakReference released = EnqueueUnreferencedEvent(queue); + + queue.SetQueueSize(10, true); + DequeueAndDrop(queue); + Assert.That(queue.ItemsInQueue, Is.Zero); + + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + Assert.That(released.IsAlive, Is.False); + } + + [MethodImpl(MethodImplOptions.NoInlining)] + private static WeakReference EnqueueUnreferencedEvent(EventMonitoredItemQueue queue) + { + var handle = new AuditSessionEventState(null); + queue.Enqueue(new EventFieldList { EventFields = [new Variant(1)], Handle = handle }); + Assert.That(queue.IsEventContainedInQueue(handle), Is.True); + return new WeakReference(handle); + } + + [MethodImpl(MethodImplOptions.NoInlining)] + private static void DequeueAndDrop(EventMonitoredItemQueue queue) + { + Assert.That(queue.Dequeue(out _), Is.True); + } + + private sealed class DirectlyFilledEventQueue : EventMonitoredItemQueue + { + public DirectlyFilledEventQueue(ITelemetryContext telemetry) + : base(false, 1, telemetry) + { + } + + public void AddDirectly(EventFieldList value) + { + m_events.Add(value); + } + + public void ReplaceNewest(EventFieldList value) + { + m_events[m_events.Count - 1] = value; + } + } + + [TestCase(true)] + [TestCase(false)] + public void EventQueueKeepsOrderAcrossDequeueCompactionAndShrink(bool discardOldest) + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + using var queue = new EventMonitoredItemQueue(false, 1, telemetry); + queue.SetQueueSize(100, true); + var handles = new List(); + + void Enqueue(int value) + { + var handle = new AuditSessionEventState(null); + handles.Add(handle); + queue.Enqueue(new EventFieldList { EventFields = [new Variant(value)], Handle = handle }); + } + + int Dequeue() + { + Assert.That(queue.Dequeue(out EventFieldList fields), Is.True); + return fields.EventFields[0].GetInt32(); + } + + // 150 events into a queue of 100 discard the oldest 50. + for (int i = 0; i < 150; i++) + { + Enqueue(i); + } + Assert.That(queue.ItemsInQueue, Is.EqualTo(100)); + Assert.That(queue.IsEventContainedInQueue(handles[49]), Is.False); + Assert.That(queue.IsEventContainedInQueue(handles[50]), Is.True); + + // dequeue across the compaction threshold, interleaved with new events. + int expected = 50; + for (int i = 0; i < 70; i++) + { + Assert.That(Dequeue(), Is.EqualTo(expected++)); + if (i % 2 == 0) + { + Enqueue(150 + (i / 2)); + } + } + Assert.That(queue.ItemsInQueue, Is.EqualTo(65)); + Assert.That(queue.IsEventContainedInQueue(handles[119]), Is.False); + Assert.That(queue.IsEventContainedInQueue(handles[120]), Is.True); + Assert.That(queue.IsEventContainedInQueue(handles[handles.Count - 1]), Is.True); + + // shrinking keeps the oldest or newest of the queued events. + queue.SetQueueSize(10, discardOldest); + Assert.That(queue.ItemsInQueue, Is.EqualTo(10)); + int first = discardOldest ? 175 : 120; + for (int i = 0; i < 10; i++) + { + Assert.That(Dequeue(), Is.EqualTo(first + i)); + } + Assert.That(queue.Dequeue(out _), Is.False); + Assert.That(queue.ItemsInQueue, Is.Zero); + Assert.That(queue.IsEventContainedInQueue(handles[handles.Count - 1]), Is.False); + + Enqueue(1000); + Assert.That(Dequeue(), Is.EqualTo(1000)); + } + [Test] public void EventOverflowKeepsPositionWhenNotDiscardingOldest() {