diff --git a/src/Prism.Events/EventStream.cs b/src/Prism.Events/EventStream.cs
new file mode 100644
index 000000000..733331a0c
--- /dev/null
+++ b/src/Prism.Events/EventStream.cs
@@ -0,0 +1,477 @@
+#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. 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.
+ ///
+ 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. 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.
+ /// 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. 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.
+ ///
+ ///
+ /// 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 .
+ ///
+ ///
+ /// 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. 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
+ /// . 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 , builds the backlog snapshot under the lock, then invokes
+ /// outside the 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 = _readPosition;
+ writePosition = _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