added fix for #3393 by tazayan · Pull Request #3394 · PrismLibrary/Prism

tazayan · 2026-04-26T04:50:17Z

Description of Change

Introduced EventStream<TPayload> 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.

Renamed private fields in EventStream<TPayload> and Backlog<T> to use a leading underscore (_) for consistency with C# naming conventions. No functional changes were made.

Updated XML docs for EventStream<TPayload> 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.

Copilot

Pull request overview

Implements the new EventStream<TPayload> event type requested in issue #3393, adding a pub/sub event that retains a rolling backlog and can replay it to late subscribers.

Changes:

Added EventStream<TPayload> implementation in Prism.Events with backlog replay + ring-buffer eviction.
Added extensive unit tests covering backlog replay, filtering, thread options, unsubscribe semantics, and weak/strong reference behavior.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 8 comments.

File	Description
`src/Prism.Events/EventStream.cs`	New event type with rolling backlog, backlog replay on subscribe, and integration with Prism’s existing subscription model.
`tests/Prism.Core.Tests/Events/EventStreamFixture.cs`	New test suite validating backlog behavior and parity with existing `PubSubEvent` semantics.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-04-27T13:40:05Z

+    /// <param name="backlogAction">
+    /// 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 <see cref="ThreadOption"/>. May be <see langword="null"/> to skip
+    /// backlog replay.
+    /// </param>


The XML docs say backlogAction is invoked synchronously on the caller's “(publisher's) thread”. Backlog replay happens during Subscribe(), so the callback runs on the subscriber/caller thread, not necessarily the publish thread; this wording is misleading and should be corrected (and kept consistent across overloads).

Copilot · 2026-04-27T13:40:05Z

+                var ring = _recentEvents.CurrentState(out uint readPosition, out uint writePosition);
+                backlog = ArrayPool<TPayload>.Shared.Rent(ring.Length);
+                int index = 0;
+                TPayload item = default;
+                while (index < ring.Length &&
+                      Backlog<TPayload>.TryRead(ref ring, ref readPosition, writePosition, out item))
+                {
+                    backlog[index] = item;
+                    index++;
+                }
+
+                backlogCount = index;


Backlog snapshotting work is performed even when backlogAction is null (array rent + copy under lock), which adds unnecessary overhead for callers that don't need replay (including the InternalSubscribe path). Consider skipping the snapshot entirely when backlogAction is null.

Suggested change

var ring = _recentEvents.CurrentState(out uint readPosition, out uint writePosition);

backlog = ArrayPool<TPayload>.Shared.Rent(ring.Length);

int index = 0;

TPayload item = default;

while (index < ring.Length &&

Backlog<TPayload>.TryRead(ref ring, ref readPosition, writePosition, out item))

{

backlog[index] = item;

index++;

}

backlogCount = index;

if (backlogAction != null)

{

var ring = _recentEvents.CurrentState(out uint readPosition, out uint writePosition);

backlog = ArrayPool<TPayload>.Shared.Rent(ring.Length);

int index = 0;

TPayload item = default;

while (index < ring.Length &&

Backlog<TPayload>.TryRead(ref ring, ref readPosition, writePosition, out item))

{

backlog[index] = item;

index++;

}

backlogCount = index;

}

Copilot · 2026-04-27T13:40:06Z

+        finally
+        {
+            if (backlog != null)
+                ArrayPool<TPayload>.Shared.Return(backlog, clearArray: true);
+        }


The pooled backlog array is always returned with clearArray: true. For value-type payloads this can add avoidable CPU cost; consider clearing only when needed (e.g., based on whether TPayload is a reference type / contains references).

Copilot · 2026-04-27T13:40:06Z

