From 9a917beed2d473575bc15fe5f961b144d1eaefd5 Mon Sep 17 00:00:00 2001
From: "Darrin W. Cullop" <Darrin.Cullop@microsoft.com>
Date: Sat, 24 May 2025 14:06:04 -0700
Subject: [PATCH 1/8] Addresses #1007 by ensuring that all changes are
 processed in order instead of processing remove changes after the others.

---
 .../Cache/TransformOnObservableFixture.cs     |  31 ++++
 .../Cache/Internal/TransformOnObservable.cs   | 149 ++++++++++++++----
 2 files changed, 146 insertions(+), 34 deletions(-)

diff --git a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
index 8b4dc16ff..55861075f 100644
--- a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
+++ b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
@@ -156,6 +156,37 @@ public void ResultFailsIfSourceFails()
         results.Error.Should().Be(expectedError);
     }
 
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public void OrderOfChangesIsPreserved(bool removeFirst)
+    {
+        // Arrange
+        using var results = _animalCache.Connect().TransformOnObservable(Observable.Return).AsAggregator();
+        var firstReason = removeFirst ? ChangeReason.Remove : ChangeReason.Add;
+        var nextReason = !removeFirst ? ChangeReason.Remove : ChangeReason.Add;
+
+        // Act
+        _animalCache.Edit(updater =>
+        {
+            if (removeFirst)
+            {
+                updater.Clear();
+                updater.AddOrUpdate(_animalFaker.Generate(InitialCount));
+            }
+            else
+            {
+                updater.AddOrUpdate(_animalFaker.Generate(InitialCount));
+                updater.Clear();
+            }
+        });
+
+        // Assert
+        results.Messages.Count.Should().Be(2);
+        results.Messages[1].Take(InitialCount).All(change => change.Reason == firstReason).Should().BeTrue();
+        results.Messages[1].Skip(InitialCount).All(change => change.Reason == nextReason).Should().BeTrue();
+    }
+
     public void Dispose()
     {
         _animalCache.Dispose();
diff --git a/src/DynamicData/Cache/Internal/TransformOnObservable.cs b/src/DynamicData/Cache/Internal/TransformOnObservable.cs
index d9ec3ce08..c61b6b284 100644
--- a/src/DynamicData/Cache/Internal/TransformOnObservable.cs
+++ b/src/DynamicData/Cache/Internal/TransformOnObservable.cs
@@ -4,7 +4,7 @@
 
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
-using DynamicData.Internal;
+using System.Reactive.Subjects;
 
 namespace DynamicData.Cache.Internal;
 
@@ -15,50 +15,131 @@ internal sealed class TransformOnObservable<TSource, TKey, TDestination>(IObserv
 {
     public IObservable<IChangeSet<TDestination, TKey>> Run() => Observable.Create<IChangeSet<TDestination, TKey>>(observer =>
     {
-        var cache = new ChangeAwareCache<TDestination, TKey>();
+        var shutdownSubject = new Subject<TKey>();
+        var changeEmitter = new ChangeEmiter(observer);
         var locker = InternalEx.NewLock();
-        var parentUpdate = false;
+        var compositeDisposable = new CompositeDisposable(shutdownSubject);
 
-        // Helper to emit any pending changes when appropriate
-        void EmitChanges(bool fromParent)
+        // Create the sub-observable that takes the result of the transformation,
+        // filters out unchanged values, and then updates the cache
+        void CreateChildSubscription(TSource obj, TKey key)
         {
-            if (fromParent || !parentUpdate)
-            {
-                var changes = cache!.CaptureChanges();
-                if (changes.Count > 0)
-                {
-                    observer.OnNext(changes);
-                }
+            IDisposable? disposable = null;
+            var completed = false;
 
-                parentUpdate = false;
-            }
-        }
+            // Add a new subscription
+            changeEmitter.AddSubscription();
 
-        // Create the sub-observable that takes the result of the transformation,
-        // filters out unchanged values, and then updates the cache
-        IObservable<TDestination> CreateSubObservable(TSource obj, TKey key) =>
-            transform(obj, key)
+            // Create the subscription
+            disposable = transform(obj, key)
                 .DistinctUntilChanged()
+                .TakeUntil(shutdownSubject.Where(shutdownKey => EqualityComparer<TKey>.Default.Equals(key, shutdownKey)))
                 .Synchronize(locker!)
-                .Do(val => cache!.AddOrUpdate(val, key));
+                .Subscribe(
+                    val =>
+                    {
+                        changeEmitter.Cache.AddOrUpdate(val, key);
+                        changeEmitter.EmitChanges(fromSource: false);
+                    },
+                    () =>
+                    {
+                        if (disposable is not null)
+                        {
+                            compositeDisposable.Remove(disposable);
+                        }
+                        changeEmitter.OnCompleted();
+                        completed = true;
+                    });
+
+            // If not already completed, add it to the CompositeDisposable
+            if (!completed)
+            {
+                compositeDisposable.Add(disposable);
+            }
+        }
 
-        // Flag a parent update is happening once inside the lock
-        var shared = source
+        // Create a subscription to the source that processes the changes inside the lock
+        var subscription = source
             .Synchronize(locker!)
-            .Do(_ => parentUpdate = true)
-            .Publish();
+            .Subscribe(
+                changes =>
+                {
+                    // Flag a parent update is happening once inside the lock
+                    changeEmitter.MarkSourceUpdate();
+
+                    // Process all the changes at once to preserve the changeset order
+                    foreach (var change in changes.ToConcreteType())
+                    {
+                        switch (change.Reason)
+                        {
+                            // Create a subscription that will update the cache
+                            case ChangeReason.Add:
+                                CreateChildSubscription(change.Current, change.Key);
+                                break;
+
+                            // Shutdown the existing subscription and remove from the cache
+                            case ChangeReason.Remove:
+                                shutdownSubject.OnNext(change.Key);
+                                changeEmitter.Cache.Remove(change.Key);
+                                break;
+
+                            // Shutdown the existing subscription and create a new one
+                            case ChangeReason.Update:
+                                shutdownSubject.OnNext(change.Key);
+                                CreateChildSubscription(change.Current, change.Key);
+                                break;
+
+                            // Let the downstream decide what this means
+                            case ChangeReason.Refresh:
+                                changeEmitter.Cache.Refresh(change.Key);
+                                break;
+                        }
+                    }
 
-        // MergeMany automatically handles Add/Update/Remove and OnCompleted/OnError correctly
-        var subMerged = shared
-            .MergeMany(CreateSubObservable)
-            .SubscribeSafe(_ => EmitChanges(fromParent: false), observer.OnError, observer.OnCompleted);
+                    // Emit all of the changes
+                    changeEmitter.EmitChanges(fromSource: true);
+                },
+                observer.OnError,
+                changeEmitter.OnCompleted);
 
-        // Subscribe to the shared Observable to handle Remove events.  MergeMany will unsubscribe from the sub-observable,
-        // but the corresponding key value needs to be removed from the Cache so the remove is observed downstream.
-        var subRemove = shared
-            .OnItemRemoved((_, key) => cache!.Remove(key), invokeOnUnsubscribe: false)
-            .SubscribeSafe(_ => EmitChanges(fromParent: true), observer.OnError);
+        // Add the source subscription to the clean up list
+        compositeDisposable.Add(subscription);
 
-        return new CompositeDisposable(shared.Connect(), subMerged, subRemove);
+        // Return the single disposable that controls everything
+        return compositeDisposable;
     });
+
+    private class ChangeEmiter(IObserver<IChangeSet<TDestination, TKey>> observer)
+    {
+        private bool _sourceUpdate;
+        private int _subscriptionCounter = 1;
+
+        public ChangeAwareCache<TDestination, TKey> Cache { get; } = new();
+
+        public void MarkSourceUpdate() => _sourceUpdate = true;
+
+        public void EmitChanges(bool fromSource)
+        {
+            if (fromSource || !_sourceUpdate)
+            {
+                var changes = Cache.CaptureChanges();
+                if (changes.Count > 0)
+                {
+                    observer.OnNext(changes);
+                }
+
+                _sourceUpdate = false;
+            }
+        }
+
+        public void AddSubscription() => Interlocked.Increment(ref _subscriptionCounter);
+
+        public void OnCompleted()
+        {
+            if (Interlocked.Decrement(ref _subscriptionCounter) == 0)
+            {
+                observer.OnCompleted();
+            }
+        }
+    }
 }

From a0331b810577ac00a89fc03a9df25ad0460fc6c5 Mon Sep 17 00:00:00 2001
From: "Darrin W. Cullop" <Darrin.Cullop@microsoft.com>
Date: Sat, 24 May 2025 16:13:58 -0700
Subject: [PATCH 2/8] Refactor the code to make it less closurey

---
 .../Cache/Internal/TransformOnObservable.cs   | 207 +++++++++---------
 1 file changed, 106 insertions(+), 101 deletions(-)

diff --git a/src/DynamicData/Cache/Internal/TransformOnObservable.cs b/src/DynamicData/Cache/Internal/TransformOnObservable.cs
index c61b6b284..d5aa403ac 100644
--- a/src/DynamicData/Cache/Internal/TransformOnObservable.cs
+++ b/src/DynamicData/Cache/Internal/TransformOnObservable.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;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
-using System.Reactive.Subjects;
 
 namespace DynamicData.Cache.Internal;
 
@@ -13,133 +13,138 @@ internal sealed class TransformOnObservable<TSource, TKey, TDestination>(IObserv
     where TKey : notnull
     where TDestination : notnull
 {
-    public IObservable<IChangeSet<TDestination, TKey>> Run() => Observable.Create<IChangeSet<TDestination, TKey>>(observer =>
+    public IObservable<IChangeSet<TDestination, TKey>> Run() => Observable.Create<IChangeSet<TDestination, TKey>>(observer => new Subscription(source, transform, observer));
+
+    private sealed class Subscription : IDisposable
     {
-        var shutdownSubject = new Subject<TKey>();
-        var changeEmitter = new ChangeEmiter(observer);
-        var locker = InternalEx.NewLock();
-        var compositeDisposable = new CompositeDisposable(shutdownSubject);
+#if NET9_0_OR_GREATER
+        private readonly Lock _synchronize = new();
+#else
+        private readonly object _synchronize = new();
+#endif
+        private readonly ChangeAwareCache<TDestination, TKey> _cache = new();
+        private readonly CompositeDisposable _compositeDisposable = [];
+        private readonly Func<TSource, TKey, IObservable<TDestination>> _transform;
+        private readonly IDisposable _sourceSubscription;
+        private readonly IObserver<IChangeSet<TDestination, TKey>> _observer;
+        private readonly Dictionary<TKey, IDisposable> _keySubscriptions = new();
+        private int _subscriptionCounter = 1;
+        private bool _sourceUpdate;
 
-        // Create the sub-observable that takes the result of the transformation,
-        // filters out unchanged values, and then updates the cache
-        void CreateChildSubscription(TSource obj, TKey key)
+        public Subscription(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource, TKey, IObservable<TDestination>> transform, IObserver<IChangeSet<TDestination, TKey>> observer)
         {
-            IDisposable? disposable = null;
-            var completed = false;
+            _observer = observer;
+            _transform = transform;
+            _sourceSubscription = source
+                .Synchronize(_synchronize)
+                .SubscribeSafe(Observer.Create<IChangeSet<TSource, TKey>>(ProcessChangeSet, observer.OnError, OnCompleted));
+        }
 
-            // Add a new subscription
-            changeEmitter.AddSubscription();
+        public void Dispose()
+        {
+            _sourceSubscription.Dispose();
+            _compositeDisposable.Dispose();
+            _keySubscriptions.Values.ForEach(sub => sub.Dispose());
+        }
 
-            // Create the subscription
-            disposable = transform(obj, key)
-                .DistinctUntilChanged()
-                .TakeUntil(shutdownSubject.Where(shutdownKey => EqualityComparer<TKey>.Default.Equals(key, shutdownKey)))
-                .Synchronize(locker!)
-                .Subscribe(
-                    val =>
-                    {
-                        changeEmitter.Cache.AddOrUpdate(val, key);
-                        changeEmitter.EmitChanges(fromSource: false);
-                    },
-                    () =>
-                    {
-                        if (disposable is not null)
-                        {
-                            compositeDisposable.Remove(disposable);
-                        }
-                        changeEmitter.OnCompleted();
-                        completed = true;
-                    });
-
-            // If not already completed, add it to the CompositeDisposable
-            if (!completed)
+        private void ProcessChangeSet(IChangeSet<TSource, TKey> changes)
+        {
+            if (changes.Count == 0)
             {
-                compositeDisposable.Add(disposable);
+                return;
             }
-        }
 
-        // Create a subscription to the source that processes the changes inside the lock
-        var subscription = source
-            .Synchronize(locker!)
-            .Subscribe(
-                changes =>
-                {
-                    // Flag a parent update is happening once inside the lock
-                    changeEmitter.MarkSourceUpdate();
+            // Flag a source update is happening
+            _sourceUpdate = true;
 
-                    // Process all the changes at once to preserve the changeset order
-                    foreach (var change in changes.ToConcreteType())
-                    {
-                        switch (change.Reason)
-                        {
-                            // Create a subscription that will update the cache
-                            case ChangeReason.Add:
-                                CreateChildSubscription(change.Current, change.Key);
-                                break;
-
-                            // Shutdown the existing subscription and remove from the cache
-                            case ChangeReason.Remove:
-                                shutdownSubject.OnNext(change.Key);
-                                changeEmitter.Cache.Remove(change.Key);
-                                break;
-
-                            // Shutdown the existing subscription and create a new one
-                            case ChangeReason.Update:
-                                shutdownSubject.OnNext(change.Key);
-                                CreateChildSubscription(change.Current, change.Key);
-                                break;
-
-                            // Let the downstream decide what this means
-                            case ChangeReason.Refresh:
-                                changeEmitter.Cache.Refresh(change.Key);
-                                break;
-                        }
-                    }
-
-                    // Emit all of the changes
-                    changeEmitter.EmitChanges(fromSource: true);
-                },
-                observer.OnError,
-                changeEmitter.OnCompleted);
-
-        // Add the source subscription to the clean up list
-        compositeDisposable.Add(subscription);
-
-        // Return the single disposable that controls everything
-        return compositeDisposable;
-    });
-
-    private class ChangeEmiter(IObserver<IChangeSet<TDestination, TKey>> observer)
-    {
-        private bool _sourceUpdate;
-        private int _subscriptionCounter = 1;
+            // Process all the changes at once to preserve the changeset order
+            foreach (var change in changes.ToConcreteType())
+            {
+                switch (change.Reason)
+                {
+                    // Create a subscription that will update the cache
+                    case ChangeReason.Add:
+                        CreateTransformSubscription(change.Current, change.Key);
+                        break;
+
+                    // Shutdown the existing subscription and remove from the cache
+                    case ChangeReason.Remove:
+                        RemoveKey(change.Key);
+                        _cache.Remove(change.Key);
+                        break;
+
+                    // Shutdown the existing subscription and create a new one
+                    case ChangeReason.Update:
+                        RemoveKey(change.Key);
+                        CreateTransformSubscription(change.Current, change.Key);
+                        break;
+
+                    // Let the downstream decide what this means
+                    case ChangeReason.Refresh:
+                        _cache.Refresh(change.Key);
+                        break;
+                }
+            }
 
-        public ChangeAwareCache<TDestination, TKey> Cache { get; } = new();
+            // Emit all of the changes
+            EmitChanges(fromSource: true);
+        }
 
-        public void MarkSourceUpdate() => _sourceUpdate = true;
+        private void RemoveKey(TKey key)
+        {
+            if (_keySubscriptions.TryGetValue(key, out var disposable))
+            {
+                disposable.Dispose();
+                _keySubscriptions.Remove(key);
+            }
+        }
 
-        public void EmitChanges(bool fromSource)
+        private void EmitChanges(bool fromSource)
         {
             if (fromSource || !_sourceUpdate)
             {
-                var changes = Cache.CaptureChanges();
+                var changes = _cache.CaptureChanges();
                 if (changes.Count > 0)
                 {
-                    observer.OnNext(changes);
+                    _observer.OnNext(changes);
                 }
 
                 _sourceUpdate = false;
             }
         }
 
-        public void AddSubscription() => Interlocked.Increment(ref _subscriptionCounter);
-
-        public void OnCompleted()
+        private void OnCompleted()
         {
             if (Interlocked.Decrement(ref _subscriptionCounter) == 0)
             {
-                observer.OnCompleted();
+                _observer.OnCompleted();
             }
         }
+
+        // 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
+            Interlocked.Increment(ref _subscriptionCounter);
+
+            // Create the transformation observable for the source item
+            // Filter out unchanged values
+            // And update the cache with the latest value
+            var disposable = _transform(obj, key)
+                .DistinctUntilChanged()
+                .Synchronize(_synchronize)
+                .SubscribeSafe(Observer.Create<TDestination>(
+                    val =>
+                    {
+                        _cache.AddOrUpdate(val, key);
+                        EmitChanges(fromSource: false);
+                    },
+                    _observer.OnError,
+                    OnCompleted));
+
+            // Add it to the Dictionary
+            _keySubscriptions.Add(key, disposable);
+        }
     }
 }

