DisposeMany() Rework (#755)

Co-authored-by: Roland Pheasant <roland_pheasant@hotmail.com>
Co-authored-by: Chris Pulman <chris.pulman@yahoo.com>

src/DynamicData.Benchmarks/Cache/DisposeMany_Cache.cs

@@ -0,0 +1,59 @@
+// Copyright (c) 2011-2019 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;
+
+using BenchmarkDotNet.Attributes;
+
+namespace DynamicData.Benchmarks.Cache
+{
+    [MemoryDiagnoser]
+    [MarkdownExporterAttribute.GitHub]
+    public class DisposeMany_Cache
+    {
+        [Benchmark]
+        [Arguments(1, 0)]
+        [Arguments(1, 1)]
+        [Arguments(10, 0)]
+        [Arguments(10, 1)]
+        [Arguments(10, 10)]
+        [Arguments(100, 0)]
+        [Arguments(100, 1)]
+        [Arguments(100, 10)]
+        [Arguments(100, 100)]
+        [Arguments(1_000, 0)]
+        [Arguments(1_000, 1)]
+        [Arguments(1_000, 10)]
+        [Arguments(1_000, 100)]
+        [Arguments(1_000, 1_000)]
+        public void AddsRemovesAndFinalization(int addCount, int removeCount)
+        {
+            using var source = new SourceCache<KeyedDisposable, int>(static item => item.Id);
+
+            using var subscription = source
+                .Connect()
+                .DisposeMany()
+                .Subscribe();
+
+            for(var i = 0; i < addCount; ++i)
+                source.AddOrUpdate(new KeyedDisposable(i));
+
+            for(var i = 0; i < removeCount; ++i)
+                source.RemoveKey(i);
+
+            subscription.Dispose();
+        }
+
+        private sealed class KeyedDisposable
+            : IDisposable
+        {
+            public KeyedDisposable(int id)
+                => Id = id;
+
+            public int Id { get; }
+
+            public void Dispose() { }
+        }
+    }
+}

src/DynamicData.Benchmarks/List/DisposeMany_List.cs

@@ -0,0 +1,49 @@
+// Copyright (c) 2011-2019 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;
+using System.Reactive.Disposables;
+
+using BenchmarkDotNet.Attributes;
+
+namespace DynamicData.Benchmarks.List
+{
+    [MemoryDiagnoser]
+    [MarkdownExporterAttribute.GitHub]
+    public class DisposeMany_List
+    {
+        [Benchmark]
+        [Arguments(1, 0)]
+        [Arguments(1, 1)]
+        [Arguments(10, 0)]
+        [Arguments(10, 1)]
+        [Arguments(10, 10)]
+        [Arguments(100, 0)]
+        [Arguments(100, 1)]
+        [Arguments(100, 10)]
+        [Arguments(100, 100)]
+        [Arguments(1_000, 0)]
+        [Arguments(1_000, 1)]
+        [Arguments(1_000, 10)]
+        [Arguments(1_000, 100)]
+        [Arguments(1_000, 1_000)]
+        public void AddsRemovesAndFinalization(int addCount, int removeCount)
+        {
+            using var source = new SourceList<IDisposable>();
+
+            using var subscription = source
+                .Connect()
+                .DisposeMany()
+                .Subscribe();
+
+            for(var i = 0; i < addCount; ++i)
+                source.Add(Disposable.Create(static () => { }));
+
+            while(source.Count > (addCount - removeCount))
+                source.RemoveAt(source.Count - 1);
+
+            subscription.Dispose();
+        }
+    }
+}

src/DynamicData.Tests/Cache/AutoRefreshFixture.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.ObjectModel;
 using System.Linq;
 using System.Reactive.Linq;
 

src/DynamicData.Tests/Cache/DisposeManyFixture.cs

@@ -1,87 +1,168 @@
 using System;
 using System.Linq;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
 
 using FluentAssertions;
 
 using Xunit;
 
 namespace DynamicData.Tests.Cache;
 
-public class DisposeManyFixture : IDisposable
+public sealed class DisposeManyFixture : IDisposable
 {
-    private readonly ChangeSetAggregator<DisposableObject, int> _results;
+    private readonly Subject<IChangeSet<DisposableObject, int>> _changeSetsSource;
 
-    private readonly ISourceCache<DisposableObject, int> _source;
+    private readonly SourceCache<DisposableObject, int> _itemsSource;
 
-    public DisposeManyFixture()
-    {
-        _source = new SourceCache<DisposableObject, int>(p => p.Id);
-        _results = new ChangeSetAggregator<DisposableObject, int>(_source.Connect().DisposeMany());
-    }
+    private readonly ChangeSetAggregator<DisposableObject, int> _results;
 
-    [Fact]
-    public void AddWillNotCallDispose()
+    public DisposeManyFixture()
     {
-        _source.AddOrUpdate(new DisposableObject(1));
-
-        _results.Messages.Count.Should().Be(1, "Should be 1 updates");
-        _results.Data.Count.Should().Be(1, "Should be 1 item in the cache");
-        _results.Data.Items.First().IsDisposed.Should().Be(false, "Should not be disposed");
+        _changeSetsSource = new();
+        _itemsSource = new(item => item.Id);
+        _results = new(Observable.Merge(_changeSetsSource, _itemsSource.Connect())
+            .DisposeMany()
+            .Do(onNext: changeSet =>
+                {
+                    foreach (var change in changeSet)
+                    {
+                        change.Current.IsDisposed.Should().BeFalse("items should not be disposed until after downstream notifications are processed");
+
+                        if (change.Previous.HasValue)
+                            change.Previous.Value.IsDisposed.Should().BeFalse("items should not be disposed until after downstream notifications are processed");
+                    }
+                },
+                onError: _ =>
+                {
+                    foreach(var item in _itemsSource.Items)
+                        item.IsDisposed.Should().BeFalse("items should not be disposed until after downstream notifications are processed");
+                },
+                onCompleted: () =>
+                {
+                    foreach(var item in _itemsSource.Items)
+                        item.IsDisposed.Should().BeFalse("items should not be disposed until after downstream notifications are processed");
+                }));
     }
 
     public void Dispose()
     {
-        _source.Dispose();
+        _changeSetsSource.Dispose();
+        _itemsSource.Dispose();
         _results.Dispose();
     }
 
     [Fact]
-    public void EverythingIsDisposedWhenStreamIsDisposed()
+    // Verifies https://github.com/reactivemarbles/DynamicData/issues/668
+    public void ErrorsArePropagated()
     {
-        _source.AddOrUpdate(Enumerable.Range(1, 10).Select(i => new DisposableObject(i)));
-        _source.Clear();
+        var error = new Exception("Test Exception");
+
+        var source = Observable.Throw<IChangeSet<object, object>>(error)
+            .DisposeMany();
+
+        FluentActions.Invoking(() => source.Subscribe()).Should().Throw<Exception>().Which.Should().BeSameAs(error);
 
-        _results.Messages.Count.Should().Be(2, "Should be 2 updates");
-        _results.Messages[1].All(d => d.Current.IsDisposed).Should().BeTrue();
+        var receivedError = null as Exception;
+        source.Subscribe(
+            onNext: static _ => { },
+            onError: error => receivedError = error);
+        receivedError.Should().BeSameAs(error);
     }
 
     [Fact]
-    public void RemoveWillCallDispose()
+    public void ItemsAreDisposedAfterRemovalOrReplacement()
     {
-        _source.AddOrUpdate(new DisposableObject(1));
-        _source.Remove(1);
+        var items = new[]
+        {
+            new DisposableObject(1),
+            new DisposableObject(2),
+            new DisposableObject(3),
+            new DisposableObject(4),
+            new DisposableObject(5),
+            new DisposableObject(1),
+            new DisposableObject(6),
+            new DisposableObject(7),
+            new DisposableObject(8)
+        };
+
+        // Exercise a variety of types of changesets.
+        _itemsSource.AddOrUpdate(items[0]); // Single add
+        _itemsSource.AddOrUpdate(items[1..5]); // Range add
+        _itemsSource.AddOrUpdate(items[5]); // Replace
+        _itemsSource.AddOrUpdate(items[5]); // Redundant update
+        _itemsSource.RemoveKey(4); // Single remove
+        _itemsSource.RemoveKeys(new[] { 1, 2 }); // Range remove
+        _itemsSource.Clear(); // Clear
+        _itemsSource.AddOrUpdate(items[6..9]);
+        _changeSetsSource.OnNext(new ChangeSet<DisposableObject, int>() // Refresh
+        {
+            new Change<DisposableObject, int>(
+                reason: ChangeReason.Refresh,
+                key: _itemsSource.Items.First().Id,
+                current: _itemsSource.Items.First())
+        });
+        _changeSetsSource.OnNext(new ChangeSet<DisposableObject, int>() // Move
+        {
+            new Change<DisposableObject, int>(
+                key: _itemsSource.Items.First().Id,
+                current: _itemsSource.Items.First(),
+                currentIndex: 1,
+                previousIndex: 0)
+        });
+
+        _results.Error.Should().BeNull();
+        _results.Messages.Count.Should().Be(10, "10 updates were made to the source");
+        _results.Data.Count.Should().Be(3, "3 items were not removed from the list");
+        _results.Data.Items.All(item => item.IsDisposed).Should().BeFalse("items remaining in the list should not be disposed");
+        items.Except(_results.Data.Items).All(item => item.IsDisposed).Should().BeTrue("items removed from the list should be disposed");
+    }
 
-        _results.Messages.Count.Should().Be(2, "Should be 2 updates");
-        _results.Data.Count.Should().Be(0, "Should be 0 items in the cache");
-        _results.Messages[1].First().Current.IsDisposed.Should().Be(true, "Should be disposed");
+    [Fact]
+    public void RemainingItemsAreDisposedAfterCompleted()
+    {
+        _itemsSource.AddOrUpdate(new[]
+        {
+            new DisposableObject(1),
+            new DisposableObject(2),
+            new DisposableObject(3)
+        });
+
+        _itemsSource.Dispose();
+        _changeSetsSource.OnCompleted();
+
+        _results.Error.Should().BeNull();
+        _results.Messages.Count.Should().Be(1, "1 update was made to the source");
+        _results.Data.Count.Should().Be(3, "3 items were not removed from the list");
+        _results.Data.Items.All(item => item.IsDisposed).Should().BeTrue("Items remaining in the list should be disposed");
     }
 
     [Fact]
-    public void UpdateWillCallDispose()
+    public void RemainingItemsAreDisposedAfterError()
     {
-        _source.AddOrUpdate(new DisposableObject(1));
-        _source.AddOrUpdate(new DisposableObject(1));
+        _itemsSource.AddOrUpdate(new DisposableObject(1));
+
+        var error = new Exception("Test Exception");
+        _changeSetsSource.OnError(error);
+
+        _itemsSource.AddOrUpdate(new DisposableObject(2));
 
-        _results.Messages.Count.Should().Be(2, "Should be 2 updates");
-        _results.Data.Count.Should().Be(1, "Should be 1 items in the cache");
-        _results.Messages[1].First().Current.IsDisposed.Should().Be(false, "Current should not be disposed");
-        _results.Messages[1].First().Previous.Value.IsDisposed.Should().Be(true, "Previous should be disposed");
+        _results.Error.Should().Be(error);
+        _results.Messages.Count.Should().Be(1, "1 update was made to the source");
+        _results.Data.Count.Should().Be(1, "1 item was not removed from the list");
+        _results.Data.Items.All(item => item.IsDisposed).Should().BeTrue("items remaining in the list should be disposed");
     }
 
     private class DisposableObject : IDisposable
     {
         public DisposableObject(int id)
-        {
-            Id = id;
-        }
+            => Id = id;
 
         public int Id { get; private set; }
 
         public bool IsDisposed { get; private set; }
 
         public void Dispose()
-        {
-            IsDisposed = true;
-        }
+            => IsDisposed = true;
     }
 }

src/DynamicData.Tests/Cache/EditDiffChangeSetOptionalFixture.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections;
 using System.Collections.Generic;
 using System.ComponentModel;
 using System.Diagnostics.CodeAnalysis;

src/DynamicData.Tests/Cache/EnumerableObservableToObservableChangeSetFixture.cs

@@ -3,7 +3,6 @@
 using System.Linq;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
-using System.Threading.Tasks;
 using DynamicData.Tests.Domain;
 
 using FluentAssertions;

src/DynamicData.Tests/Cache/FilterOnObservableFixture.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections;
 using System.Collections.Generic;
 using System.Linq;
 using System.Reactive.Linq;

src/DynamicData.Tests/Cache/InnerJoinFixture.cs

@@ -1,7 +1,4 @@
 using System;
-using System.Linq;
-
-using DynamicData.Kernel;
 
 using FluentAssertions;
 

src/DynamicData.Tests/Cache/ToObservableChangeSetFixture.cs

@@ -1,8 +1,6 @@
 using System;
 using System.Collections.Generic;
-using System.Linq;
 using System.Reactive.Linq;
-using System.Threading.Tasks;
 using DynamicData.Tests.Domain;
 
 using FluentAssertions;

src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs

@@ -6,8 +6,6 @@
 
 using FluentAssertions;
 
-using Xunit;
-
 namespace DynamicData.Tests.Cache;
 
 public class ToObservableChangeSetFixtureWithCompletion : IDisposable

src/DynamicData.Tests/Cache/ToObservableOptionalFixture.cs

@@ -3,10 +3,8 @@
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using System.Reactive.Linq;
-using System.Reactive.Subjects;
 using System.Threading.Tasks;
 using DynamicData.Kernel;
-using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;

src/DynamicData.Tests/Cache/TransformWithInlineUpdateFixture.cs

@@ -1,8 +1,5 @@
 using System;
 using System.Linq;
-using System.Reactive;
-using System.Reactive.Subjects;
-using DynamicData.Kernel;
 using DynamicData.Tests.Domain;
 
 using FluentAssertions;

src/DynamicData.Tests/Domain/Market.cs

@@ -1,7 +1,6 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
-using System.Diagnostics;
 using System.Linq;
 using System.Reactive.Linq;
 using DynamicData.Kernel;