diff --git a/src/DynamicData.Tests/Cache/MergeManyChangeSetsCacheFixture.cs b/src/DynamicData.Tests/Cache/MergeManyChangeSetsCacheFixture.cs index 37ceb06f4..37ca6b198 100644 --- a/src/DynamicData.Tests/Cache/MergeManyChangeSetsCacheFixture.cs +++ b/src/DynamicData.Tests/Cache/MergeManyChangeSetsCacheFixture.cs @@ -811,6 +811,47 @@ public void MergeManyChangeSetsWorksCorrectlyWithValueTypes() results.Summary.Overall.Removes.Should().Be(PricesPerMarket); } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void OrderOfChangesIsPreserved(bool removeFirst) + { + // Arrange + var markets = Enumerable.Range(0, MarketCount).Select(n => new Market(n)).ToArray(); + AddUniquePrices(markets); + _marketCache.AddOrUpdate(markets); + var markets2 = Enumerable.Range(0, MarketCount).Select(n => new Market(n)).ToArray(); + AddUniquePrices(markets2); + using var results = _marketCache.Connect().MergeManyChangeSets(m => m.LatestPrices, MarketPrice.EqualityComparer).AsAggregator(); + (var firstReason, var nextReason, int expectedChanges) = removeFirst + ? (ChangeReason.Remove, ChangeReason.Add, 2 * MarketCount * PricesPerMarket) + : (ChangeReason.Add, ChangeReason.Remove, 3 * MarketCount * PricesPerMarket); + + // Act + _marketCache.Edit(updater => + { + if (removeFirst) + { + updater.Clear(); + updater.AddOrUpdate(markets2); + } + else + { + + updater.AddOrUpdate(markets2); + updater.Clear(); + } + }); + + // Assert + results.Messages.Count.Should().Be(2); + results.Messages[0].All(change => change.Reason is ChangeReason.Add).Should().BeTrue(); + results.Messages[1].Count.Should().Be(expectedChanges); + results.Messages[1].Take(MarketCount * PricesPerMarket).All(change => change.Reason == firstReason).Should().BeTrue(); + results.Messages[1].Skip(MarketCount * PricesPerMarket).All(change => change.Reason == nextReason).Should().BeTrue(); + } + public void Dispose() { _marketCacheResults.Dispose(); diff --git a/src/DynamicData/Cache/Internal/ChangeSetCache.cs b/src/DynamicData/Cache/Internal/ChangeSetCache.cs index d98c0a824..f14ba57f4 100644 --- a/src/DynamicData/Cache/Internal/ChangeSetCache.cs +++ b/src/DynamicData/Cache/Internal/ChangeSetCache.cs @@ -16,7 +16,7 @@ internal sealed class ChangeSetCache where TKey : notnull { public ChangeSetCache(IObservable> source) => - Source = source.IgnoreSameReferenceUpdate().Do(Cache.Clone); + Source = source.Do(Cache.Clone); public Cache Cache { get; } = new(); diff --git a/src/DynamicData/Cache/Internal/DynamicGrouper.cs b/src/DynamicData/Cache/Internal/DynamicGrouper.cs index fe0776cd7..47c7f5c3b 100644 --- a/src/DynamicData/Cache/Internal/DynamicGrouper.cs +++ b/src/DynamicData/Cache/Internal/DynamicGrouper.cs @@ -44,37 +44,44 @@ public void ProcessChangeSet(IChangeSet changeSet, IObserver change) => ProcessChange(change, _suspendTracker); - case ChangeReason.Update when _groupSelector is not null: - PerformAddOrUpdate(change.Key, _groupSelector(change.Current, change.Key), change.Current, suspendTracker); - break; + private void ProcessChange(Change change, SuspendTracker? suspendTracker) + { + switch (change.Reason) + { + case ChangeReason.Add when _groupSelector is not null: + PerformAddOrUpdate(change.Key, _groupSelector(change.Current, change.Key), change.Current, suspendTracker); + break; - case ChangeReason.Update: - PerformUpdate(change.Key, suspendTracker); - break; + case ChangeReason.Remove: + PerformRemove(change.Key, suspendTracker); + break; - case ChangeReason.Refresh when _groupSelector is not null: - PerformRefresh(change.Key, _groupSelector(change.Current, change.Key), change.Current, suspendTracker); - break; + case ChangeReason.Update when _groupSelector is not null: + PerformAddOrUpdate(change.Key, _groupSelector(change.Current, change.Key), change.Current, suspendTracker); + break; - case ChangeReason.Refresh: - PerformRefresh(change.Key, suspendTracker); - break; - } - } + case ChangeReason.Update: + PerformUpdate(change.Key, suspendTracker); + break; - if (observer != null) - { - EmitChanges(observer); + case ChangeReason.Refresh when _groupSelector is not null: + PerformRefresh(change.Key, _groupSelector(change.Current, change.Key), change.Current, suspendTracker); + break; + + case ChangeReason.Refresh: + PerformRefresh(change.Key, suspendTracker); + break; } } diff --git a/src/DynamicData/Cache/Internal/GroupOnObservable.cs b/src/DynamicData/Cache/Internal/GroupOnObservable.cs index 4ae4df004..cf36cfd54 100644 --- a/src/DynamicData/Cache/Internal/GroupOnObservable.cs +++ b/src/DynamicData/Cache/Internal/GroupOnObservable.cs @@ -2,7 +2,6 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System.Reactive.Disposables; using System.Reactive.Linq; using DynamicData.Internal; @@ -13,49 +12,58 @@ internal sealed class GroupOnObservable(IObservable> Run() => Observable.Create>(observer => + public IObservable> Run() => + Observable.Create>(observer => new Subscription(source, selectGroup, observer)); + + // Maintains state for a single subscription + private sealed class Subscription : CacheParentSubscription> { - var grouper = new DynamicGrouper(); - var locker = InternalEx.NewLock(); - var parentUpdate = false; - - IObservable CreateGroupObservable(TObject item, TKey key) => - selectGroup(item, key) - .DistinctUntilChanged() - .Synchronize(locker!) - .Do( - onNext: groupKey => grouper!.AddOrUpdate(key, groupKey, item, !parentUpdate ? observer : null), - onError: observer.OnError); - - // Create a shared connection to the source - var shared = source - .Synchronize(locker) - .Do(_ => parentUpdate = true) - .Publish(); - - // First process the changesets - var subChanges = shared - .SubscribeSafe( - onNext: changeSet => grouper.ProcessChangeSet(changeSet), - onError: observer.OnError); - - // Next process the Grouping observables created for each item - var subMergeMany = shared - .MergeMany(CreateGroupObservable) - .SubscribeSafe( - onError: observer.OnError, - onCompleted: observer.OnCompleted); - - // Finally, emit the results - var subResults = shared - .SubscribeSafe( - onNext: _ => + private readonly DynamicGrouper _grouper = new(); + private readonly Func> _selectGroup; + + public Subscription(IObservable> source, Func> selectGroup, IObserver> observer) + : base(observer) + { + _selectGroup = selectGroup; + CreateParentSubscription(source); + } + + protected override void ParentOnNext(IChangeSet changes) + { + // Process all the changes at once to preserve the changeset order + foreach (var change in changes.ToConcreteType()) + { + _grouper.ProcessChange(change); + + switch (change.Reason) { - grouper.EmitChanges(observer); - parentUpdate = false; - }, - onError: observer.OnError); + // Shutdown existing sub (if any) and create a new one that + // Will update the group key for the current item + case ChangeReason.Add or ChangeReason.Update: + AddGroupSubscription(change.Current, change.Key); + break; + + // Shutdown the existing subscription + case ChangeReason.Remove: + RemoveChildSubscription(change.Key); + break; + } + } + } + + protected override void ChildOnNext((TGroupKey, TObject) tuple, TKey parentKey) => + _grouper.AddOrUpdate(parentKey, tuple.Item1, tuple.Item2); + + protected override void EmitChanges(IObserver> observer) => + _grouper.EmitChanges(observer); + + protected override void Dispose(bool disposing) + { + _grouper.Dispose(); + base.Dispose(disposing); + } - return new CompositeDisposable(shared.Connect(), subMergeMany, subChanges, grouper); - }); + private void AddGroupSubscription(TObject obj, TKey key) => + AddChildSubscription(MakeChildObservable(_selectGroup(obj, key).DistinctUntilChanged().Select(groupKey => (groupKey, obj))), key); + } } diff --git a/src/DynamicData/Cache/Internal/MergeChangeSets.cs b/src/DynamicData/Cache/Internal/MergeChangeSets.cs index 041445fdf..c55c60d3e 100644 --- a/src/DynamicData/Cache/Internal/MergeChangeSets.cs +++ b/src/DynamicData/Cache/Internal/MergeChangeSets.cs @@ -43,14 +43,14 @@ public IObservable> Run() => Observable.Create, int> CreateChange(IObservable> source, int index, Lock locker) => - new(ChangeReason.Add, index, new ChangeSetCache(source.Synchronize(locker))); + new(ChangeReason.Add, index, new ChangeSetCache(source.IgnoreSameReferenceUpdate().Synchronize(locker))); // Create a ChangeSet Observable that produces ChangeSets with a single Add event for each new sub-observable private static IObservable, int>> CreateContainerObservable(IObservable>> source, Lock locker) => source.Select((src, index) => new ChangeSet, int>(new[] { CreateChange(src, index, locker) })); #else private static Change, int> CreateChange(IObservable> source, int index, object locker) => - new(ChangeReason.Add, index, new ChangeSetCache(source.Synchronize(locker))); + new(ChangeReason.Add, index, new ChangeSetCache(source.IgnoreSameReferenceUpdate().Synchronize(locker))); // Create a ChangeSet Observable that produces ChangeSets with a single Add event for each new sub-observable private static IObservable, int>> CreateContainerObservable(IObservable>> source, object locker) => diff --git a/src/DynamicData/Cache/Internal/MergeManyCacheChangeSets.cs b/src/DynamicData/Cache/Internal/MergeManyCacheChangeSets.cs index 1b704470d..56ae6a07f 100644 --- a/src/DynamicData/Cache/Internal/MergeManyCacheChangeSets.cs +++ b/src/DynamicData/Cache/Internal/MergeManyCacheChangeSets.cs @@ -2,7 +2,6 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System.Reactive.Disposables; using System.Reactive.Linq; using DynamicData.Internal; @@ -11,53 +10,68 @@ namespace DynamicData.Cache.Internal; /// /// Operator that is similiar to MergeMany but intelligently handles Cache ChangeSets. /// -internal sealed class MergeManyCacheChangeSets(IObservable> source, Func>> selector, IEqualityComparer? equalityComparer, IComparer? comparer) +internal sealed class MergeManyCacheChangeSets(IObservable> source, Func>> changeSetSelector, IEqualityComparer? equalityComparer, IComparer? comparer) where TObject : notnull where TKey : notnull where TDestination : notnull where TDestinationKey : notnull { public IObservable> Run() => Observable.Create>( - observer => + observer => new Subscription(source, changeSetSelector, observer, equalityComparer, comparer)); + + // Maintains state for a single subscription + private sealed class Subscription : CacheParentSubscription, TKey, IChangeSet, IChangeSet> + { + private readonly Cache, TKey> _cache = new(); + private readonly ChangeSetMergeTracker _changeSetMergeTracker; + + public Subscription( + IObservable> source, + Func>> changeSetSelector, + IObserver> observer, + IEqualityComparer? equalityComparer, + IComparer? comparer) + : base(observer) + { + _changeSetMergeTracker = new(() => _cache.Items, comparer, equalityComparer); + + // Child Observable has to go into the ChangeSetCache so the locking protects it + CreateParentSubscription(source.Transform((obj, key) => + new ChangeSetCache(MakeChildObservable(changeSetSelector(obj, key).IgnoreSameReferenceUpdate())))); + } + + protected override void ParentOnNext(IChangeSet, TKey> changes) { - var locker = InternalEx.NewLock(); - var cache = new Cache, TKey>(); - var parentUpdate = false; - - // This is manages all of the changes - var changeTracker = new ChangeSetMergeTracker(() => cache.Items, comparer, equalityComparer); - - // Transform to a cache changeset of child caches, synchronize, update the local copy, and publish. - var shared = source - .Transform((obj, key) => new ChangeSetCache(selector(obj, key).Synchronize(locker))) - .Synchronize(locker) - .Do(changes => + // Process all the changes at once to preserve the changeset order + foreach (var change in changes.ToConcreteType()) + { + switch (change.Reason) { - cache.Clone(changes); - parentUpdate = true; - }) - .Publish(); - - // Merge the child changeset changes together and apply to the tracker - var subMergeMany = shared - .MergeMany(cacheChangeSet => cacheChangeSet.Source) - .SubscribeSafe( - changes => changeTracker.ProcessChangeSet(changes, !parentUpdate ? observer : null), - observer.OnError, - observer.OnCompleted); - - // When a source item is removed, all of its sub-items need to be removed - var subRemove = shared - .OnItemRemoved(changeSetCache => changeTracker.RemoveItems(changeSetCache.Cache.KeyValues), invokeOnUnsubscribe: false) - .OnItemUpdated((_, prev) => changeTracker.RemoveItems(prev.Cache.KeyValues)) - .SubscribeSafe( - _ => - { - changeTracker.EmitChanges(observer); - parentUpdate = false; - }, - observer.OnError); - - return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove); - }); + // Shutdown existing sub (if any) and create a new one that + // Will update the cache and emit the changes + case ChangeReason.Add or ChangeReason.Update: + _cache.AddOrUpdate(change.Current, change.Key); + AddChildSubscription(change.Current.Source, change.Key); + if (change.Previous.HasValue) + { + _changeSetMergeTracker.RemoveItems(change.Previous.Value.Cache.KeyValues); + } + break; + + // Shutdown the existing subscription and remove from the cache + case ChangeReason.Remove: + _cache.Remove(change.Key); + RemoveChildSubscription(change.Key); + _changeSetMergeTracker.RemoveItems(change.Current.Cache.KeyValues); + break; + } + } + } + + protected override void ChildOnNext(IChangeSet changes, TKey parentKey) => + _changeSetMergeTracker.ProcessChangeSet(changes, null); + + protected override void EmitChanges(IObserver> observer) => + _changeSetMergeTracker.EmitChanges(observer); + } } diff --git a/src/DynamicData/Cache/Internal/MergeManyCacheChangeSetsSourceCompare.cs b/src/DynamicData/Cache/Internal/MergeManyCacheChangeSetsSourceCompare.cs index 6f0f80e15..754dcd929 100644 --- a/src/DynamicData/Cache/Internal/MergeManyCacheChangeSetsSourceCompare.cs +++ b/src/DynamicData/Cache/Internal/MergeManyCacheChangeSetsSourceCompare.cs @@ -2,9 +2,9 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System.Reactive.Disposables; using System.Reactive.Linq; using DynamicData.Internal; +using DynamicData.PLinq; namespace DynamicData.Cache.Internal; @@ -24,66 +24,78 @@ internal sealed class MergeManyCacheChangeSetsSourceCompare? _equalityComparer = (equalityComparer != null) ? new ParentChildEqualityCompare(equalityComparer) : null; - public IObservable> Run() => Observable.Create>( - observer => + public IObservable> Run() => + Observable.Create>(observer => new Subscription(source, _changeSetSelector, observer, _comparer, _equalityComparer, reevalOnRefresh)) + .TransformImmutable(entry => entry.Child); + + // Maintains state for a single subscription + private sealed class Subscription : CacheParentSubscription, TKey, IChangeSet, IChangeSet> + { + private readonly Cache, TKey> _cache = new(); + private readonly ChangeSetMergeTracker _changeSetMergeTracker; + private readonly bool _reevalOnRefresh; + + public Subscription( + IObservable> source, + Func>> changeSetSelector, + IObserver> observer, + IComparer comparer, + IEqualityComparer? equalityComparer, + bool reevalOnRefresh) + : base(observer) { - var locker = InternalEx.NewLock(); - var cache = new Cache, TKey>(); - var parentUpdate = false; - - // This is manages all of the changes - var changeTracker = new ChangeSetMergeTracker(() => cache.Items, _comparer, _equalityComparer); - - // Transform to an cache changeset of child caches of ParentChildEntry, synchronize, update the local copy, and publish. - var shared = source - .Transform((obj, key) => new ChangeSetCache(_changeSetSelector(obj, key).Synchronize(locker))) - .Synchronize(locker) - .Do(changes => - { - cache.Clone(changes); - parentUpdate = true; - }) - .Publish(); - - // Merge the child changeset changes together and apply to the tracker - var subMergeMany = shared - .MergeMany(changeSetCache => changeSetCache.Source) - .SubscribeSafe( - changes => changeTracker.ProcessChangeSet(changes, !parentUpdate ? observer : null), - observer.OnError, - observer.OnCompleted); - - // When a source item is removed, all of its sub-items need to be removed - var parentObservable = shared - .OnItemRemoved(cacheChangeSet => changeTracker.RemoveItems(cacheChangeSet.Cache.KeyValues), invokeOnUnsubscribe: false) - .OnItemUpdated((_, prev) => changeTracker.RemoveItems(prev.Cache.KeyValues)); - - // If requested, handle refresh events as well - if (reevalOnRefresh) - { - parentObservable = parentObservable.OnItemRefreshed(cacheChangeSet => changeTracker.RefreshItems(cacheChangeSet.Cache.Keys)); - } + _changeSetMergeTracker = new(() => _cache.Items, comparer, equalityComparer); + _reevalOnRefresh = reevalOnRefresh; - // Subscribe to handle all the requested changes and emit them downstream - var subParent = parentObservable - .SubscribeSafe( - _ => - { - changeTracker.EmitChanges(observer); - parentUpdate = false; - }, - observer.OnError); + // Child Observable has to go into the ChangeSetCache so the locking protects it + CreateParentSubscription(source.Transform((obj, key) => + new ChangeSetCache(MakeChildObservable(changeSetSelector(obj, key).IgnoreSameReferenceUpdate())))); + } - return new CompositeDisposable(shared.Connect(), subMergeMany, subParent); - }).TransformImmutable(entry => entry.Child); + protected override void ParentOnNext(IChangeSet, TKey> changes) + { + // Process all the changes at once to preserve the changeset order + foreach (var change in changes.ToConcreteType()) + { + switch (change.Reason) + { + // Shutdown existing sub (if any) and create a new one that + // Will update the cache and emit the changes + case ChangeReason.Add or ChangeReason.Update: + _cache.AddOrUpdate(change.Current, change.Key); + AddChildSubscription(change.Current.Source, change.Key); + if (change.Previous.HasValue) + { + _changeSetMergeTracker.RemoveItems(change.Previous.Value.Cache.KeyValues); + } + break; + + // Shutdown the existing subscription and remove from the cache + case ChangeReason.Remove: + _cache.Remove(change.Key); + RemoveChildSubscription(change.Key); + _changeSetMergeTracker.RemoveItems(change.Current.Cache.KeyValues); + break; + + case ChangeReason.Refresh: + if (_reevalOnRefresh) + { + _changeSetMergeTracker.RefreshItems(change.Current.Cache.Keys); + } + break; + } + } + } - private sealed class ParentChildEntry(TObject parent, TDestination child) - { - public TObject Parent { get; } = parent; + protected override void ChildOnNext(IChangeSet changes, TKey parentKey) => + _changeSetMergeTracker.ProcessChangeSet(changes, null); - public TDestination Child { get; } = child; + protected override void EmitChanges(IObserver> observer) => + _changeSetMergeTracker.EmitChanges(observer); } + private sealed record ParentChildEntry(TObject Parent, TDestination Child); + private sealed class ParentChildCompare(IComparer comparerParent, IComparer comparerChild) : Comparer { public override int Compare(ParentChildEntry? x, ParentChildEntry? y) => (x, y) switch diff --git a/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs b/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs index 8106a0a5a..054d24d3d 100644 --- a/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs +++ b/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs @@ -2,7 +2,6 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System.Reactive.Disposables; using System.Reactive.Linq; using DynamicData.Internal; using DynamicData.List.Internal; @@ -18,41 +17,55 @@ internal sealed class MergeManyListChangeSets(IObse where TDestination : notnull { public IObservable> Run() => Observable.Create>( - observer => + observer => new Subscription(source, selector, observer, equalityComparer)); + + // Maintains state for a single subscription + private sealed class Subscription : CacheParentSubscription, TKey, IChangeSet, IChangeSet> + { + private readonly ChangeSetMergeTracker _changeSetMergeTracker = new(); + + public Subscription( + IObservable> source, + Func>> selector, + IObserver> observer, + IEqualityComparer? equalityComparer) + : base(observer) + { + // RemoveIndex outside of the Lock, but add locking before going to ClonedChangeSet so the contents are protected + CreateParentSubscription(source.Transform((obj, key) => + new ClonedListChangeSet(MakeChildObservable(selector(obj, key).RemoveIndex()), equalityComparer))); + } + + protected override void ParentOnNext(IChangeSet, TKey> changes) { - var locker = InternalEx.NewLock(); - var parentUpdate = false; - - // This is manages all of the changes - var changeTracker = new ChangeSetMergeTracker(); - - // Transform to a cache changeset of child lists, synchronize, and publish. - var shared = source - .Transform((obj, key) => new ClonedListChangeSet(selector(obj, key).Synchronize(locker), equalityComparer)) - .Synchronize(locker) - .Do(_ => parentUpdate = true) - .Publish(); - - // Merge the child changeset changes together and apply to the tracker - var subMergeMany = shared - .MergeMany(clonedList => clonedList.Source.RemoveIndex()) - .SubscribeSafe( - changes => changeTracker.ProcessChangeSet(changes, !parentUpdate ? observer : null), - observer.OnError, - observer.OnCompleted); - - // When a source item is removed, all of its sub-items need to be removed - var subRemove = shared - .OnItemRemoved(clonedList => changeTracker.RemoveItems(clonedList.List), invokeOnUnsubscribe: false) - .OnItemUpdated((_, prev) => changeTracker.RemoveItems(prev.List)) - .SubscribeSafe( - _ => - { - changeTracker.EmitChanges(observer); - parentUpdate = false; - }, - observer.OnError); - - return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove); - }); + // Process all the changes at once to preserve the changeset order + foreach (var change in changes.ToConcreteType()) + { + switch (change.Reason) + { + // Shutdown existing sub (if any) and create a new one + // Remove any items from the previous list + case ChangeReason.Add or ChangeReason.Update: + AddChildSubscription(change.Current.Source, change.Key); + if (change.Previous.HasValue) + { + _changeSetMergeTracker.RemoveItems(change.Previous.Value.List); + } + break; + + // Shutdown the existing subscription and remove from the cache + case ChangeReason.Remove: + RemoveChildSubscription(change.Key); + _changeSetMergeTracker.RemoveItems(change.Current.List); + break; + } + } + } + + protected override void ChildOnNext(IChangeSet child, TKey parentKey) => + _changeSetMergeTracker.ProcessChangeSet(child, null); + + protected override void EmitChanges(IObserver> observer) => + _changeSetMergeTracker.EmitChanges(observer); + } } diff --git a/src/DynamicData/Cache/Internal/TransformManyAsync.cs b/src/DynamicData/Cache/Internal/TransformManyAsync.cs index 563e5082e..080c78b99 100644 --- a/src/DynamicData/Cache/Internal/TransformManyAsync.cs +++ b/src/DynamicData/Cache/Internal/TransformManyAsync.cs @@ -2,87 +2,93 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System.Reactive.Disposables; using System.Reactive.Linq; - using DynamicData.Internal; namespace DynamicData.Cache.Internal; -internal sealed class TransformManyAsync(IObservable> source, Func>>> selector, IEqualityComparer? equalityComparer, IComparer? comparer, Action>? errorHandler = null) +internal sealed class TransformManyAsync(IObservable> source, Func>>> transformer, IEqualityComparer? equalityComparer, IComparer? comparer, Action>? errorHandler = null) where TSource : notnull where TKey : notnull where TDestination : notnull where TDestinationKey : notnull { public IObservable> Run() => Observable.Create>( - observer => - { - var locker = InternalEx.NewLock(); - var cache = new Cache, TKey>(); - var parentUpdate = false; + observer => new Subscription(source, transformer, observer, equalityComparer, comparer, errorHandler)); - // This is manages all of the changes - var changeTracker = new ChangeSetMergeTracker(() => cache.Items, comparer, equalityComparer); + // Maintains state for a single subscription + private sealed class Subscription : CacheParentSubscription, TKey, IChangeSet, IChangeSet> + { + private readonly Cache, TKey> _cache = new(); + private readonly ChangeSetMergeTracker _changeSetMergeTracker; + public Subscription(IObservable> source, Func>>> transform, IObserver> observer, IEqualityComparer? equalityComparer, IComparer? comparer, Action>? errorHandler = null) + : base(observer) + { // Transform Helper - async Task>> InvokeSelector(TSource obj, TKey key) + async Task>> ErrorHandlingTransform(TSource obj, TKey key) { - if (errorHandler != null) + try { - try - { - return await selector(obj, key).ConfigureAwait(false); - } - catch (Exception e) - { - errorHandler.Invoke(new Error(e, obj, key)); - return Observable.Empty>(); - } + return await transform(obj, key).ConfigureAwait(false); + } + catch (Exception e) + { + errorHandler.Invoke(new Error(e, obj, key)); + return Observable.Empty>(); } - - return await selector(obj, key).ConfigureAwait(false); } - // Transformation Function: - // Create the Child Observable by invoking the async selector, appending the synchronize, and creating a new ChangeSetCache instance. - ChangeSetCache Transform_(TSource obj, TKey key) => - new(Observable.Defer(() => InvokeSelector(obj, key)).Synchronize(locker!)); + ChangeSetCache Transformer(TSource obj, TKey key) => + new(MakeChildObservable(Observable.Defer(() => transform(obj, key)))); + + ChangeSetCache SafeTransformer(TSource obj, TKey key) => + new(MakeChildObservable(Observable.Defer(() => ErrorHandlingTransform(obj, key)))); + + _changeSetMergeTracker = new(() => _cache.Items, comparer, equalityComparer); + + if (errorHandler is null) + { + CreateParentSubscription(source.Transform(Transformer)); + } + else + { + CreateParentSubscription(source.Transform(SafeTransformer)); + } + } - // Transform to a cache changeset of child caches, synchronize, clone changes to the local copy, and publish. - var shared = source - .Transform(Transform_) - .Synchronize(locker) - .Do( - changes => - { - cache.Clone(changes); - parentUpdate = true; - }) - .Publish(); + protected override void ParentOnNext(IChangeSet, TKey> changes) + { + // Process all the changes at once to preserve the changeset order + foreach (var change in changes.ToConcreteType()) + { + switch (change.Reason) + { + // Shutdown existing sub (if any) and create a new one that + // Will update the cache and emit the changes + case ChangeReason.Add or ChangeReason.Update: + _cache.AddOrUpdate(change.Current, change.Key); + AddChildSubscription(change.Current.Source, change.Key); + if (change.Previous.HasValue) + { + _changeSetMergeTracker.RemoveItems(change.Previous.Value.Cache.KeyValues); + } + break; - // Merge the child changeset changes together and apply to the tracker - // Emit the changeset if not currently handling a parent stream update - var subMergeMany = shared - .MergeMany(cacheChangeSet => cacheChangeSet.Source) - .SubscribeSafe( - changes => changeTracker.ProcessChangeSet(changes, !parentUpdate ? observer : null), - observer.OnError); + // Shutdown the existing subscription and remove from the cache + case ChangeReason.Remove: + _cache.Remove(change.Key); + RemoveChildSubscription(change.Key); + _changeSetMergeTracker.RemoveItems(change.Current.Cache.KeyValues); + break; + } + } + } - // When a source item is removed, all of its sub-items need to be removed - // Emit any pending changes - var subRemove = shared - .OnItemRemoved(changeSetCache => changeTracker.RemoveItems(changeSetCache.Cache.KeyValues), invokeOnUnsubscribe: false) - .OnItemUpdated((_, prev) => changeTracker.RemoveItems(prev.Cache.KeyValues)) - .SubscribeSafe( - _ => - { - changeTracker.EmitChanges(observer); - parentUpdate = false; - }, - observer.OnError, - observer.OnCompleted); + protected override void ChildOnNext(IChangeSet child, TKey parentKey) => + _changeSetMergeTracker.ProcessChangeSet(child); - return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove); - }); + protected override void EmitChanges(IObserver> observer) => + _changeSetMergeTracker.EmitChanges(observer); + } } diff --git a/src/DynamicData/Cache/Internal/TransformOnObservable.cs b/src/DynamicData/Cache/Internal/TransformOnObservable.cs index e0690254f..ea113414e 100644 --- a/src/DynamicData/Cache/Internal/TransformOnObservable.cs +++ b/src/DynamicData/Cache/Internal/TransformOnObservable.cs @@ -2,8 +2,6 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System.Diagnostics; -using System.Reactive.Disposables; using System.Reactive.Linq; using DynamicData.Internal; @@ -18,43 +16,21 @@ public IObservable> Run() => Observable.Create>(observer => new Subscription(source, transform, observer, transformOnRefresh)); // Maintains state for a single subscription - private sealed class Subscription : IDisposable + private sealed class Subscription : CacheParentSubscription> { -#if NET9_0_OR_GREATER - private readonly Lock _synchronize = new(); -#else - private readonly object _synchronize = new(); -#endif private readonly ChangeAwareCache _cache = new(); - private readonly KeyedDisposable _transformSubscriptions = new(); private readonly Func> _transform; - private readonly IDisposable _sourceSubscription; - private readonly IObserver> _observer; private readonly bool _transformOnRefresh; - private int _subscriptionCounter = 1; - private int _updateCounter; public Subscription(IObservable> source, Func> transform, IObserver> observer, bool transformOnRefresh) + : base(observer) { - _observer = observer; _transform = transform; _transformOnRefresh = transformOnRefresh; - _sourceSubscription = source - .Do(_ => IncrementUpdates()) - .Synchronize(_synchronize) - .SubscribeSafe(ProcessSourceChangeSet, observer.OnError, CheckCompleted); + CreateParentSubscription(source); } - public void Dispose() - { - lock (_synchronize) - { - _sourceSubscription.Dispose(); - _transformSubscriptions.Dispose(); - } - } - - private void ProcessSourceChangeSet(IChangeSet changes) + protected override void ParentOnNext(IChangeSet changes) { // Process all the changes at once to preserve the changeset order foreach (var change in changes.ToConcreteType()) @@ -64,19 +40,19 @@ private void ProcessSourceChangeSet(IChangeSet changes) // Shutdown existing sub (if any) and create a new one that // Will update the cache and emit the changes case ChangeReason.Add or ChangeReason.Update: - CreateTransformSubscription(change.Current, change.Key); + AddTransformSubscription(change.Current, change.Key); break; // Shutdown the existing subscription and remove from the cache case ChangeReason.Remove: - _transformSubscriptions.Remove(change.Key); _cache.Remove(change.Key); + RemoveChildSubscription(change.Key); break; case ChangeReason.Refresh: if (_transformOnRefresh) { - CreateTransformSubscription(change.Current, change.Key); + AddTransformSubscription(change.Current, change.Key); } else { @@ -86,62 +62,21 @@ private void ProcessSourceChangeSet(IChangeSet changes) break; } } - - // Emit any pending changes - EmitChanges(); } - private void IncrementUpdates() => Interlocked.Increment(ref _updateCounter); - - private void EmitChanges() - { - if (Interlocked.Decrement(ref _updateCounter) == 0) - { - var changes = _cache.CaptureChanges(); - if (changes.Count > 0) - { - _observer.OnNext(changes); - } - } - - Debug.Assert(_updateCounter >= 0, "Should never be negative"); - } + protected override void ChildOnNext(TDestination child, TKey parentKey) => + _cache.AddOrUpdate(child, parentKey); - private void CheckCompleted() + protected override void EmitChanges(IObserver> observer) { - if (Interlocked.Decrement(ref _subscriptionCounter) == 0) + var changes = _cache.CaptureChanges(); + if (changes.Count > 0) { - _observer.OnCompleted(); + observer.OnNext(changes); } - - Debug.Assert(_subscriptionCounter >= 0, "Should never be negative"); } - // Create the sub-observable that takes the result of the transformation, - // filters out unchanged values, and then updates the cache - private void CreateTransformSubscription(TSource obj, TKey key) - { - // Add a new subscription. Do first so cleanup of existing subs doesn't trigger OnCompleted. - Interlocked.Increment(ref _subscriptionCounter); - - // Create a container for the Disposable and add to the KeyedDisposable - var disposableContainer = _transformSubscriptions.Add(key, new SingleAssignmentDisposable()); - - // Create the transformation observable for the source item, filter unchanged, and update the cache - // Will Dispose immediately if OnCompleted fires upon subscription because OnCompleted disposes the container - // Remove the TransformSubscription if it completes because its not needed anymore - disposableContainer.Disposable = _transform(obj, key) - .DistinctUntilChanged() - .Do(_ => IncrementUpdates()) - .Synchronize(_synchronize) - .Finally(CheckCompleted) - .SubscribeSafe(val => TransformOnNext(val, key), _observer.OnError, () => _transformSubscriptions.Remove(key)); - } - - private void TransformOnNext(TDestination latestValue, TKey key) - { - _cache.AddOrUpdate(latestValue, key); - EmitChanges(); - } + private void AddTransformSubscription(TSource obj, TKey key) => + AddChildSubscription(MakeChildObservable(_transform(obj, key).DistinctUntilChanged()), key); } } diff --git a/src/DynamicData/Internal/CacheParentSubscription.cs b/src/DynamicData/Internal/CacheParentSubscription.cs new file mode 100644 index 000000000..b9a89e296 --- /dev/null +++ b/src/DynamicData/Internal/CacheParentSubscription.cs @@ -0,0 +1,134 @@ +// Copyright (c) 2011-2023 Roland Pheasant. All rights reserved. +// Roland Pheasant licenses this file to you under the MIT license. +// See the LICENSE file in the project root for full license information. + +using System.Diagnostics; +using System.Reactive.Disposables; +using System.Reactive.Linq; + +namespace DynamicData.Internal; + +/// +/// Base class for subscriptions that need to manage child subscriptions and emit updates +/// when either the parent or child gets a new value. +/// +/// Type of the Parent ChangeSet. +/// Type for the Parent ChangeSet Key. +/// Type for the Child Subscriptions. +/// Type for the Final Observable. +/// Observer to use for emitting events. +internal abstract class CacheParentSubscription(IObserver observer) : IDisposable + where TParent : notnull + where TKey : notnull + where TChild : notnull +{ +#if NET9_0_OR_GREATER + private readonly Lock _synchronize = new(); +#else + private readonly object _synchronize = new(); +#endif + private readonly KeyedDisposable _childSubscriptions = new(); + private readonly SingleAssignmentDisposable _parentSubscription = new(); + private readonly IObserver _observer = observer; + private int _subscriptionCounter = 1; + private int _updateCounter; + private bool _disposedValue; + + public void Dispose() + { + // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method + Dispose(disposing: true); + GC.SuppressFinalize(this); + } + + protected abstract void ParentOnNext(IChangeSet changes); + + protected abstract void ChildOnNext(TChild child, TKey parentKey); + + protected abstract void EmitChanges(IObserver observer); + + protected void AddChildSubscription(IObservable observable, TKey parentKey) + { + // Add a new subscription. Do first so cleanup of existing subs doesn't trigger OnCompleted. + Interlocked.Increment(ref _subscriptionCounter); + + // Create a container for the Disposable and add to the KeyedDisposable + var disposableContainer = _childSubscriptions.Add(parentKey, new SingleAssignmentDisposable()); + + // Create the subscription + // Will Dispose immediately if OnCompleted fires upon subscription because OnCompleted disposes the container + // Remove the child subscription if it completes because its not needed anymore + disposableContainer.Disposable = observable + .Finally(CheckCompleted) + .SubscribeSafe( + val => + { + ChildOnNext(val, parentKey); + ExitUpdate(); + }, + _observer.OnError, + () => RemoveChildSubscription(parentKey)); + } + + protected void RemoveChildSubscription(TKey parentKey) => _childSubscriptions.Remove(parentKey); + + protected void CreateParentSubscription(IObservable> source) => + _parentSubscription.Disposable = + source + .Synchronize(_synchronize) + .Do(_ => EnterUpdate()) + .SubscribeSafe( + changes => + { + ParentOnNext(changes); + ExitUpdate(); + }, + _observer.OnError, + CheckCompleted); + + protected virtual void Dispose(bool disposing) + { + if (!_disposedValue) + { + if (disposing) + { + lock (_synchronize) + { + _parentSubscription.Dispose(); + _childSubscriptions.Dispose(); + } + } + _disposedValue = true; + } + } + + // This must be called by the derived class on anything passed to AddChildSubscription + // Manual step so that the derived class has full control on where it is called + protected IObservable MakeChildObservable(IObservable observable) => + observable + .Synchronize(_synchronize) + .Do(_ => EnterUpdate()) + ; + + private void EnterUpdate() => Interlocked.Increment(ref _updateCounter); + + private void ExitUpdate() + { + if (Interlocked.Decrement(ref _updateCounter) == 0) + { + EmitChanges(_observer); + } + + Debug.Assert(_updateCounter >= 0, "Should never be negative"); + } + + private void CheckCompleted() + { + if (Interlocked.Decrement(ref _subscriptionCounter) == 0) + { + _observer.OnCompleted(); + } + + Debug.Assert(_subscriptionCounter >= 0, "Should never be negative"); + } +}