diff --git a/src/DynamicData/Internal/CacheParentSubscription.cs b/src/DynamicData/Internal/CacheParentSubscription.cs
index 3a33143d..e817c9fb 100644
--- a/src/DynamicData/Internal/CacheParentSubscription.cs
+++ b/src/DynamicData/Internal/CacheParentSubscription.cs
@@ -13,7 +13,8 @@ namespace DynamicData.Internal;
/// when either the parent or child gets a new value.
/// Uses a for serialization and lock-free delivery.
/// Same-thread reentrant delivery preserves child-during-parent ordering.
-/// OnDrainComplete calls EmitChanges after the outermost delivery, outside the lock.
+/// Accumulated changes are emitted once per delivery frame, where a frame is one
+/// notification plus anything delivered synchronously beneath it on the same thread.
///
/// Type of the Parent ChangeSet.
/// Type for the Parent ChangeSet Key.
@@ -29,6 +30,7 @@ internal abstract class CacheParentSubscription _observer;
private int _subscriptionCounter = 1; // Starts at 1 for the parent subscription
+ private int _frameDepth;
private bool _isCompleted;
private bool _hasTerminated;
private bool _disposedValue;
@@ -40,7 +42,7 @@ internal abstract class CacheParentSubscription observer)
{
_observer = observer;
- _queue = new SharedDeliveryQueue(onDrainComplete: OnDrainComplete);
+ _queue = new SharedDeliveryQueue();
}
///
@@ -76,9 +78,9 @@ protected void AddChildSubscription(IObservable observable, TKey parentK
disposableContainer.Disposable = observable
.Finally(CheckCompleted)
.SubscribeSafe(
- onNext: val => ChildOnNext(val, parentKey),
+ onNext: val => DeliverChild(val, parentKey),
onError: TerminalError,
- onCompleted: () => RemoveChildSubscription(parentKey));
+ onCompleted: () => CompleteChild(parentKey));
}
protected void RemoveChildSubscription(TKey parentKey) => _childSubscriptions.Remove(parentKey);
@@ -88,9 +90,9 @@ protected void CreateParentSubscription(IObservable> s
source
.SynchronizeSafe(_queue)
.SubscribeSafe(
- onNext: ParentOnNext,
+ onNext: DeliverParent,
onError: TerminalError,
- onCompleted: CheckCompleted);
+ onCompleted: CompleteParent);
protected virtual void Dispose(bool disposing)
{
@@ -116,8 +118,55 @@ protected virtual void Dispose(bool disposing)
protected IObservable MakeChildObservable(IObservable observable) =>
observable.SynchronizeSafe(_queue);
- private void OnDrainComplete()
+ private void DeliverParent(IChangeSet changes)
{
+ using var frame = BeginFrame();
+ ParentOnNext(changes);
+ }
+
+ private void DeliverChild(TChild child, TKey parentKey)
+ {
+ using var frame = BeginFrame();
+ ChildOnNext(child, parentKey);
+ }
+
+ private void CompleteParent()
+ {
+ using var frame = BeginFrame();
+ CheckCompleted();
+ }
+
+ private void CompleteChild(TKey parentKey)
+ {
+ using var frame = BeginFrame();
+ RemoveChildSubscription(parentKey);
+ }
+
+ ///
+ /// Opens a delivery frame that stays open until the returned is disposed.
+ /// Deliveries nested beneath this one, which the queue runs inline on the same thread, open and close
+ /// their own frame and leave the emit to the outermost, so one upstream notification and everything it
+ /// triggers synchronously produce a single downstream changeset. No lock is needed around the depth
+ /// because the queue has already serialized delivery.
+ ///
+ /// A tracker that closes the frame when disposed.
+ private FrameTracker BeginFrame()
+ {
+ ++_frameDepth;
+ return new FrameTracker(this);
+ }
+
+ ///
+ /// Closes the current delivery frame, emitting the accumulated changes only when the outermost
+ /// frame closes.
+ ///
+ private void EndFrame()
+ {
+ if (--_frameDepth != 0)
+ {
+ return;
+ }
+
EmitChanges(_observer);
if (Volatile.Read(ref _isCompleted) && !_hasTerminated)
@@ -142,4 +191,15 @@ private void CheckCompleted()
Debug.Assert(_subscriptionCounter >= 0, "Should never be negative");
}
+
+ ///
+ /// Closes the delivery frame opened by when disposed, so a frame can be
+ /// scoped with instead of pairing the calls by hand.
+ ///
+ /// The subscription whose frame is being tracked.
+ private readonly struct FrameTracker(CacheParentSubscription owner) : IDisposable
+ {
+ ///
+ public void Dispose() => owner.EndFrame();
+ }
}
diff --git a/src/DynamicData/Internal/SharedDeliveryQueue.cs b/src/DynamicData/Internal/SharedDeliveryQueue.cs
index 3ca1f2b2..cdcd762f 100644
--- a/src/DynamicData/Internal/SharedDeliveryQueue.cs
+++ b/src/DynamicData/Internal/SharedDeliveryQueue.cs
@@ -29,8 +29,6 @@ internal sealed class SharedDeliveryQueue : IDisposable
///
private readonly Queue _order = new();
- private readonly Action? _onDrainComplete;
-
#if NET9_0_OR_GREATER
private readonly Lock _gate;
#else
@@ -42,22 +40,12 @@ internal sealed class SharedDeliveryQueue : IDisposable
/// Initializes a new instance of the class with its own internal lock.
public SharedDeliveryQueue()
- : this(onDrainComplete: null)
- {
- }
-
- ///
- /// Initializes a new instance of the class with its own internal lock
- /// and a callback that fires outside the lock after each drain cycle completes.
- ///
- public SharedDeliveryQueue(Action? onDrainComplete)
{
#if NET9_0_OR_GREATER
_gate = new Lock();
#else
_gate = new object();
#endif
- _onDrainComplete = onDrainComplete;
}
#if NET9_0_OR_GREATER
@@ -167,8 +155,6 @@ private void DrainAll()
return;
}
- _onDrainComplete?.Invoke();
-
// Atomically re-check for work and release ownership if there is none. Checking
// and releasing in separate lock scopes would let a producer enqueue in between,
// see that a drain is in progress, and rely on us to deliver an item we never saw.