From 8dad65966d4f2cc1efaf6300d94b58a15ba12538 Mon Sep 17 00:00:00 2001
From: "Darrin W. Cullop" <Darrin.Cullop@microsoft.com>
Date: Sat, 24 May 2025 17:13:40 -0700
Subject: [PATCH 3/8] More cleanup / tests

---
 .../Cache/TransformOnObservableFixture.cs     | 16 +++++-
 .../Cache/Internal/TransformOnObservable.cs   | 57 ++++++++++---------
 2 files changed, 46 insertions(+), 27 deletions(-)

diff --git a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
index 55861075f..baff768c6 100644
--- a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
+++ b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
@@ -141,13 +141,27 @@ IObservable<string> CreateChildObs(Animal a, int id) =>
         results.IsCompleted.Should().Be(completeSource && completeChildren);
     }
 
+    [Fact]
+    public void ResultFailsIfChildFails()
+    {
+        // Arrange
+        var expectedError = new Exception("Expected");
+        var throwObservable = Observable.Throw<IChangeSet<Animal, int>>(expectedError);
+
+        // Act
+        using var results = _animalCache.Connect().TransformOnObservable(_ => throwObservable).AsAggregator();
+
+        // Assert
+        results.Error.Should().Be(expectedError);
+    }
+
     [Fact]
     public void ResultFailsIfSourceFails()
     {
         // Arrange
         var expectedError = new Exception("Expected");
         var throwObservable = Observable.Throw<IChangeSet<Animal, int>>(expectedError);
-        using var results = _animalCache.Connect().Concat(throwObservable).TransformOnObservable(animal => Observable.Return(animal)).AsAggregator();
+        using var results = _animalCache.Connect().Concat(throwObservable).TransformOnObservable(Observable.Return).AsAggregator();
 
         // Act
         _animalCache.Dispose();
diff --git a/src/DynamicData/Cache/Internal/TransformOnObservable.cs b/src/DynamicData/Cache/Internal/TransformOnObservable.cs
index d5aa403ac..2bbbb3ea5 100644
--- a/src/DynamicData/Cache/Internal/TransformOnObservable.cs
+++ b/src/DynamicData/Cache/Internal/TransformOnObservable.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;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
+using DynamicData.Internal;
 
 namespace DynamicData.Cache.Internal;
 
@@ -13,8 +13,10 @@ internal sealed class TransformOnObservable<TSource, TKey, TDestination>(IObserv
     where TKey : notnull
     where TDestination : notnull
 {
-    public IObservable<IChangeSet<TDestination, TKey>> Run() => Observable.Create<IChangeSet<TDestination, TKey>>(observer => new Subscription(source, transform, observer));
+    public IObservable<IChangeSet<TDestination, TKey>> Run() =>
+        Observable.Create<IChangeSet<TDestination, TKey>>(observer => new Subscription(source, transform, observer));
 
+    // Maintains state for a single subscription
     private sealed class Subscription : IDisposable
     {
 #if NET9_0_OR_GREATER
@@ -27,7 +29,7 @@ private sealed class Subscription : IDisposable
         private readonly Func<TSource, TKey, IObservable<TDestination>> _transform;
         private readonly IDisposable _sourceSubscription;
         private readonly IObserver<IChangeSet<TDestination, TKey>> _observer;
-        private readonly Dictionary<TKey, IDisposable> _keySubscriptions = new();
+        private readonly Dictionary<TKey, IDisposable> _transformSubscriptions = [];
         private int _subscriptionCounter = 1;
         private bool _sourceUpdate;
 
@@ -37,14 +39,17 @@ public Subscription(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource,
             _transform = transform;
             _sourceSubscription = source
                 .Synchronize(_synchronize)
-                .SubscribeSafe(Observer.Create<IChangeSet<TSource, TKey>>(ProcessChangeSet, observer.OnError, OnCompleted));
+                .SubscribeSafe(
+                    ProcessChangeSet,
+                    observer.OnError,
+                    CheckCompleted);
         }
 
         public void Dispose()
         {
             _sourceSubscription.Dispose();
             _compositeDisposable.Dispose();
-            _keySubscriptions.Values.ForEach(sub => sub.Dispose());
+            _transformSubscriptions.Values.ForEach(sub => sub.Dispose());
         }
 
         private void ProcessChangeSet(IChangeSet<TSource, TKey> changes)
@@ -62,8 +67,9 @@ private void ProcessChangeSet(IChangeSet<TSource, TKey> changes)
             {
                 switch (change.Reason)
                 {
-                    // Create a subscription that will update the cache
-                    case ChangeReason.Add:
+                    // 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);
                         break;
 
@@ -73,12 +79,6 @@ private void ProcessChangeSet(IChangeSet<TSource, TKey> changes)
                         _cache.Remove(change.Key);
                         break;
 
-                    // Shutdown the existing subscription and create a new one
-                    case ChangeReason.Update:
-                        RemoveKey(change.Key);
-                        CreateTransformSubscription(change.Current, change.Key);
-                        break;
-
                     // Let the downstream decide what this means
                     case ChangeReason.Refresh:
                         _cache.Refresh(change.Key);
@@ -86,16 +86,16 @@ private void ProcessChangeSet(IChangeSet<TSource, TKey> changes)
                 }
             }
 
-            // Emit all of the changes
+            // Emit any pending changes
             EmitChanges(fromSource: true);
         }
 
         private void RemoveKey(TKey key)
         {
-            if (_keySubscriptions.TryGetValue(key, out var disposable))
+            if (_transformSubscriptions.TryGetValue(key, out var disposable))
             {
                 disposable.Dispose();
-                _keySubscriptions.Remove(key);
+                _transformSubscriptions.Remove(key);
             }
         }
 
@@ -113,7 +113,7 @@ private void EmitChanges(bool fromSource)
             }
         }
 
-        private void OnCompleted()
+        private void CheckCompleted()
         {
             if (Interlocked.Decrement(ref _subscriptionCounter) == 0)
             {
@@ -128,23 +128,28 @@ private void CreateTransformSubscription(TSource obj, TKey key)
             // Add a new subscription
             Interlocked.Increment(ref _subscriptionCounter);
 
+            // Clean up any previous subscriptions
+            RemoveKey(key);
+
             // Create the transformation observable for the source item
             // Filter out unchanged values
             // And update the cache with the latest value
             var disposable = _transform(obj, key)
                 .DistinctUntilChanged()
                 .Synchronize(_synchronize)
-                .SubscribeSafe(Observer.Create<TDestination>(
-                    val =>
-                    {
-                        _cache.AddOrUpdate(val, key);
-                        EmitChanges(fromSource: false);
-                    },
-                    _observer.OnError,
-                    OnCompleted));
+                .Finally(CheckCompleted)
+                .SubscribeSafe(
+                    val => TransformOnNext(val, key),
+                    _observer.OnError);
 
             // Add it to the Dictionary
-            _keySubscriptions.Add(key, disposable);
+            _transformSubscriptions.Add(key, disposable);
+        }
+
+        private void TransformOnNext(TDestination latestValue, TKey key)
+        {
+            _cache.AddOrUpdate(latestValue, key);
+            EmitChanges(fromSource: false);
         }
     }
 }

