Ensure TransformOnObservable preserves ChangeSet order

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();
@@ -156,6 +170,38 @@ 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].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();
+    }
+
     public void Dispose()
     {
         _animalCache.Dispose();

src/DynamicData/Cache/Internal/TransformOnObservable.cs

@@ -13,52 +13,143 @@ 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));
+
+    // Maintains state for a single subscription
+    private sealed class Subscription : IDisposable
     {
-        var cache = new ChangeAwareCache<TDestination, TKey>();
-        var locker = InternalEx.NewLock();
-        var parentUpdate = false;
+#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> _transformSubscriptions = [];
+        private int _subscriptionCounter = 1;
+        private bool _sourceUpdate;
+
+        public Subscription(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource, TKey, IObservable<TDestination>> transform, IObserver<IChangeSet<TDestination, TKey>> observer)
+        {
+            _observer = observer;
+            _transform = transform;
+            _sourceSubscription = source
+                .Synchronize(_synchronize)
+                .SubscribeSafe(
+                    ProcessChangeSet,
+                    observer.OnError,
+                    CheckCompleted);
+        }
+
+        public void Dispose()
+        {
+            _sourceSubscription.Dispose();
+            _compositeDisposable.Dispose();
+            _transformSubscriptions.Values.ForEach(sub => sub.Dispose());
+        }
 
-        // Helper to emit any pending changes when appropriate
-        void EmitChanges(bool fromParent)
+        private void ProcessChangeSet(IChangeSet<TSource, TKey> changes)
         {
-            if (fromParent || !parentUpdate)
+            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())
             {
-                var changes = cache!.CaptureChanges();
+                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:
+                        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;
+
+                    // Let the downstream decide what this means
+                    case ChangeReason.Refresh:
+                        _cache.Refresh(change.Key);
+                        break;
+                }
+            }
+
+            // Emit any pending changes
+            EmitChanges(fromSource: true);
+        }
+
+        private void RemoveKey(TKey key)
+        {
+            if (_transformSubscriptions.TryGetValue(key, out var disposable))
+            {
+                disposable.Dispose();
+                _transformSubscriptions.Remove(key);
+            }
+        }
+
+        private void EmitChanges(bool fromSource)
+        {
+            if (fromSource || !_sourceUpdate)
+            {
+                var changes = _cache.CaptureChanges();
                 if (changes.Count > 0)
                 {
-                    observer.OnNext(changes);
+                    _observer.OnNext(changes);
                 }
 
-                parentUpdate = false;
+                _sourceUpdate = false;
+            }
+        }
+
+        private void CheckCompleted()
+        {
+            if (Interlocked.Decrement(ref _subscriptionCounter) == 0)
+            {
+                _observer.OnCompleted();
             }
         }
 
         // 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)
+        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(locker!)
-                .Do(val => cache!.AddOrUpdate(val, key));
-
-        // Flag a parent update is happening once inside the lock
-        var shared = source
-            .Synchronize(locker!)
-            .Do(_ => parentUpdate = true)
-            .Publish();
-
-        // MergeMany automatically handles Add/Update/Remove and OnCompleted/OnError correctly
-        var subMerged = shared
-            .MergeMany(CreateSubObservable)
-            .SubscribeSafe(_ => EmitChanges(fromParent: false), observer.OnError, observer.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);
-
-        return new CompositeDisposable(shared.Connect(), subMerged, subRemove);
-    });
+                .Synchronize(_synchronize)
+                .Finally(CheckCompleted)
+                .SubscribeSafe(
+                    val => TransformOnNext(val, key),
+                    _observer.OnError);
+
+            // Add it to the Dictionary
+            _transformSubscriptions.Add(key, disposable);
+        }
+
+        private void TransformOnNext(TDestination latestValue, TKey key)
+        {
+            _cache.AddOrUpdate(latestValue, key);
+            EmitChanges(fromSource: false);
+        }
+    }
 }