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); } } }