From 9efb8e3f9847aa9e68db1fad5398faf91492a463 Mon Sep 17 00:00:00 2001
From: "Darrin W. Cullop" <Darrin.Cullop@microsoft.com>
Date: Sat, 24 May 2025 17:33:23 -0700
Subject: [PATCH 4/8] More improvements

---
 src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
index baff768c6..e243032d3 100644
--- a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
+++ b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
@@ -197,6 +197,7 @@ public void OrderOfChangesIsPreserved(bool removeFirst)
 
         // Assert
         results.Messages.Count.Should().Be(2);
+        results.Messages[1].Count.Should().Be(InitialCount * 2);
         results.Messages[1].Take(InitialCount).All(change => change.Reason == firstReason).Should().BeTrue();
         results.Messages[1].Skip(InitialCount).All(change => change.Reason == nextReason).Should().BeTrue();
     }

From 530cc653b71c42be5cb54e8aeb4992a85f6b3b01 Mon Sep 17 00:00:00 2001
From: "Darrin W. Cullop" <Darrin.Cullop@microsoft.com>
Date: Sun, 25 May 2025 06:51:19 -0700
Subject: [PATCH 5/8] More cleanup

---
 .../Cache/TransformOnObservableFixture.cs        |  7 ++++---
 .../Cache/Internal/TransformOnObservable.cs      | 16 ++++------------
 2 files changed, 8 insertions(+), 15 deletions(-)