+    /// </summary>
+    public EventStream()
+    {
+        _recentEvents = new Backlog<TPayload>((int)BacklogSize());


BacklogSize() can be overridden to return 0, which would construct a ring buffer with length 0 and cause a divide-by-zero in Backlog.Write (_writePosition % _size). Add validation to ensure the capacity is >= 1 (either when creating _recentEvents or inside Backlog's constructor).

Suggested change

_recentEvents = new Backlog<TPayload>((int)BacklogSize());

int backlogSize = (int)BacklogSize();

if (backlogSize < 1)

{

throw new ArgumentOutOfRangeException(nameof(backlogSize), backlogSize, "Backlog size must be greater than or equal to 1.");

}

_recentEvents = new Backlog<TPayload>(backlogSize);

Copilot · 2026-04-27T13:40:06Z

+        private uint _writePosition;
+        private uint _readPosition;
+
+        /// <summary>
+        /// Initializes a new <see cref="Backlog{T}"/> with the specified capacity.
+        /// </summary>
+        /// <param name="capacity">Maximum number of items the ring buffer can hold.</param>
+        public Backlog(int capacity)
+        {
+            _size = capacity;
+            _ring = new T[capacity];
+            _writePosition = 0;
+            _readPosition = 0;
+        }
+
+        /// <summary>
+        /// Writes <paramref name="item"/> to the ring buffer. If the buffer is full, the
+        /// oldest unread item is evicted by advancing the read position before writing.
+        /// </summary>
+        /// <param name="item">The item to write.</param>
+        public void Write(T item)
+        {
+            bool isFull = _writePosition - _readPosition >= _ring.Length;
+
+
+            if (isFull)
+            {
+                _readPosition++;
+            }


The ring buffer cursors use uint arithmetic (_writePosition - _readPosition) to detect fullness. After enough publishes, these counters will overflow and wrap, which can break the fullness/ordering logic. Consider using ulong/long or tracking count separately to avoid overflow-related corruption in long-running apps.

Copilot · 2026-04-27T13:40:07Z

+                var subscription = Subscriptions.ElementAt(i);
+                var strategy = subscription.GetExecutionStrategy();
+                if (strategy == null)
+                    Subscriptions.Remove(subscription);// prune dead weak refs


When pruning dead subscriptions during publish, the code removes by value (Subscriptions.Remove(subscription)) even though the index is known. Since Subscriptions is backed by a List<IEventSubscription> (EventBase.cs:13), removing by index (when possible) avoids an extra linear search.

Suggested change

Subscriptions.Remove(subscription);// prune dead weak refs

Subscriptions.RemoveAt(i);// prune dead weak refs

Copilot · 2026-04-27T13:40:07Z

+#if NET8_0_OR_GREATER
+using System.Buffers;
+using Prism.Events.Properties;
+
+namespace Prism.Events;
+


EventStream.cs uses Action, List<>, Predicate<>, LINQ (Cast/FirstOrDefault/ElementAt), and exceptions, but only imports System.Buffers and Prism.Events.Properties. This will not compile without adding the required using directives (e.g., System, System.Collections.Generic, System.Linq).

brianlagunas · 2026-04-27T14:23:59Z

Not accepting this PR. Details in the related issue #3393

tazayan added 3 commits April 25, 2026 22:55

Rename private fields to use leading underscore convention

d8e386d

Renamed private fields in EventStream<TPayload> and Backlog<T> to use a leading underscore (_) for consistency with C# naming conventions. No functional changes were made.

tazayan requested review from brianlagunas and dansiegel as code owners April 26, 2026 04:50

dansiegel requested a review from Copilot April 27, 2026 13:34

Copilot started reviewing on behalf of dansiegel April 27, 2026 13:34 View session

Copilot AI reviewed Apr 27, 2026

View reviewed changes

brianlagunas closed this Apr 27, 2026

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

added fix for #3393#3394

added fix for #3393#3394
tazayan wants to merge 3 commits intoPrismLibrary:masterfrom
tazayan:dev/eventstreaming

tazayan commented Apr 26, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Apr 27, 2026

Uh oh!

Copilot AI Apr 27, 2026

Uh oh!

Copilot AI Apr 27, 2026

Uh oh!

Copilot AI Apr 27, 2026

Uh oh!

Copilot AI Apr 27, 2026

Uh oh!

Copilot AI Apr 27, 2026

Uh oh!

Copilot AI Apr 27, 2026

Uh oh!

Uh oh!

brianlagunas commented Apr 27, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

-        _recentEvents = new Backlog<TPayload>((int)BacklogSize());
+        int backlogSize = (int)BacklogSize();
+        if (backlogSize < 1)
+        {
+            throw new ArgumentOutOfRangeException(nameof(backlogSize), backlogSize, "Backlog size must be greater than or equal to 1.");
+        }
+        _recentEvents = new Backlog<TPayload>(backlogSize);

	Subscriptions.Remove(subscription);// prune dead weak refs
	Subscriptions.RemoveAt(i);// prune dead weak refs

Uh oh!

Conversation

tazayan commented Apr 26, 2026

Description of Change

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 27, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 27, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 27, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 27, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 27, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 27, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 27, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

brianlagunas commented Apr 27, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants