From be80805e503f7d314cb29dff26fa040e5890be23 Mon Sep 17 00:00:00 2001 From: Azat Tazayan Date: Sat, 25 Apr 2026 22:55:42 -0500 Subject: [PATCH 1/3] Add EventStream with backlog and comprehensive tests Introduced EventStream for .NET 8+, providing a strongly-typed pub/sub event stream with a fixed-size rolling backlog using a ring buffer. Added multiple Subscribe overloads supporting thread options, strong/weak references, filtering, and backlog replay. Overrode publish and subscription logic for backlog management and thread safety. Included extensive unit tests covering all major behaviors and edge cases. --- src/Prism.Events/EventStream.cs | 461 +++++++++++ .../Events/EventStreamFixture.cs | 726 ++++++++++++++++++ 2 files changed, 1187 insertions(+) create mode 100644 src/Prism.Events/EventStream.cs create mode 100644 tests/Prism.Core.Tests/Events/EventStreamFixture.cs diff --git a/src/Prism.Events/EventStream.cs b/src/Prism.Events/EventStream.cs new file mode 100644 index 000000000..fb96aeacc --- /dev/null +++ b/src/Prism.Events/EventStream.cs @@ -0,0 +1,461 @@ +#if NET8_0_OR_GREATER +using System.Buffers; +using Prism.Events.Properties; + +namespace Prism.Events; + +/// +/// A strongly-typed pub/sub event stream that extends with a fixed-size +/// rolling backlog. When a new subscriber calls Subscribe, it immediately receives all +/// events currently held in the backlog via the supplied backlogAction callback, then +/// continues to receive future published events through the main action callback. +/// +/// +/// +/// The backlog is implemented as an in-memory ring buffer whose capacity is determined by +/// . When the buffer is full, the oldest entry is evicted to make room +/// for the newest one. The default capacity is 10. +/// +/// +/// This class is only available on .NET 8 and later (NET8_0_OR_GREATER) because it relies +/// on spans and other modern runtime features. +/// +/// +/// Thread safety: all mutations to the subscription list and the ring buffer are performed inside +/// a on . Subscriber callbacks are +/// dispatched outside the lock to avoid deadlocks. +/// +/// +/// The type of the event payload. +public class EventStream : EventBase +{ + private const int DefaultBacklogSize = 10; + + private Backlog recentEvents; + + /// + /// Initializes a new instance of with the default + /// backlog size of 10. + /// + public EventStream() + { + recentEvents = new Backlog((int)BacklogSize()); + } + + /// + /// Subscribes to the event stream on the publisher's thread without a filter. + /// Any events currently held in the backlog are immediately delivered to + /// . + /// + /// The callback invoked for each future published event. + /// + /// The callback invoked once for each event currently in the backlog at the time of + /// subscription. May be to skip backlog replay. + /// + /// + /// A that can be used to unsubscribe from the event stream. + /// + public SubscriptionToken Subscribe(Action action, Action backlogAction) + { + return Subscribe(action, backlogAction, ThreadOption.PublisherThread, false, null); + } + + /// + /// Subscribes to the event stream on the specified thread without a filter. + /// Any events currently held in the backlog are immediately delivered to + /// . + /// + /// The callback invoked for each future published event. + /// + /// The callback invoked once for each event currently in the backlog at the time of + /// subscription. May be to skip backlog replay. + /// + /// + /// Specifies the thread on which is invoked. + /// See for available options. + /// + /// + /// A that can be used to unsubscribe from the event stream. + /// + public SubscriptionToken Subscribe(Action action, Action backlogAction, ThreadOption threadOption) + { + return Subscribe(action, backlogAction, threadOption, false, null); + } + + /// + /// Subscribes to the event stream on the specified thread, optionally keeping a strong + /// reference to the subscriber, without a filter. + /// Any events currently held in the backlog are immediately delivered to + /// . + /// + /// The callback invoked for each future published event. + /// + /// The callback invoked once for each event currently in the backlog at the time of + /// subscription. May be to skip backlog replay. + /// + /// + /// Specifies the thread on which is invoked. + /// + /// + /// When , the event stream holds a strong reference to + /// , preventing it from being garbage-collected. When + /// a weak reference is used and the subscription is pruned + /// automatically once the subscriber is collected. + /// + /// + /// A that can be used to unsubscribe from the event stream. + /// + public SubscriptionToken Subscribe(Action action, Action backlogAction, ThreadOption threadOption, bool keepSubscriberReferenceAlive) + { + return Subscribe(action, backlogAction, threadOption, keepSubscriberReferenceAlive, null); + } + + /// + /// Subscribes to the event stream with full control over thread marshalling, reference + /// lifetime, and payload filtering. + /// Any events currently held in the backlog that satisfy are + /// immediately delivered to . + /// + /// The callback invoked for each future published event. + /// + /// The callback invoked once for each matching event currently in the backlog at the time + /// of subscription. May be to skip backlog replay. + /// + /// + /// Specifies the thread on which is invoked. + /// + /// + /// When , the event stream holds a strong reference to + /// . When a weak reference is used. + /// + /// + /// A predicate applied to each incoming payload. Only payloads for which the predicate + /// returns are forwarded to . Pass + /// to receive all payloads. + /// + /// + /// A that can be used to unsubscribe from the event stream. + /// + /// + /// Thrown when is and + /// has not been set. + /// + public SubscriptionToken Subscribe(Action action, Action backlogAction, ThreadOption threadOption, bool keepSubscriberReferenceAlive, Predicate filter) + { + IDelegateReference actionReference = new DelegateReference(action, keepSubscriberReferenceAlive); + IDelegateReference filterReference; + if (filter != null) + { + filterReference = new DelegateReference(filter, keepSubscriberReferenceAlive); + } + else + { + filterReference = new DelegateReference(new Predicate(delegate { return true; }), true); + } + + EventSubscription subscription; + + switch (threadOption) + { + case ThreadOption.PublisherThread: + subscription = new EventSubscription(actionReference, filterReference); + break; + case ThreadOption.BackgroundThread: + subscription = new BackgroundEventSubscription(actionReference, filterReference); + break; + case ThreadOption.UIThread: + if (SynchronizationContext == null) throw new InvalidOperationException(Resources.EventAggregatorNotConstructedOnUIThread); + subscription = new DispatcherEventSubscription(actionReference, filterReference, SynchronizationContext); + break; + default: + subscription = new EventSubscription(actionReference, filterReference); + break; + } + + return Subscribe(subscription, backlogAction); + } + + /// + /// Registers an already-constructed , assigns it a + /// new , and replays the current backlog via + /// while holding the subscription lock. + /// + /// The subscription to register. + /// + /// The callback invoked once for each event currently in the backlog. May be + /// to skip backlog replay. + /// + /// The assigned to the subscription. + /// + /// Thrown when is . + /// + private SubscriptionToken Subscribe(EventSubscription eventSubscription, Action backlogAction) + { + if (eventSubscription == null) + throw new ArgumentNullException(nameof(eventSubscription)); + + eventSubscription.SubscriptionToken = new SubscriptionToken(Unsubscribe); + + TPayload[] backlog = null; + + try + { + int backlogCount = 0; + + lock (Subscriptions) + { + Subscriptions.Add(eventSubscription); + var ring = recentEvents.CurrentState(out uint readPosition, out uint writePosition); + backlog = ArrayPool.Shared.Rent(ring.Length); + int index = 0; + TPayload item = default; + while (index < ring.Length && + Backlog.TryRead(ref ring, ref readPosition, writePosition, out item)) + { + backlog[index] = item; + index++; + } + + backlogCount = index; + } + + if (backlogAction != null && backlogCount > 0) + { + foreach (var item in backlog.AsSpan(0, backlogCount)) + { + if(eventSubscription.Filter(item)) + { + backlogAction(item); + } + } + } + } + finally + { + if (backlog != null) + ArrayPool.Shared.Return(backlog, clearArray: true); + } + + return eventSubscription.SubscriptionToken; + } + + /// + /// Removes the first subscription whose action delegate matches + /// from the subscribers' list. + /// + /// + /// The delegate that was passed to Subscribe. + /// + public virtual void Unsubscribe(Action subscriber) + { + lock (Subscriptions) + { + IEventSubscription eventSubscription = Subscriptions.Cast>().FirstOrDefault(evt => evt.Action == subscriber); + if (eventSubscription != null) + { + Subscriptions.Remove(eventSubscription); + } + } + } + + /// + /// Publishes to all active subscribers and appends it to the + /// rolling backlog so that future subscribers can replay it. + /// + /// The event payload to publish. + public virtual void Publish(TPayload payload) + { + InternalPublish(payload); + } + + /// + /// Returns if there is a subscriber matching . + /// + /// The used when subscribing to the event. + /// if there is an that matches; otherwise . + public virtual bool Contains(Action subscriber) + { + IEventSubscription eventSubscription; + lock (Subscriptions) + { + eventSubscription = Subscriptions.Cast>().FirstOrDefault(evt => evt.Action == subscriber); + } + return eventSubscription != null; + } + + /// + /// Core publish implementation. Writes the payload to the ring-buffer backlog, collects + /// the execution strategies of all live subscriptions (pruning dead weak-reference + /// subscriptions in the process), then dispatches outside the lock to avoid deadlocks. + /// + /// + /// The boxed arguments array; element [0] must be of type . + /// + override protected void InternalPublish(params object[] arguments) + { + List> strategies; + + lock (Subscriptions) + { + recentEvents.Write((TPayload)arguments[0]); + + strategies = new List>(Subscriptions.Count); + + for (int i = Subscriptions.Count - 1; i >= 0; i--) + { + var subscription = Subscriptions.ElementAt(i); + var strategy = subscription.GetExecutionStrategy(); + if (strategy == null) + Subscriptions.Remove(subscription);// prune dead weak refs + else + strategies.Add(strategy); + } + } + + // Dispatch outside the lock — never invoke callbacks under a lock + foreach (var strategy in strategies) + strategy(arguments); + } + + /// + /// Overrides to route subscriptions through + /// the backlog-aware Subscribe overload. Backlog replay is skipped (null + /// backlog action) when subscribing via the base-class path. + /// + /// The subscription to register. + /// The assigned to the subscription. + override protected SubscriptionToken InternalSubscribe(IEventSubscription eventSubscription) + { + return Subscribe((EventSubscription)eventSubscription, null); + } + + /// + /// Returns the capacity of the rolling backlog ring buffer. + /// Override this method in a derived class to change the number of events retained for + /// late subscribers. The default value is 10. + /// + /// The number of recent events to retain in the backlog. + virtual protected uint BacklogSize() => DefaultBacklogSize; + + /// + /// A fixed-capacity, thread-unsafe ring buffer used to store the most recent + /// items published to the stream. When the buffer is full the + /// oldest item is silently overwritten. Intended for use only inside the + /// lock region. + /// + /// The element type stored in the ring buffer. + private class Backlog + { + private readonly int size; + private readonly T[] ring; + private uint writePosition; + private uint readPosition; + + /// + /// Initializes a new with the specified capacity. + /// + /// Maximum number of items the ring buffer can hold. + public Backlog(int capacity) + { + size = capacity; + ring = new T[capacity]; + writePosition = 0; + readPosition = 0; + } + + /// + /// Writes to the ring buffer. If the buffer is full, the + /// oldest unread item is evicted by advancing the read position before writing. + /// + /// The item to write. + public void Write(T item) + { + bool isFull = writePosition - readPosition >= ring.Length; + + + if (isFull) + { + readPosition++; + } + + ring[writePosition % size] = item; + writePosition++; + } + + /// + /// Attempts to read the next item from the ring buffer, advancing the internal read + /// cursor on success. + /// + /// + /// When this method returns , contains the next item; otherwise + /// the default value of . + /// + /// + /// if an item was read; if the buffer + /// is empty. + /// + public bool TryRead(out T item) + { + var span = new ReadOnlySpan(ring); + return TryRead(ref span, ref readPosition, writePosition, out item); + } + + /// + /// Returns a snapshot of the underlying ring array as a + /// together with the current read and write cursor positions, allowing a caller to + /// iterate the backlog without mutating this instance. + /// + /// + /// Receives the current read cursor. Pass this value to + /// to begin + /// iterating from the oldest available item. + /// + /// + /// Receives the current write cursor. Pass this value to + /// as the stop + /// condition. + /// + /// A over the internal ring array. + public ReadOnlySpan CurrentState(out uint readPosition, out uint writePosition) + { + readPosition = this.readPosition; + writePosition = this.writePosition; + return ring; + } + + /// + /// Stateless overload that reads the next item from an externally supplied ring span + /// using caller-managed cursor values. This allows concurrent snapshot reads without + /// mutating the instance. + /// + /// A over the ring array. + /// + /// The current read cursor. Incremented by one on a successful read. + /// + /// The write cursor used as the upper bound. + /// + /// When this method returns , contains the item at + /// ; otherwise the default value of + /// . + /// + /// + /// if an item was read; if + /// has reached + /// (buffer is empty or fully consumed). + /// + public static bool TryRead(ref ReadOnlySpan ring, ref uint readPosition, uint writePosition, out T item) + { + bool isEmpty = readPosition >= writePosition; + if (isEmpty) + { + item = default(T); + return false; + } + + item = ring[(int)readPosition % ring.Length]; + readPosition++; + return true; + } + } +} +#endif diff --git a/tests/Prism.Core.Tests/Events/EventStreamFixture.cs b/tests/Prism.Core.Tests/Events/EventStreamFixture.cs new file mode 100644 index 000000000..e0d566eb7 --- /dev/null +++ b/tests/Prism.Core.Tests/Events/EventStreamFixture.cs @@ -0,0 +1,726 @@ +#if NET8_0_OR_GREATER +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Prism.Events; +using Xunit; +using static Prism.Tests.Events.PubSubEventFixture; + +namespace Prism.Tests.Events +{ + public class EventStreamFixture + { + // ── Basic subscribe / publish ──────────────────────────────────────────── + + [Fact] + public void CanSubscribeAndPublishEvent() + { + var stream = new TestableEventStream(); + bool received = false; + stream.Subscribe(_ => received = true, null); + + stream.Publish("hello"); + + Assert.True(received); + } + + [Fact] + public void CanHaveMultipleSubscribersAndPublish() + { + var stream = new TestableEventStream(); + var results = new List(); + stream.Subscribe(v => results.Add("a:" + v), null); + stream.Subscribe(v => results.Add("b:" + v), null); + + stream.Publish("x"); + + Assert.Contains("a:x", results); + Assert.Contains("b:x", results); + } + + // ── Backlog replay on subscribe ────────────────────────────────────────── + + [Fact] + public void NewSubscriberReceivesBacklogOnSubscribe() + { + var stream = new TestableEventStream(); + stream.Publish("first"); + stream.Publish("second"); + + var backlog = new List(); + stream.Subscribe(_ => { }, v => backlog.Add(v)); + + Assert.Equal(new[] { "first", "second" }, backlog); + } + + [Fact] + public void BacklogActionIsNullSafeAndDoesNotThrow() + { + var stream = new TestableEventStream(); + stream.Publish("item"); + + // Should not throw when backlogAction is null + var ex = Record.Exception(() => stream.Subscribe(_ => { }, null)); + Assert.Null(ex); + } + + [Fact] + public void BacklogIsEmptyForFreshStream() + { + var stream = new TestableEventStream(); + var backlog = new List(); + stream.Subscribe(_ => { }, v => backlog.Add(v)); + + Assert.Empty(backlog); + } + + [Fact] + public void BacklogDoesNotIncludeEventsPublishedAfterSubscription() + { + var stream = new TestableEventStream(); + stream.Publish("before"); + + var backlog = new List(); + var live = new List(); + stream.Subscribe(v => live.Add(v), v => backlog.Add(v)); + + stream.Publish("after"); + + Assert.Equal(new[] { "before" }, backlog); + Assert.Equal(new[] { "after" }, live); + } + + // ── Rolling ring-buffer eviction ───────────────────────────────────────── + + [Fact] + public void BacklogEvictsOldestWhenCapacityExceeded() + { + // Default capacity is 10; publish 12 items → only last 10 replayed + var stream = new TestableEventStream(); + for (int i = 1; i <= 12; i++) + stream.Publish(i); + + var backlog = new List(); + stream.Subscribe(_ => { }, v => backlog.Add(v)); + + Assert.Equal(10, backlog.Count); + Assert.Equal(Enumerable.Range(3, 10), backlog); + } + + [Fact] + public void CustomBacklogSizeIsRespected() + { + var stream = new SmallBacklogEventStream(); // capacity = 3 + for (int i = 1; i <= 5; i++) + stream.Publish(i); + + var backlog = new List(); + stream.Subscribe(_ => { }, v => backlog.Add(v)); + + Assert.Equal(3, backlog.Count); + Assert.Equal(new[] { 3, 4, 5 }, backlog); + } + + // ── Filter ─────────────────────────────────────────────────────────────── + + [Fact] + public void FilterPreventsUnmatchedPayloadsFromBeingDelivered() + { + var stream = new TestableEventStream(); + var received = new List(); + stream.Subscribe( + v => received.Add(v), + null, + ThreadOption.PublisherThread, + true, + v => v.StartsWith("keep")); + + stream.Publish("keep-this"); + stream.Publish("drop-this"); + + Assert.Equal(new[] { "keep-this" }, received); + } + + [Fact] + public void FilterIsAppliedToBacklogReplay() + { + var stream = new TestableEventStream(); + stream.Publish("keep-1"); + stream.Publish("drop-1"); + stream.Publish("keep-2"); + + var backlog = new List(); + stream.Subscribe( + _ => { }, + v => backlog.Add(v), + ThreadOption.PublisherThread, + true, + v => v.StartsWith("keep")); + + Assert.Equal(new[] { "keep-1", "keep-2" }, backlog); + } + + // ── Unsubscribe ─────────────────────────────────────────────────────────── + + [Fact] + public void UnsubscribeByDelegateStopsDelivery() + { + var stream = new TestableEventStream(); + var received = new List(); + Action handler = v => received.Add(v); + stream.Subscribe(handler, null, ThreadOption.PublisherThread, true); + + stream.Publish("before"); + stream.Unsubscribe(handler); + stream.Publish("after"); + + Assert.Equal(new[] { "before" }, received); + } + + [Fact] + public void UnsubscribeByTokenStopsDelivery() + { + var stream = new TestableEventStream(); + var received = new List(); + var token = stream.Subscribe(v => received.Add(v), null, ThreadOption.PublisherThread, true); + + stream.Publish("before"); + stream.Unsubscribe(token); + stream.Publish("after"); + + Assert.Equal(new[] { "before" }, received); + } + + [Fact] + public void UnsubscribeNonSubscriberDoesNotThrow() + { + var stream = new TestableEventStream(); + Action stranger = _ => { }; + + var ex = Record.Exception(() => stream.Unsubscribe(stranger)); + Assert.Null(ex); + } + + [Fact] + public void UnsubscribeRemovesOnlyFirstMatchingDelegate() + { + var stream = new TestableEventStream(); + int callCount = 0; + Action handler = _ => callCount++; + stream.Subscribe(handler, null, ThreadOption.PublisherThread, true); + stream.Subscribe(handler, null, ThreadOption.PublisherThread, true); + + stream.Publish("x"); + Assert.Equal(2, callCount); + + callCount = 0; + stream.Unsubscribe(handler); + stream.Publish("x"); + Assert.Equal(1, callCount); + } + + // ── Thread options ──────────────────────────────────────────────────────── + + [Fact] + public void SubscribeOnPublisherThreadCreatesEventSubscription() + { + var stream = new TestableEventStream(); + stream.Subscribe(_ => { }, null, ThreadOption.PublisherThread, true); + + Assert.Equal( + typeof(EventSubscription), + stream.BaseSubscriptions.Single().GetType()); + } + + [Fact] + public void SubscribeOnBackgroundThreadCreatesBackgroundEventSubscription() + { + var stream = new TestableEventStream(); + stream.Subscribe(_ => { }, null, ThreadOption.BackgroundThread, true); + + Assert.Equal( + typeof(BackgroundEventSubscription), + stream.BaseSubscriptions.Single().GetType()); + } + + [Fact] + public void SubscribeOnUIThreadThrowsWhenSynchronizationContextIsNull() + { + var stream = new TestableEventStream(); + + Assert.Throws( + () => stream.Subscribe(_ => { }, null, ThreadOption.UIThread, true)); + } + + [Fact] + public void SubscribeOnUIThreadCreatesDispatcherEventSubscription() + { + var stream = new TestableEventStream(); + stream.SynchronizationContext = new SynchronizationContext(); + + stream.Subscribe(_ => { }, null, ThreadOption.UIThread, true); + + Assert.Equal( + typeof(DispatcherEventSubscription), + stream.BaseSubscriptions.Single().GetType()); + } + + // ── Weak references / GC ───────────────────────────────────────────────── + + [Fact] + public async Task DeadWeakReferenceSubscriptionIsPrunedOnPublish() + { + var stream = new TestableEventStream(); + SubscribeWithoutKeepAlive(stream); + + await Task.Delay(100); + GC.Collect(); + + stream.Publish("trigger"); + + Assert.Empty(stream.BaseSubscriptions); + } + + [Fact] + public void StrongReferenceSubscriberIsNotCollected() + { + var stream = new TestableEventStream(); + var action = new ExternalAction(); + stream.Subscribe(action.ExecuteAction, null, ThreadOption.PublisherThread, true); + + var weakRef = new WeakReference(action); + action = null; + GC.Collect(); + GC.Collect(); + + Assert.True(weakRef.IsAlive); + stream.Publish("test"); + Assert.Equal("test", ((ExternalAction)weakRef.Target).PassedValue); + } + + // ── Subscription token / Contains ──────────────────────────────────────── + + [Fact] + public void ContainsByDelegateReturnsTrueAfterSubscribe() + { + var stream = new TestableEventStream(); + Action handler = _ => { }; + stream.Subscribe(handler, null, ThreadOption.PublisherThread, true); + + Assert.True(stream.Contains(handler)); + } + + [Fact] + public void ContainsByTokenReturnsTrueAfterSubscribeAndFalseAfterUnsubscribe() + { + var stream = new TestableEventStream(); + var token = stream.Subscribe(_ => { }, null, ThreadOption.PublisherThread, true); + + Assert.True(stream.Contains(token)); + + stream.Unsubscribe(token); + Assert.False(stream.Contains(token)); + } + + // ── Concurrent subscribe while publishing ───────────────────────────────── + + [Fact] + public void CanAddSubscriptionWhileEventIsFiring() + { + var stream = new TestableEventStream(); + var lateHandler = new ActionHelper(); + var earlyHandler = new ActionHelper + { + ActionToExecute = () => stream.Subscribe(lateHandler.Action, null, ThreadOption.PublisherThread, true) + }; + + stream.Subscribe(earlyHandler.Action, null, ThreadOption.PublisherThread, true); + Assert.False(stream.Contains(lateHandler.Action)); + + stream.Publish("fire"); + + Assert.True(stream.Contains(lateHandler.Action)); + } + + // ── Helpers ─────────────────────────────────────────────────────────────── + + [MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.NoInlining)] + private static void SubscribeWithoutKeepAlive(TestableEventStream stream) + { + stream.Subscribe(new ExternalAction().ExecuteAction, null); + } + + [Fact] + public void EnsureSubscriptionListIsEmptyAfterPublishingAMessage() + { + var pubSubEvent = new TestableEventStream(); + SubscribeExternalActionWithoutReference(pubSubEvent); + GC.Collect(); + pubSubEvent.Publish("testPayload"); + Assert.True(pubSubEvent.BaseSubscriptions.Count == 0, "Subscriptionlist is not empty"); + } + + [Fact] + public void EnsureSubscriptionListIsNotEmptyWithoutPublishOrSubscribe() + { + var pubSubEvent = new TestableEventStream(); + SubscribeExternalActionWithoutReference(pubSubEvent); + GC.Collect(); + Assert.True(pubSubEvent.BaseSubscriptions.Count == 1, "Subscriptionlist is empty"); + } + + [Fact] + public void EnsureSubscriptionListIsEmptyAfterSubscribeAgainAMessage() + { + var pubSubEvent = new TestableEventStream(); + SubscribeExternalActionWithoutReference(pubSubEvent); + GC.Collect(); + SubscribeExternalActionWithoutReference(pubSubEvent); + pubSubEvent.Prune(); + Assert.True(pubSubEvent.BaseSubscriptions.Count == 1, "Subscriptionlist is empty"); + } + + private static void SubscribeExternalActionWithoutReference(TestableEventStream pubSubEvent) + { + var externalAction = new ExternalAction(); + pubSubEvent.Subscribe(externalAction.ExecuteAction, externalAction.ExecuteActionBacklog); + } + + + [Fact] + public void CanSubscribeAndRaiseEvent() + { + TestableEventStream pubSubEvent = new TestableEventStream(); + bool published = false; + pubSubEvent.Subscribe(delegate { published = true; },null, ThreadOption.PublisherThread, true, delegate { return true; }); + pubSubEvent.Publish(null); + + Assert.True(published); + } + + [Fact] + public void CanSubscribeAndRaiseCustomEvent() + { + var customEvent = new TestableEventStream(); + Payload payload = new Payload(); + var action = new ActionHelper(); + customEvent.Subscribe(action.Action, action.Action); + + customEvent.Publish(payload); + + Assert.Same(action.ActionArg(), payload); + } + + [Fact] + public void CanHaveMultipleSubscribersAndRaiseCustomEvent() + { + var customEvent = new TestableEventStream(); + Payload payload = new Payload(); + var action1 = new ActionHelper(); + var action2 = new ActionHelper(); + customEvent.Subscribe(action1.Action, action1.Action); + customEvent.Subscribe(action2.Action, action2.Action); + + customEvent.Publish(payload); + + Assert.Same(action1.ActionArg(), payload); + Assert.Same(action2.ActionArg(), payload); + } + + + [Fact] + public void SubscribeTakesExecuteDelegateThreadOptionAndFilter() + { + TestableEventStream pubSubEvent = new TestableEventStream(); + var action = new ActionHelper(); + pubSubEvent.Subscribe(action.Action, action.Action); + + pubSubEvent.Publish("test"); + + Assert.Equal("test", action.ActionArg()); + + } + + [Fact] + public void FilterEnablesActionTarget() + { + TestableEventStream pubSubEvent = new TestableEventStream(); + var goodFilter = new MockFilter { FilterReturnValue = true }; + var actionGoodFilter = new ActionHelper(); + var badFilter = new MockFilter { FilterReturnValue = false }; + var actionBadFilter = new ActionHelper(); + pubSubEvent.Subscribe(actionGoodFilter.Action, actionGoodFilter.Action, ThreadOption.PublisherThread, true, goodFilter.FilterString); + pubSubEvent.Subscribe(actionBadFilter.Action, actionBadFilter.Action, ThreadOption.PublisherThread, true, badFilter.FilterString); + + pubSubEvent.Publish("test"); + + Assert.True(actionGoodFilter.ActionCalled); + Assert.False(actionBadFilter.ActionCalled); + + } + + [Fact] + public void FilterEnablesActionTarget_Weak() + { + TestableEventStream pubSubEvent = new TestableEventStream(); + var goodFilter = new MockFilter { FilterReturnValue = true }; + var actionGoodFilter = new ActionHelper(); + var badFilter = new MockFilter { FilterReturnValue = false }; + var actionBadFilter = new ActionHelper(); + pubSubEvent.Subscribe(actionGoodFilter.Action, actionGoodFilter.Action, ThreadOption.PublisherThread, false, goodFilter.FilterString); + pubSubEvent.Subscribe(actionBadFilter.Action, actionBadFilter.Action, ThreadOption.PublisherThread, false, badFilter.FilterString); + + pubSubEvent.Publish("test"); + + Assert.True(actionGoodFilter.ActionCalled); + Assert.False(actionBadFilter.ActionCalled); + + } + + [Fact] + public void SubscribeDefaultsThreadOptionAndNoFilter() + { + TestableEventStream pubSubEvent = new TestableEventStream(); + SynchronizationContext.SetSynchronizationContext(new SynchronizationContext()); + SynchronizationContext calledSyncContext = null; + var myAction = new ActionHelper() + { + ActionToExecute = + () => calledSyncContext = SynchronizationContext.Current + }; + pubSubEvent.Subscribe(myAction.Action, myAction.Action); + + pubSubEvent.Publish("test"); + + Assert.Equal(SynchronizationContext.Current, calledSyncContext); + } + + + [Fact] + public void ShouldUnsubscribeFromPublisherThread() + { + var PubSubEvent = new TestableEventStream(); + + var actionEvent = new ActionHelper(); + PubSubEvent.Subscribe( + actionEvent.Action, + actionEvent.Action, + ThreadOption.PublisherThread); + + Assert.True(PubSubEvent.Contains(actionEvent.Action)); + PubSubEvent.Unsubscribe(actionEvent.Action); + Assert.False(PubSubEvent.Contains(actionEvent.Action)); + } + + [Fact] + public void UnsubscribeShouldNotFailWithNonSubscriber() + { + TestableEventStream pubSubEvent = new TestableEventStream(); + + Action subscriber = delegate { }; + pubSubEvent.Unsubscribe(subscriber); + } + + [Fact] + public void ShouldUnsubscribeFromBackgroundThread() + { + var PubSubEvent = new TestableEventStream(); + + var actionEvent = new ActionHelper(); + PubSubEvent.Subscribe( + actionEvent.Action, + actionEvent.Action, + ThreadOption.BackgroundThread); + + Assert.True(PubSubEvent.Contains(actionEvent.Action)); + PubSubEvent.Unsubscribe(actionEvent.Action); + Assert.False(PubSubEvent.Contains(actionEvent.Action)); + } + + [Fact] + public void ShouldUnsubscribeFromUIThread() + { + var PubSubEvent = new TestableEventStream(); + PubSubEvent.SynchronizationContext = new SynchronizationContext(); + + var actionEvent = new ActionHelper(); + PubSubEvent.Subscribe( + actionEvent.Action, + actionEvent.Action, + ThreadOption.UIThread); + + Assert.True(PubSubEvent.Contains(actionEvent.Action)); + PubSubEvent.Unsubscribe(actionEvent.Action); + Assert.False(PubSubEvent.Contains(actionEvent.Action)); + } + + [Fact] + public void ShouldUnsubscribeASingleDelegate() + { + var PubSubEvent = new TestableEventStream(); + + int callCount = 0; + + var actionEvent = new ActionHelper() { ActionToExecute = () => callCount++ }; + PubSubEvent.Subscribe(actionEvent.Action, actionEvent.Action); + PubSubEvent.Subscribe(actionEvent.Action, actionEvent.Action); + + PubSubEvent.Publish(null); + Assert.Equal(2, callCount); + + callCount = 0; + PubSubEvent.Unsubscribe(actionEvent.Action); + PubSubEvent.Publish(null); + Assert.Equal(1, callCount); + } + + + [Fact] + public async Task ShouldNotExecuteOnGarbageCollectedDelegateReferenceWhenNotKeepAlive() + { + var PubSubEvent = new TestableEventStream(); + + ExternalAction externalAction = new ExternalAction(); + PubSubEvent.Subscribe(externalAction.ExecuteAction, externalAction.ExecuteAction); + + PubSubEvent.Publish("testPayload"); + Assert.Equal("testPayload", externalAction.PassedValue); + + WeakReference actionEventReference = new WeakReference(externalAction); + externalAction = null; + await Task.Delay(100); + GC.Collect(); + Assert.False(actionEventReference.IsAlive); + + PubSubEvent.Publish("testPayload"); + } + + + [Fact] + public async Task ShouldNotExecuteOnGarbageCollectedFilterReferenceWhenNotKeepAlive() + { + var PubSubEvent = new TestableEventStream(); + + bool wasCalled = false; + var actionEvent = new ActionHelper() { ActionToExecute = () => wasCalled = true }; + + ExternalFilter filter = new ExternalFilter(); + PubSubEvent.Subscribe(actionEvent.Action, actionEvent.Action, ThreadOption.PublisherThread, false, filter.AlwaysTrueFilter); + + PubSubEvent.Publish("testPayload"); + Assert.True(wasCalled); + + wasCalled = false; + WeakReference filterReference = new WeakReference(filter); + filter = null; + await Task.Delay(100); + GC.Collect(); + Assert.False(filterReference.IsAlive); + + PubSubEvent.Publish("testPayload"); + Assert.False(wasCalled); + } + + + [Fact] + public void InlineDelegateDeclarationsDoesNotGetCollectedIncorrectlyWithWeakReferences() + { + var PubSubEvent = new TestableEventStream(); + bool published = false; + PubSubEvent.Subscribe(delegate { published = true; }, null, ThreadOption.PublisherThread, false, delegate { return true; }); + GC.Collect(); + PubSubEvent.Publish(null); + + Assert.True(published); + } + + [Fact] + public void ShouldNotGarbageCollectDelegateReferenceWhenUsingKeepAlive() + { + var PubSubEvent = new TestableEventStream(); + + var externalAction = new ExternalAction(); + PubSubEvent.Subscribe(externalAction.ExecuteAction, null, ThreadOption.PublisherThread, true); + + WeakReference actionEventReference = new WeakReference(externalAction); + externalAction = null; + GC.Collect(); + GC.Collect(); + Assert.True(actionEventReference.IsAlive); + + PubSubEvent.Publish("testPayload"); + + Assert.Equal("testPayload", ((ExternalAction)actionEventReference.Target).PassedValue); + } + + [Fact] + public void RegisterReturnsTokenThatCanBeUsedToUnsubscribe() + { + var PubSubEvent = new TestableEventStream(); + var emptyAction = new ActionHelper(); + + var token = PubSubEvent.Subscribe(emptyAction.Action, null); + PubSubEvent.Unsubscribe(token); + + Assert.False(PubSubEvent.Contains(emptyAction.Action)); + } + + [Fact] + public void ContainsShouldSearchByToken() + { + var PubSubEvent = new TestableEventStream(); + var emptyAction = new ActionHelper(); + var token = PubSubEvent.Subscribe(emptyAction.Action, null); + + Assert.True(PubSubEvent.Contains(token)); + + PubSubEvent.Unsubscribe(emptyAction.Action); + Assert.False(PubSubEvent.Contains(token)); + } + + [Fact] + public void SubscribeDefaultsToPublisherThread() + { + var PubSubEvent = new TestableEventStream(); + Action action = delegate { }; + var token = PubSubEvent.Subscribe(action, action, ThreadOption.PublisherThread, true); + + Assert.Single(PubSubEvent.BaseSubscriptions); + Assert.Equal(typeof(EventSubscription), PubSubEvent.BaseSubscriptions.ElementAt(0).GetType()); + } + + + // ── Test doubles ────────────────────────────────────────────────────────── + + private class TestableEventStream : EventStream + { + public ICollection BaseSubscriptions => Subscriptions; + } + + private class SmallBacklogEventStream : EventStream + { + protected override uint BacklogSize() => 3; + } + + private class ExternalAction + { + public string PassedValue; + public bool Executed = false; + + public void ExecuteAction(string value) + { + PassedValue = value; + Executed = true; + } + + public void ExecuteActionBacklog(string value) + { + Executed = true; + } + } + } +} +#endif From d8e386db1f83485022a1e6dd8a3a2fc3af49595a Mon Sep 17 00:00:00 2001 From: Azat Tazayan Date: Sat, 25 Apr 2026 23:02:35 -0500 Subject: [PATCH 2/3] Rename private fields to use leading underscore convention Renamed private fields in EventStream and Backlog to use a leading underscore (_) for consistency with C# naming conventions. No functional changes were made. --- src/Prism.Events/EventStream.cs | 42 ++++++++++++++++----------------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/src/Prism.Events/EventStream.cs b/src/Prism.Events/EventStream.cs index fb96aeacc..d46dda4e0 100644 --- a/src/Prism.Events/EventStream.cs +++ b/src/Prism.Events/EventStream.cs @@ -31,7 +31,7 @@ public class EventStream : EventBase { private const int DefaultBacklogSize = 10; - private Backlog recentEvents; + private Backlog _recentEvents; /// /// Initializes a new instance of with the default @@ -39,7 +39,7 @@ public class EventStream : EventBase /// public EventStream() { - recentEvents = new Backlog((int)BacklogSize()); + _recentEvents = new Backlog((int)BacklogSize()); } /// @@ -205,7 +205,7 @@ private SubscriptionToken Subscribe(EventSubscription eventSubscriptio lock (Subscriptions) { Subscriptions.Add(eventSubscription); - var ring = recentEvents.CurrentState(out uint readPosition, out uint writePosition); + var ring = _recentEvents.CurrentState(out uint readPosition, out uint writePosition); backlog = ArrayPool.Shared.Rent(ring.Length); int index = 0; TPayload item = default; @@ -297,7 +297,7 @@ override protected void InternalPublish(params object[] arguments) lock (Subscriptions) { - recentEvents.Write((TPayload)arguments[0]); + _recentEvents.Write((TPayload)arguments[0]); strategies = new List>(Subscriptions.Count); @@ -346,10 +346,10 @@ override protected SubscriptionToken InternalSubscribe(IEventSubscription eventS /// The element type stored in the ring buffer. private class Backlog { - private readonly int size; - private readonly T[] ring; - private uint writePosition; - private uint readPosition; + private readonly int _size; + private readonly T[] _ring; + private uint _writePosition; + private uint _readPosition; /// /// Initializes a new with the specified capacity. @@ -357,10 +357,10 @@ private class Backlog /// Maximum number of items the ring buffer can hold. public Backlog(int capacity) { - size = capacity; - ring = new T[capacity]; - writePosition = 0; - readPosition = 0; + _size = capacity; + _ring = new T[capacity]; + _writePosition = 0; + _readPosition = 0; } /// @@ -370,16 +370,16 @@ public Backlog(int capacity) /// The item to write. public void Write(T item) { - bool isFull = writePosition - readPosition >= ring.Length; + bool isFull = _writePosition - _readPosition >= _ring.Length; if (isFull) { - readPosition++; + _readPosition++; } - ring[writePosition % size] = item; - writePosition++; + _ring[_writePosition % _size] = item; + _writePosition++; } /// @@ -396,8 +396,8 @@ public void Write(T item) /// public bool TryRead(out T item) { - var span = new ReadOnlySpan(ring); - return TryRead(ref span, ref readPosition, writePosition, out item); + var span = new ReadOnlySpan(_ring); + return TryRead(ref span, ref _readPosition, _writePosition, out item); } /// @@ -418,9 +418,9 @@ public bool TryRead(out T item) /// A over the internal ring array. public ReadOnlySpan CurrentState(out uint readPosition, out uint writePosition) { - readPosition = this.readPosition; - writePosition = this.writePosition; - return ring; + readPosition = _readPosition; + writePosition = _writePosition; + return _ring; } /// From ae4dcf06d9c9efd5d9a176866a389c455eda5704 Mon Sep 17 00:00:00 2001 From: Azat Tazayan Date: Sat, 25 Apr 2026 23:36:28 -0500 Subject: [PATCH 3/3] Clarify backlogAction threading in EventStream docs Updated XML docs for EventStream to specify that backlogAction is always invoked synchronously on the publisher's thread, regardless of ThreadOption. Added remarks to distinguish thread marshalling behavior for backlogAction vs. action. Clarified lock usage during backlog snapshot and invocation timing. --- src/Prism.Events/EventStream.cs | 30 +++++++++++++++++++++++------- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/src/Prism.Events/EventStream.cs b/src/Prism.Events/EventStream.cs index d46dda4e0..733331a0c 100644 --- a/src/Prism.Events/EventStream.cs +++ b/src/Prism.Events/EventStream.cs @@ -50,7 +50,9 @@ public EventStream() /// The callback invoked for each future published event. /// /// The callback invoked once for each event currently in the backlog at the time of - /// subscription. May be to skip backlog replay. + /// subscription. Always invoked synchronously on the caller's (publisher's) thread, + /// regardless of . May be to skip + /// backlog replay. /// /// /// A that can be used to unsubscribe from the event stream. @@ -68,7 +70,9 @@ public SubscriptionToken Subscribe(Action action, Action bac /// The callback invoked for each future published event. /// /// The callback invoked once for each event currently in the backlog at the time of - /// subscription. May be to skip backlog replay. + /// subscription. Always invoked synchronously on the caller's (publisher's) thread, + /// regardless of . May be to skip + /// backlog replay. /// /// /// Specifies the thread on which is invoked. @@ -91,7 +95,9 @@ public SubscriptionToken Subscribe(Action action, Action bac /// The callback invoked for each future published event. /// /// The callback invoked once for each event currently in the backlog at the time of - /// subscription. May be to skip backlog replay. + /// subscription. Always invoked synchronously on the caller's (publisher's) thread, + /// regardless of . May be to skip + /// backlog replay. /// /// /// Specifies the thread on which is invoked. @@ -116,13 +122,22 @@ public SubscriptionToken Subscribe(Action action, Action bac /// Any events currently held in the backlog that satisfy are /// immediately delivered to . /// + /// + /// is always invoked synchronously on the caller's + /// (publisher's) thread before this method returns, regardless of + /// . Only future events delivered via + /// are marshalled according to . + /// /// The callback invoked for each future published event. /// /// The callback invoked once for each matching event currently in the backlog at the time - /// of subscription. May be to skip backlog replay. + /// of subscription. Always invoked synchronously on the caller's (publisher's) thread, + /// regardless of . May be to skip + /// backlog replay. /// /// /// Specifies the thread on which is invoked. + /// Does not affect the thread used to invoke . /// /// /// When , the event stream holds a strong reference to @@ -177,8 +192,8 @@ public SubscriptionToken Subscribe(Action action, Action bac /// /// Registers an already-constructed , assigns it a - /// new , and replays the current backlog via - /// while holding the subscription lock. + /// new , builds the backlog snapshot under the lock, then invokes + /// outside the lock. /// /// The subscription to register. /// @@ -318,7 +333,8 @@ override protected void InternalPublish(params object[] arguments) } /// - /// Overrides to route subscriptions through + /// Overrides to route + /// subscriptions through /// the backlog-aware Subscribe overload. Backlog replay is skipped (null /// backlog action) when subscribing via the base-class path. ///