diff --git a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
index e243032d3..5867cf00f 100644
--- a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
+++ b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
@@ -177,8 +177,9 @@ public void OrderOfChangesIsPreserved(bool removeFirst)
     {
         // Arrange
         using var results = _animalCache.Connect().TransformOnObservable(Observable.Return).AsAggregator();
-        var firstReason = removeFirst ? ChangeReason.Remove : ChangeReason.Add;
-        var nextReason = !removeFirst ? ChangeReason.Remove : ChangeReason.Add;
+        (var firstReason, var nextReason, var expectedChanges) = removeFirst 
+            ? (ChangeReason.Remove, ChangeReason.Add, InitialCount * 2)
+            : (ChangeReason.Add, ChangeReason.Remove, InitialCount * 3);
 
         // Act
         _animalCache.Edit(updater =>
@@ -197,7 +198,7 @@ public void OrderOfChangesIsPreserved(bool removeFirst)
 
         // Assert
         results.Messages.Count.Should().Be(2);
-        results.Messages[1].Count.Should().Be(InitialCount * 2);
+        results.Messages[1].Count.Should().Be(expectedChanges);
         results.Messages[1].Take(InitialCount).All(change => change.Reason == firstReason).Should().BeTrue();
         results.Messages[1].Skip(InitialCount).All(change => change.Reason == nextReason).Should().BeTrue();
     }
diff --git a/src/DynamicData/Cache/Internal/TransformOnObservable.cs b/src/DynamicData/Cache/Internal/TransformOnObservable.cs
index 2bbbb3ea5..26ba266db 100644
--- a/src/DynamicData/Cache/Internal/TransformOnObservable.cs
+++ b/src/DynamicData/Cache/Internal/TransformOnObservable.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;
 
@@ -25,11 +24,10 @@ private sealed class Subscription : IDisposable
         private readonly object _synchronize = new();
 #endif
         private readonly ChangeAwareCache<TDestination, TKey> _cache = new();
-        private readonly CompositeDisposable _compositeDisposable = [];
+        private readonly Dictionary<TKey, IDisposable> _transformSubscriptions = [];
         private readonly Func<TSource, TKey, IObservable<TDestination>> _transform;
         private readonly IDisposable _sourceSubscription;
         private readonly IObserver<IChangeSet<TDestination, TKey>> _observer;
-        private readonly Dictionary<TKey, IDisposable> _transformSubscriptions = [];
         private int _subscriptionCounter = 1;
         private bool _sourceUpdate;
 
@@ -39,20 +37,16 @@ public Subscription(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource,
             _transform = transform;
             _sourceSubscription = source
                 .Synchronize(_synchronize)
-                .SubscribeSafe(
-                    ProcessChangeSet,
-                    observer.OnError,
-                    CheckCompleted);
+                .SubscribeSafe(ProcessSourceChangeSet, observer.OnError, CheckCompleted);
         }
 
         public void Dispose()
         {
             _sourceSubscription.Dispose();
-            _compositeDisposable.Dispose();
             _transformSubscriptions.Values.ForEach(sub => sub.Dispose());
         }
 
-        private void ProcessChangeSet(IChangeSet<TSource, TKey> changes)
+        private void ProcessSourceChangeSet(IChangeSet<TSource, TKey> changes)
         {
             if (changes.Count == 0)
             {
@@ -138,9 +132,7 @@ private void CreateTransformSubscription(TSource obj, TKey key)
                 .DistinctUntilChanged()
                 .Synchronize(_synchronize)
                 .Finally(CheckCompleted)
-                .SubscribeSafe(
-                    val => TransformOnNext(val, key),
-                    _observer.OnError);
+                .SubscribeSafe(val => TransformOnNext(val, key), _observer.OnError);
 
             // Add it to the Dictionary
             _transformSubscriptions.Add(key, disposable);

From 11a1366aff6e38ffc86ec6a49d723d3649e75f45 Mon Sep 17 00:00:00 2001
From: "Darrin W. Cullop" <Darrin.Cullop@microsoft.com>
Date: Sun, 25 May 2025 09:56:24 -0700
Subject: [PATCH 6/8] A bit more refactoring Created KeyedDisposable helper
 class More unit tests

---
 .../Cache/TransformOnObservableFixture.cs     | 33 +++++++
 .../Cache/Internal/TransformOnObservable.cs   | 86 +++++++++----------
 src/DynamicData/Internal/KeyedDisposable.cs   | 80 +++++++++++++++++
 3 files changed, 156 insertions(+), 43 deletions(-)
 create mode 100644 src/DynamicData/Internal/KeyedDisposable.cs

diff --git a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
index 5867cf00f..a85f1e061 100644
--- a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
+++ b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
@@ -1,7 +1,9 @@
 using System;
 using System.Linq;
 using System.Reactive;
+using System.Reactive.Concurrency;
 using System.Reactive.Linq;
+using System.Reactive.Subjects;
 using System.Threading.Tasks;
 using Bogus;
 using DynamicData.Kernel;
@@ -203,6 +205,37 @@ public void OrderOfChangesIsPreserved(bool removeFirst)
         results.Messages[1].Skip(InitialCount).All(change => change.Reason == nextReason).Should().BeTrue();
     }
 
+    [Fact]
+    public async Task SimultaneousUpdatesAreEmittedTogether()
+    {
+        // Arrange
+        using var subject = new Subject<Unit>();
+
+        IObservable<string> CreateChildObs(Animal a, int id) =>
+            Observable.Return($"{a.Name}-{id}")
+                .Concat(subject.ObserveOn(DefaultScheduler.Instance).Select(_ => a.Name).Take(1));
+
+        var shared = _animalCache.Connect().TransformOnObservable(CreateChildObs).Publish();
+        using var results = shared.AsAggregator();
+        var task = Task.Run(async () => await shared);
+        using var cleanup = shared.Connect();
+        _animalCache.Dispose();
+        subject.OnNext(Unit.Default);
+
+        // Act
+        await task;
+
+        // Assert
+        _animalResults.Data.Count.Should().Be(InitialCount);
+        results.Data.Count.Should().Be(_animalResults.Data.Count);
+        results.Summary.Overall.Adds.Should().Be(InitialCount);
+        results.Summary.Overall.Updates.Should().Be(InitialCount);
+        results.Messages.Count.Should().BeLessThan(InitialCount + 1, "At least some updates should be grouped");
+        results.Messages.Skip(1).All(message => message.All(change => change.Reason is ChangeReason.Update)).Should().BeTrue();
+        _animalCache.Items.ForEach(animal => results.Data.Lookup(animal.Id).Should().Be(Optional.Some(animal.Name)));
+    }
+
+
     public void Dispose()
     {
         _animalCache.Dispose();
diff --git a/src/DynamicData/Cache/Internal/TransformOnObservable.cs b/src/DynamicData/Cache/Internal/TransformOnObservable.cs
index 26ba266db..e0690254f 100644
--- a/src/DynamicData/Cache/Internal/TransformOnObservable.cs
+++ b/src/DynamicData/Cache/Internal/TransformOnObservable.cs
@@ -2,18 +2,20 @@
 // 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;
 
 namespace DynamicData.Cache.Internal;
 
-internal sealed class TransformOnObservable<TSource, TKey, TDestination>(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource, TKey, IObservable<TDestination>> transform)
+internal sealed class TransformOnObservable<TSource, TKey, TDestination>(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource, TKey, IObservable<TDestination>> transform, bool transformOnRefresh = false)
     where TSource : notnull
     where TKey : notnull
     where TDestination : notnull
 {
     public IObservable<IChangeSet<TDestination, TKey>> Run() =>
-        Observable.Create<IChangeSet<TDestination, TKey>>(observer => new Subscription(source, transform, observer));
+        Observable.Create<IChangeSet<TDestination, TKey>>(observer => new Subscription(source, transform, observer, transformOnRefresh));
 
     // Maintains state for a single subscription
     private sealed class Subscription : IDisposable
@@ -24,38 +26,36 @@ private sealed class Subscription : IDisposable
         private readonly object _synchronize = new();
 #endif
         private readonly ChangeAwareCache<TDestination, TKey> _cache = new();
-        private readonly Dictionary<TKey, IDisposable> _transformSubscriptions = [];
+        private readonly KeyedDisposable<TKey> _transformSubscriptions = new();
         private readonly Func<TSource, TKey, IObservable<TDestination>> _transform;
         private readonly IDisposable _sourceSubscription;
         private readonly IObserver<IChangeSet<TDestination, TKey>> _observer;
+        private readonly bool _transformOnRefresh;
         private int _subscriptionCounter = 1;
-        private bool _sourceUpdate;
+        private int _updateCounter;
 
-        public Subscription(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource, TKey, IObservable<TDestination>> transform, IObserver<IChangeSet<TDestination, TKey>> observer)
+        public Subscription(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource, TKey, IObservable<TDestination>> transform, IObserver<IChangeSet<TDestination, TKey>> observer, bool transformOnRefresh)
         {
             _observer = observer;
             _transform = transform;
+            _transformOnRefresh = transformOnRefresh;
             _sourceSubscription = source
+                .Do(_ => IncrementUpdates())
                 .Synchronize(_synchronize)
                 .SubscribeSafe(ProcessSourceChangeSet, observer.OnError, CheckCompleted);
         }
 
         public void Dispose()
         {
-            _sourceSubscription.Dispose();
-            _transformSubscriptions.Values.ForEach(sub => sub.Dispose());
+            lock (_synchronize)
+            {
+                _sourceSubscription.Dispose();
+                _transformSubscriptions.Dispose();
+            }
         }
 
         private void ProcessSourceChangeSet(IChangeSet<TSource, TKey> changes)
         {
-            if (changes.Count == 0)
-            {
-                return;
-            }
-
-            // Flag a source update is happening
-            _sourceUpdate = true;
-
             // Process all the changes at once to preserve the changeset order
             foreach (var change in changes.ToConcreteType())
             {
@@ -69,42 +69,42 @@ private void ProcessSourceChangeSet(IChangeSet<TSource, TKey> changes)
 
                     // Shutdown the existing subscription and remove from the cache
                     case ChangeReason.Remove:
-                        RemoveKey(change.Key);
+                        _transformSubscriptions.Remove(change.Key);
                         _cache.Remove(change.Key);
                         break;
 
-                    // Let the downstream decide what this means
                     case ChangeReason.Refresh:
-                        _cache.Refresh(change.Key);
+                        if (_transformOnRefresh)
+                        {
+                            CreateTransformSubscription(change.Current, change.Key);
+                        }
+                        else
+                        {
+                            // Let the downstream decide what this means
+                            _cache.Refresh(change.Key);
+                        }
                         break;
                 }
             }
 
             // Emit any pending changes
-            EmitChanges(fromSource: true);
+            EmitChanges();
         }
 
-        private void RemoveKey(TKey key)
-        {
-            if (_transformSubscriptions.TryGetValue(key, out var disposable))
-            {
-                disposable.Dispose();
-                _transformSubscriptions.Remove(key);
-            }
-        }
+        private void IncrementUpdates() => Interlocked.Increment(ref _updateCounter);
 
-        private void EmitChanges(bool fromSource)
+        private void EmitChanges()
         {
-            if (fromSource || !_sourceUpdate)
+            if (Interlocked.Decrement(ref _updateCounter) == 0)
             {
                 var changes = _cache.CaptureChanges();
                 if (changes.Count > 0)
                 {
                     _observer.OnNext(changes);
                 }
-
-                _sourceUpdate = false;
             }
+
+            Debug.Assert(_updateCounter >= 0, "Should never be negative");
         }
 
         private void CheckCompleted()
@@ -113,35 +113,35 @@ private void CheckCompleted()
             {
                 _observer.OnCompleted();
             }
+
+            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
+            // Add a new subscription.  Do first so cleanup of existing subs doesn't trigger OnCompleted.
             Interlocked.Increment(ref _subscriptionCounter);
 
-            // Clean up any previous subscriptions
-            RemoveKey(key);
+            // 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 out unchanged values
-            // And update the cache with the latest value
-            var disposable = _transform(obj, key)
+            // 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);
-
-            // Add it to the Dictionary
-            _transformSubscriptions.Add(key, disposable);
+                .SubscribeSafe(val => TransformOnNext(val, key), _observer.OnError, () => _transformSubscriptions.Remove(key));
         }
 
         private void TransformOnNext(TDestination latestValue, TKey key)
         {
             _cache.AddOrUpdate(latestValue, key);
-            EmitChanges(fromSource: false);
+            EmitChanges();
         }
     }
 }
diff --git a/src/DynamicData/Internal/KeyedDisposable.cs b/src/DynamicData/Internal/KeyedDisposable.cs
new file mode 100644
index 000000000..e5658ac99
--- /dev/null
+++ b/src/DynamicData/Internal/KeyedDisposable.cs
@@ -0,0 +1,80 @@
+// 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.
+
+namespace DynamicData.Internal;
+
+/// <summary>
+/// Manages Disposables by Key:
+/// 1) Adding a disposable with the same key will dispose/replace the previous one.
+/// 2) Adding when the container is Disposed will Dispose it immediately.
+/// </summary>
+/// <typeparam name="TKey">Type to use for the Key.</typeparam>
+internal sealed class KeyedDisposable<TKey> : IDisposable
+    where TKey : notnull
+{
+    private readonly Dictionary<TKey, IDisposable> _disposables = [];
+    private bool _disposedValue;
+
+    public int Count => _disposables.Count;
+
+    public IEnumerable<TKey> Keys => _disposables.Keys;
+
+    public bool ContainsKey(TKey key) => _disposables.ContainsKey(key);
+
+    public bool IsDisposed => _disposedValue;
+
+    public TDisposable Add<TDisposable>(TKey key, TDisposable disposable)
+        where TDisposable : IDisposable
+    {
+        disposable.ThrowArgumentNullExceptionIfNull(nameof(disposable));
+
+        if (!_disposedValue)
+        {
+            Remove(key);
+            _disposables.Add(key, disposable);
+        }
+        else
+        {
+            disposable.Dispose();
+        }
+
+        return disposable;
+    }
+
+    public void Remove(TKey key)
+    {
+#if NET6_0_OR_GREATER
+        if (_disposables.Remove(key, out var disposable))
+        {
+            disposable.Dispose();
+        }
+#else
+        if (_disposables.TryGetValue(key, out var disposable))
+        {
+            disposable.Dispose();
+            _disposables.Remove(key);
+        }
+#endif
+    }
+
+    public void Dispose()
+    {
+        // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method
+        Dispose(disposing: true);
+        GC.SuppressFinalize(this);
+    }
+
+    private void Dispose(bool disposing)
+    {
+        if (!_disposedValue)
+        {
+            _disposedValue = true;
+            if (disposing)
+            {
+                _disposables.Values.ForEach(d => d.Dispose());
+                _disposables.Clear();
+            }
+        }
+    }
+}

