From b20cffd16a02875d308fb7d4751ea0f864748ccb Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Sat, 4 Jul 2026 02:06:17 -0500 Subject: [PATCH] Fixed that subscriptions occurring during an `.Edit()` operation upon a `SourceCache<>` or `SourceList<>` would still publish an initial notification to the subscriber. This resulted in the potential for notifications to be duplicated, for items added or manipulated during the edit. Instead, such subscriptions now have their notifications deferred until the `.Edit()` is complete. Resolves #1129. --- .../Cache/SourceCacheFixture.cs | 53 +++++++++++- .../List/SourceListFixture.cs | 50 ++++++++++++ src/DynamicData/Cache/ObservableCache.cs | 44 ++++++++-- src/DynamicData/List/SourceList.cs | 80 +++++++++++++------ 4 files changed, 194 insertions(+), 33 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs index 99b79fbfd..24b628e32 100644 --- a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs +++ b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs @@ -1,12 +1,13 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Disposables; using System.Reactive.Linq; using System.Threading; using System.Threading.Tasks; using DynamicData.Tests.Domain; - +using DynamicData.Tests.Utilities; using FluentAssertions; using Xunit; @@ -354,5 +355,55 @@ public void ConnectDuringDeliveryDoesNotDuplicate() addCounts.GetValueOrDefault("k2").Should().Be(1, "k2 should appear once, not duplicated from snapshot + queued delivery"); } + // Covers https://github.com/reactivemarbles/DynamicData/issues/1129 + [Fact] + public void ConnectDuringEditDoesNotDuplicate() + { + using var items = new SourceCache(static item => item); + + using var subscriptions = new CompositeDisposable(); + + // An initial subscription is required to initiate internal buffering of changes, during the upcoming .Edit(). + // That is, we want there to be changes buffered, internally, when the mid-edit subscription comes in, to + // ensure that they don't get duplicated. This is the scenario that came in up #1129. + subscriptions.Add(items + .Connect() + .Subscribe()); + + CacheItemRecordingObserver? results = null; + + items.Edit(inner => + { + inner.AddOrUpdate(1); + + subscriptions.Add(items + .Connect() + .ValidateChangeSets(static item => item) + .RecordCacheItems(out results)); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + + inner.AddOrUpdate(2); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + }); + + results.Should().NotBeNull("the edit delegate should have been invoked"); + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().ContainSingle("subscribers should only receive a single initial changeset"); + results.RecordedItemsByKey.Should().BeEquivalentTo( + new Dictionary() + { + [1] = 1, + [2] = 2 + }, + options => options.WithoutStrictOrdering(), + "all items in the source should have propagated downstream"); + + results.HasCompleted.Should().BeFalse("the source has not yet completed"); + } + private sealed record TestItem(string Key, string Value); } diff --git a/src/DynamicData.Tests/List/SourceListFixture.cs b/src/DynamicData.Tests/List/SourceListFixture.cs index dc37bdaed..9323113b7 100644 --- a/src/DynamicData.Tests/List/SourceListFixture.cs +++ b/src/DynamicData.Tests/List/SourceListFixture.cs @@ -1,13 +1,63 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Disposables; + using FluentAssertions; using Xunit; +using DynamicData.Tests.Utilities; + namespace DynamicData.Tests.List; public class SourceListFixture { + // Covers https://github.com/reactivemarbles/DynamicData/issues/1129 + [Fact] + public void ConnectDuringEditDoesNotDuplicate() + { + using var items = new SourceList(); + + using var subscriptions = new CompositeDisposable(); + + // An initial subscription is required to initiate internal buffering of changes, during the upcoming .Edit(). + // That is, we want there to be changes buffered, internally, when the mid-edit subscription comes in, to + // ensure that they don't get duplicated. This is the scenario that came in up #1129. + subscriptions.Add(items + .Connect() + .Subscribe()); + + ListItemRecordingObserver? results = null; + + items.Edit(inner => + { + inner.Add(1); + + subscriptions.Add(items + .Connect() + .ValidateChangeSets() + .RecordListItems(out results)); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + + inner.Add(2); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + }); + + results.Should().NotBeNull("the edit delegate should have been invoked"); + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().ContainSingle("subscribers should only receive a single initial changeset"); + results.RecordedItems.Should().BeEquivalentTo( + new[] { 1, 2, }, + options => options.WithStrictOrdering(), + "all items in the source should have propagated downstream"); + + results.HasCompleted.Should().BeFalse("the source has not yet completed"); + } + [Fact] public void InitialChangeIsRange() { diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 5cbeca988..b950a0058 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -6,9 +6,8 @@ using System.Reactive.Disposables; using System.Reactive.Linq; using System.Reactive.Subjects; -using System.Threading; + using DynamicData.Binding; -using DynamicData.Cache; using DynamicData.Cache.Internal; // ReSharper disable once CheckNamespace @@ -42,6 +41,9 @@ internal sealed class ObservableCache : IObservableCache _notifications; + [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "Disposed with _cleanUp")] + private readonly BehaviorSubject _isEditInProgress; + private int _editLevel; // The level of recursion in editing. private long _currentVersion; // Monotonic counter incremented under lock for each enqueued change notification. @@ -53,6 +55,7 @@ public ObservableCache(IObservable> source) _readerWriter = new ReaderWriter(); _notifications = new DeliveryQueue(_locker, new CacheUpdateObserver(this)); _suspensionTracker = new(() => new SuspensionTracker()); + _isEditInProgress = new(false); var loader = source.Subscribe( changeSet => @@ -83,6 +86,7 @@ public ObservableCache(Func? keySelector = null) _readerWriter = new ReaderWriter(keySelector); _notifications = new DeliveryQueue(_locker, new CacheUpdateObserver(this)); _suspensionTracker = new(() => new SuspensionTracker()); + _isEditInProgress = new(false); _cleanUp = Disposable.Create(NotifyCompleted); } @@ -115,14 +119,22 @@ public IObservable> Connect(Func? predi { lock (_locker) { - var observable = (!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) + var observable = ((!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) + && (!_isEditInProgress.Value)) // Create the Connection Observable ? CreateConnectObservable(predicate, suppressEmptyChangeSets) // Defer until notifications are no longer suspended - : _suspensionTracker.Value.NotificationsSuspendedObservable.Do(static _ => { }, observer.OnCompleted) - .Where(static b => !b).Take(1).Select(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)).Switch(); + : Observable.CombineLatest( + _suspensionTracker.Value.NotificationsSuspendedObservable, + _isEditInProgress, + static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) + .Do(static _ => { }, observer.OnCompleted) + .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) + .Take(1) + .Select(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)) + .Switch(); return observable.SubscribeSafe(observer); } @@ -139,14 +151,22 @@ public IObservable> Watch(TKey key) => { lock (_locker) { - var observable = (!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) + var observable = ((!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) + && (!_isEditInProgress.Value)) // Create the Watch Observable ? CreateWatchObservable(key) // Defer until notifications are no longer suspended - : _suspensionTracker.Value.NotificationsSuspendedObservable.Do(static _ => { }, observer.OnCompleted) - .Where(static b => !b).Take(1).Select(_ => CreateWatchObservable(key)).Switch(); + : Observable.CombineLatest( + _suspensionTracker.Value.NotificationsSuspendedObservable, + _isEditInProgress, + static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) + .Do(static _ => { }, observer.OnCompleted) + .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) + .Take(1) + .Select(_ => CreateWatchObservable(key)) + .Switch(); return observable.SubscribeSafe(observer); } @@ -181,6 +201,7 @@ internal void UpdateFromIntermediate(Action> update ChangeSet? changes = null; _editLevel++; + _isEditInProgress.OnNext(_editLevel is not 0); if (_editLevel == 1) { var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; @@ -197,6 +218,8 @@ internal void UpdateFromIntermediate(Action> update { notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } + + _isEditInProgress.OnNext(_editLevel is not 0); } internal void UpdateFromSource(Action> updateAction) @@ -208,6 +231,7 @@ internal void UpdateFromSource(Action> updateActio ChangeSet? changes = null; _editLevel++; + _isEditInProgress.OnNext(_editLevel is not 0); if (_editLevel == 1) { var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; @@ -224,6 +248,8 @@ internal void UpdateFromSource(Action> updateActio { notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } + + _isEditInProgress.OnNext(_editLevel is not 0); } private IObservable> CreateConnectObservable(Func? predicate, bool suppressEmptyChangeSets) => @@ -386,6 +412,7 @@ public void OnError(Exception error) { cache._changesPreview.OnError(error); cache._changes.OnError(error); + cache._isEditInProgress.OnError(error); if (cache._countChanged.IsValueCreated) { @@ -402,6 +429,7 @@ public void OnCompleted() { cache._changes.OnCompleted(); cache._changesPreview.OnCompleted(); + cache._isEditInProgress.OnCompleted(); if (cache._countChanged.IsValueCreated) { diff --git a/src/DynamicData/List/SourceList.cs b/src/DynamicData/List/SourceList.cs index 3db42bc11..8ee2968d6 100644 --- a/src/DynamicData/List/SourceList.cs +++ b/src/DynamicData/List/SourceList.cs @@ -36,6 +36,9 @@ public sealed class SourceList : ISourceList private readonly ReaderWriter _readerWriter = new(); + [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "Disposal is superfluous after completion, and causes a bunch of test failures")] + private readonly BehaviorSubject _isEditInProgress; + private int _editLevel; /// @@ -44,6 +47,8 @@ public sealed class SourceList : ISourceList /// The source. public SourceList(IObservable>? source = null) { + _isEditInProgress = new(false); + var loader = source is null ? Disposable.Empty : LoadFromSource(source); _cleanUp = Disposable.Create( @@ -78,40 +83,32 @@ public SourceList(IObservable>? source = null) /// public IObservable> Connect(Func? predicate = null) - { - var observable = Observable.Create>( - observer => + => Observable.Create>(observer => + { + lock (_locker) { - lock (_locker) - { - if (_readerWriter.Items.Length > 0) - { - observer.OnNext( - new ChangeSet - { - new(ListChangeReason.AddRange, _readerWriter.Items, 0) - }); - } + var observable = !_isEditInProgress.Value - var source = _changes.Finally(observer.OnCompleted); + // Create the Connection Observable + ? CreateConnectObservable(predicate) - return source.SubscribeSafe(observer); - } - }); - - if (predicate is not null) - { - observable = new FilterStatic(observable, predicate).Run(); - } + // Defer until notifications are no longer suspended + : _isEditInProgress + .Where(static isEditInProgress => !isEditInProgress) + .Take(1) + .Select(_ => CreateConnectObservable(predicate)) + .Switch(); - return observable; - } + return observable.SubscribeSafe(observer); + } + }); /// public void Dispose() { _cleanUp.Dispose(); _changesPreview.Dispose(); + _isEditInProgress.OnCompleted(); } /// @@ -124,6 +121,7 @@ public void Edit(Action> updateAction) IChangeSet? changes = null; _editLevel++; + _isEditInProgress.OnNext(_editLevel is not 0); if (_editLevel == 1) { @@ -140,6 +138,8 @@ public void Edit(Action> updateAction) { InvokeNext(changes); } + + _isEditInProgress.OnNext(_editLevel is not 0); } } @@ -156,6 +156,36 @@ public IObservable> Preview(Func? predicate = null) return observable; } + private IObservable> CreateConnectObservable(Func? predicate) + { + var observable = Observable.Create>( + observer => + { + lock (_locker) + { + if (_readerWriter.Items.Length > 0) + { + observer.OnNext( + new ChangeSet + { + new(ListChangeReason.AddRange, _readerWriter.Items, 0) + }); + } + + var source = _changes.Finally(observer.OnCompleted); + + return source.SubscribeSafe(observer); + } + }); + + if (predicate is not null) + { + observable = new FilterStatic(observable, predicate).Run(); + } + + return observable; + } + private void InvokeNext(IChangeSet changes) { if (changes.Count == 0) @@ -195,6 +225,7 @@ private void OnCompleted() { _changesPreview.OnCompleted(); _changes.OnCompleted(); + _isEditInProgress.OnCompleted(); } } @@ -204,6 +235,7 @@ private void OnError(Exception exception) { _changesPreview.OnError(exception); _changes.OnError(exception); + _isEditInProgress.OnError(exception); } } }