From cfff5b936e09d6688b60471314640912be3912ca Mon Sep 17 00:00:00 2001
From: "Darrin W. Cullop" <Darrin.Cullop@microsoft.com>
Date: Sun, 25 May 2025 10:27:16 -0700
Subject: [PATCH 7/8] Fix unit test

---
 src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
index a85f1e061..720b41009 100644
--- a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
+++ b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
@@ -213,7 +213,7 @@ public async Task SimultaneousUpdatesAreEmittedTogether()
 
         IObservable<string> CreateChildObs(Animal a, int id) =>
             Observable.Return($"{a.Name}-{id}")
-                .Concat(subject.ObserveOn(DefaultScheduler.Instance).Select(_ => a.Name).Take(1));
+                .Concat(subject.ObserveOn(TaskPoolScheduler.Default).Select(_ => a.Name).Take(1));
 
         var shared = _animalCache.Connect().TransformOnObservable(CreateChildObs).Publish();
         using var results = shared.AsAggregator();

From eafcbe0a14c51e0bfed0afce1dccc23d8d08e82b Mon Sep 17 00:00:00 2001
From: "Darrin W. Cullop" <Darrin.Cullop@microsoft.com>
Date: Sun, 25 May 2025 10:35:32 -0700
Subject: [PATCH 8/8] Remove bad test

---
 .../Cache/TransformOnObservableFixture.cs     | 31 -------------------
 1 file changed, 31 deletions(-)

diff --git a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
index 720b41009..cc37806b7 100644
--- a/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
+++ b/src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
@@ -205,37 +205,6 @@ public void OrderOfChangesIsPreserved(bool removeFirst)
         results.Messages[1].Skip(InitialCount).All(change => change.Reason == nextReason).Should().BeTrue();
     }
 
-    [Fact]
-    public async Task SimultaneousUpdatesAreEmittedTogether()
-    {
-        // Arrange
-        using var subject = new Subject<Unit>();
-
-        IObservable<string> CreateChildObs(Animal a, int id) =>
-            Observable.Return($"{a.Name}-{id}")
-                .Concat(subject.ObserveOn(TaskPoolScheduler.Default).Select(_ => a.Name).Take(1));
-
-        var shared = _animalCache.Connect().TransformOnObservable(CreateChildObs).Publish();
-        using var results = shared.AsAggregator();
-        var task = Task.Run(async () => await shared);
-        using var cleanup = shared.Connect();
-        _animalCache.Dispose();
-        subject.OnNext(Unit.Default);
-
-        // Act
-        await task;
-
-        // Assert
-        _animalResults.Data.Count.Should().Be(InitialCount);
-        results.Data.Count.Should().Be(_animalResults.Data.Count);
-        results.Summary.Overall.Adds.Should().Be(InitialCount);
-        results.Summary.Overall.Updates.Should().Be(InitialCount);
-        results.Messages.Count.Should().BeLessThan(InitialCount + 1, "At least some updates should be grouped");
-        results.Messages.Skip(1).All(message => message.All(change => change.Reason is ChangeReason.Update)).Should().BeTrue();
-        _animalCache.Items.ForEach(animal => results.Data.Lookup(animal.Id).Should().Be(Optional.Some(animal.Name)));
-    }
-
-
     public void Dispose()
     {
         _animalCache.Dispose();