diff --git a/src/DynamicData.Benchmarks/Cache/DisposeMany_Cache.cs b/src/DynamicData.Benchmarks/Cache/DisposeMany_Cache.cs
new file mode 100644
index 000000000..1e4f83e65
--- /dev/null
+++ b/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() { }
+        }
+    }
+}
diff --git a/src/DynamicData.Benchmarks/List/DisposeMany_List.cs b/src/DynamicData.Benchmarks/List/DisposeMany_List.cs
new file mode 100644
index 000000000..53cbe5d5e
--- /dev/null
+++ b/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();
+        }
+    }
+}
diff --git a/src/DynamicData.Tests/Cache/AutoRefreshFixture.cs b/src/DynamicData.Tests/Cache/AutoRefreshFixture.cs
index 123f19b31..e84c43ac5 100644
--- a/src/DynamicData.Tests/Cache/AutoRefreshFixture.cs
+++ b/src/DynamicData.Tests/Cache/AutoRefreshFixture.cs
@@ -1,5 +1,4 @@
 using System;
-using System.Collections.ObjectModel;
 using System.Linq;
 using System.Reactive.Linq;
 
diff --git a/src/DynamicData.Tests/Cache/DisposeManyFixture.cs b/src/DynamicData.Tests/Cache/DisposeManyFixture.cs
index 2a02b1827..f7296551a 100644
--- a/src/DynamicData.Tests/Cache/DisposeManyFixture.cs
+++ b/src/DynamicData.Tests/Cache/DisposeManyFixture.cs
@@ -1,5 +1,7 @@
 using System;
 using System.Linq;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
 
 using FluentAssertions;
 
@@ -7,81 +9,160 @@
 
 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;
     }
 }
diff --git a/src/DynamicData.Tests/Cache/EditDiffChangeSetOptionalFixture.cs b/src/DynamicData.Tests/Cache/EditDiffChangeSetOptionalFixture.cs
index c45eaf042..e814d0a7e 100644
--- a/src/DynamicData.Tests/Cache/EditDiffChangeSetOptionalFixture.cs
+++ b/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;
diff --git a/src/DynamicData.Tests/Cache/EnumerableObservableToObservableChangeSetFixture.cs b/src/DynamicData.Tests/Cache/EnumerableObservableToObservableChangeSetFixture.cs
index bf18bfaff..c99eafde9 100644
--- a/src/DynamicData.Tests/Cache/EnumerableObservableToObservableChangeSetFixture.cs
+++ b/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;
diff --git a/src/DynamicData.Tests/Cache/FilterOnObservableFixture.cs b/src/DynamicData.Tests/Cache/FilterOnObservableFixture.cs
index 5a384da64..2e3b148d5 100644
--- a/src/DynamicData.Tests/Cache/FilterOnObservableFixture.cs
+++ b/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;
diff --git a/src/DynamicData.Tests/Cache/InnerJoinFixture.cs b/src/DynamicData.Tests/Cache/InnerJoinFixture.cs
index 736644c47..8553a6248 100644
--- a/src/DynamicData.Tests/Cache/InnerJoinFixture.cs
+++ b/src/DynamicData.Tests/Cache/InnerJoinFixture.cs
@@ -1,7 +1,4 @@
 using System;
-using System.Linq;
-
-using DynamicData.Kernel;
 
 using FluentAssertions;
 
diff --git a/src/DynamicData.Tests/Cache/ToObservableChangeSetFixture.cs b/src/DynamicData.Tests/Cache/ToObservableChangeSetFixture.cs
index 6f2edbdd7..4bb441cc7 100644
--- a/src/DynamicData.Tests/Cache/ToObservableChangeSetFixture.cs
+++ b/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;
diff --git a/src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs b/src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs
index c72ebe112..1d594f990 100644
--- a/src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs
+++ b/src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs
@@ -6,8 +6,6 @@
 
 using FluentAssertions;
 
-using Xunit;
-
 namespace DynamicData.Tests.Cache;
 
 public class ToObservableChangeSetFixtureWithCompletion : IDisposable
diff --git a/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.cs b/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.cs
index 8023faaf4..509e93a58 100644
--- a/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.cs
+++ b/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;
diff --git a/src/DynamicData.Tests/Cache/TransformWithInlineUpdateFixture.cs b/src/DynamicData.Tests/Cache/TransformWithInlineUpdateFixture.cs
index fd184c5f7..cb32054c9 100644
--- a/src/DynamicData.Tests/Cache/TransformWithInlineUpdateFixture.cs
+++ b/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;
diff --git a/src/DynamicData.Tests/Domain/Market.cs b/src/DynamicData.Tests/Domain/Market.cs
index 1e8b25c47..e32439a58 100644
--- a/src/DynamicData.Tests/Domain/Market.cs
+++ b/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;
diff --git a/src/DynamicData.Tests/Issues/OnItemRemovedIssue.cs b/src/DynamicData.Tests/Issues/OnItemRemovedIssue.cs
index e3ea5b582..07d0e8e99 100644
--- a/src/DynamicData.Tests/Issues/OnItemRemovedIssue.cs
+++ b/src/DynamicData.Tests/Issues/OnItemRemovedIssue.cs
@@ -1,9 +1,6 @@
 using System;
 using System.Collections.Generic;
 using System.Collections.ObjectModel;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 using DynamicData.Binding;
 using Xunit;
 
diff --git a/src/DynamicData.Tests/List/DisposeManyFixture.cs b/src/DynamicData.Tests/List/DisposeManyFixture.cs
index 0c4542750..705e5a418 100644
--- a/src/DynamicData.Tests/List/DisposeManyFixture.cs
+++ b/src/DynamicData.Tests/List/DisposeManyFixture.cs
@@ -1,5 +1,8 @@
 using System;
 using System.Linq;
+using System.Reactive;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
 
 using FluentAssertions;
 
@@ -7,84 +10,146 @@
 
 namespace DynamicData.Tests.List;
 
-public class DisposeManyFixture : IDisposable
+public sealed class DisposeManyFixture : IDisposable
 {
-    private readonly ChangeSetAggregator<DisposableObject> _results;
+    private readonly Subject<IChangeSet<DisposableObject>> _changeSetsSource;
 
-    private readonly ISourceList<DisposableObject> _source;
+    private readonly SourceList<DisposableObject> _itemsSource;
 
-    public DisposeManyFixture()
-    {
-        _source = new SourceList<DisposableObject>();
-        _results = new ChangeSetAggregator<DisposableObject>(_source.Connect().DisposeMany());
-    }
+    private readonly ChangeSetAggregator<DisposableObject> _results;
 
-    [Fact]
-    public void AddWillNotCallDispose()
+    public DisposeManyFixture()
     {
-        _source.Add(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();
+        _results = new(Observable.Merge(_changeSetsSource, _itemsSource.Connect())
+            .DisposeMany()
+            .Do(onNext: changeSet =>
+                {
+                    foreach (var change in changeSet)
+                    {
+                        if (change.Item.Current is not null)
+                            change.Item.Current.IsDisposed.Should().BeFalse("items should not be disposed until after downstream notifications are processed");
+
+                        if (change.Item.Previous.HasValue)
+                            change.Item.Previous.Value.IsDisposed.Should().BeFalse("items should not be disposed until after downstream notifications are processed");
+
+                        if (change.Range is not null)
+                            foreach (var item in change.Range)
+                                item.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()
     {
-        var toadd = Enumerable.Range(1, 10).Select(i => new DisposableObject(i));
-        _source.AddRange(toadd);
-        _source.Clear();
+        var error = new Exception("Test Exception");
+
+        var source = Observable.Throw<IChangeSet<object>>(error)
+            .DisposeMany();
 
-        _results.Messages.Count.Should().Be(2, "Should be 2 updates");
+        FluentActions.Invoking(() => source.Subscribe()).Should().Throw<Exception>().Which.Should().BeSameAs(error);
 
-        var itemsCleared = _results.Messages[1].First().Range;
-        itemsCleared.All(d => d.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.Add(new DisposableObject(1));
-        _source.RemoveAt(0);
+        var items = Enumerable.Range(1, 10)
+            .Select(id => new DisposableObject(id))
+            .ToArray();
+
+        // Exercise a variety of types of changesets.
+        _itemsSource.Add(items[0]); // Trivial single add
+        _itemsSource.AddRange(items[1..3]); // Trivial range add
+        _itemsSource.Insert(index: 1, item: items[3]); // Non-trivial single add
+        _itemsSource.InsertRange(index: 2, items: items[4..6]); // Non-trivial range add
+        _itemsSource.RemoveAt(index: 3); // Single remove
+        _itemsSource.RemoveRange(index: 2, count: 2); // Range remove
+        _itemsSource.ReplaceAt(index: 1, item: items[6]); // Replace
+        _itemsSource.Move(1, 0); // Move
+        _itemsSource.Clear(); // Clear
+        _itemsSource.AddRange(items[7..10]);
+        _changeSetsSource.OnNext(new ChangeSet<DisposableObject>() // Refresh
+        {
+            new(ListChangeReason.Refresh, current: _itemsSource.Items.First(), index: 0)
+        });
+
+        _results.Exception.Should().BeNull();
+        _results.Messages.Count.Should().Be(11, "11 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().Item.Current.IsDisposed.Should().Be(true, "Should be disposed");
+    [Fact]
+    public void RemainingItemsAreDisposedAfterCompleted()
+    {
+        _itemsSource.AddRange(new[]
+        {
+            new DisposableObject(1),
+            new DisposableObject(2),
+            new DisposableObject(3),
+        });
+        _itemsSource.Dispose();
+        _changeSetsSource.OnCompleted();
+
+        _results.Exception.Should().BeNull();
+        _results.Messages.Count.Should().Be(1, "1 update was made to the list");
+        _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.Add(new DisposableObject(1));
-        _source.ReplaceAt(0, new DisposableObject(1));
+        _itemsSource.Add(new(1));
+        
+        var error = new Exception("Test Exception");
+        _changeSetsSource.OnError(error);
+
+        _itemsSource.Add(new(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().Item.Current.IsDisposed.Should().Be(false, "Current should not be disposed");
-        _results.Messages[1].First().Item.Previous.Value.IsDisposed.Should().Be(true, "Previous should be disposed");
+        _results.Exception.Should().Be(error);
+        _results.Messages.Count.Should().Be(1, "1 update was made to the list");
+        _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;
     }
 }
diff --git a/src/DynamicData.Tests/List/SortFixture.cs b/src/DynamicData.Tests/List/SortFixture.cs
index 54fbe70ab..37f01bc26 100644
--- a/src/DynamicData.Tests/List/SortFixture.cs
+++ b/src/DynamicData.Tests/List/SortFixture.cs
@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using System.Reactive.Subjects;
 using DynamicData.Binding;
diff --git a/src/DynamicData.Tests/List/ToCollectionFixture.cs b/src/DynamicData.Tests/List/ToCollectionFixture.cs
index e4969862f..a46aa790f 100644
--- a/src/DynamicData.Tests/List/ToCollectionFixture.cs
+++ b/src/DynamicData.Tests/List/ToCollectionFixture.cs
@@ -1,7 +1,6 @@
 using System;
 using System.Collections.Generic;
 using System.Reactive.Linq;
-using System.Text;
 using FluentAssertions;
 using Xunit;
 
diff --git a/src/DynamicData/Binding/BindingListEventsSuspender.cs b/src/DynamicData/Binding/BindingListEventsSuspender.cs
index e386dea79..720ba21f8 100644
--- a/src/DynamicData/Binding/BindingListEventsSuspender.cs
+++ b/src/DynamicData/Binding/BindingListEventsSuspender.cs
@@ -3,7 +3,6 @@
 // See the LICENSE file in the project root for full license information.
 
 #if SUPPORTS_BINDINGLIST
-using System;
 using System.ComponentModel;
 using System.Reactive.Disposables;
 
diff --git a/src/DynamicData/Binding/SortedBindingListAdaptor.cs b/src/DynamicData/Binding/SortedBindingListAdaptor.cs
index a19934b1e..d6fd5cb3b 100644
--- a/src/DynamicData/Binding/SortedBindingListAdaptor.cs
+++ b/src/DynamicData/Binding/SortedBindingListAdaptor.cs
@@ -3,10 +3,7 @@
 // See the LICENSE file in the project root for full license information.
 
 #if SUPPORTS_BINDINGLIST
-using System;
-using System.Collections.Generic;
 using System.ComponentModel;
-using System.Linq;
 
 namespace DynamicData.Binding
 {
diff --git a/src/DynamicData/Cache/Internal/DisposeMany.cs b/src/DynamicData/Cache/Internal/DisposeMany.cs
index 4f8d57ff4..11c8db0e9 100644
--- a/src/DynamicData/Cache/Internal/DisposeMany.cs
+++ b/src/DynamicData/Cache/Internal/DisposeMany.cs
@@ -2,59 +2,63 @@
 // 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;
 using System.Reactive.Linq;
 
-using DynamicData.Kernel;
-
 namespace DynamicData.Cache.Internal;
 
-internal sealed class DisposeMany<TObject, TKey>(IObservable<IChangeSet<TObject, TKey>> source, Action<TObject> removeAction)
+internal sealed class DisposeMany<TObject, TKey>(IObservable<IChangeSet<TObject, TKey>> source)
     where TObject : notnull
     where TKey : notnull
 {
-    private readonly Action<TObject> _removeAction = removeAction ?? throw new ArgumentNullException(nameof(removeAction));
+    private readonly IObservable<IChangeSet<TObject, TKey>> _source = source;
 
-    private readonly IObservable<IChangeSet<TObject, TKey>> _source = source ?? throw new ArgumentNullException(nameof(source));
+    public IObservable<IChangeSet<TObject, TKey>> Run()
+        => Observable.Create<IChangeSet<TObject, TKey>>(observer =>
+        {
+            var cachedItems = new Dictionary<TKey, TObject>();
 
-    public IObservable<IChangeSet<TObject, TKey>> Run() => Observable.Create<IChangeSet<TObject, TKey>>(
-            observer =>
-            {
-                var locker = new object();
-                var cache = new Cache<TObject, TKey>();
-                var subscriber = _source.Synchronize(locker).Do(changes => RegisterForRemoval(changes, cache), observer.OnError).SubscribeSafe(observer);
+            return _source.SubscribeSafe(Observer.Create<IChangeSet<TObject, TKey>>(
+                onNext: changeSet =>
+                {
+                    observer.OnNext(changeSet);
 
-                return Disposable.Create(
-                    () =>
+                    foreach (var change in changeSet.ToConcreteType())
                     {
-                        subscriber.Dispose();
-
-                        lock (locker)
+                        switch (change.Reason)
                         {
-                            cache.Items.ForEach(t => _removeAction(t));
-                            cache.Clear();
+                            case ChangeReason.Update:
+                                if (change.Previous.HasValue && !EqualityComparer<TObject>.Default.Equals(change.Current, change.Previous.Value))
+                                    (change.Previous.Value as IDisposable)?.Dispose();
+                                break;
+
+                            case ChangeReason.Remove:
+                                (change.Current as IDisposable)?.Dispose();
+                                break;
                         }
-                    });
-            });
+                    }
 
-    private void RegisterForRemoval(IChangeSet<TObject, TKey> changes, Cache<TObject, TKey> cache)
-    {
-        changes.ToConcreteType().ForEach(
-            change =>
-            {
-                switch (change.Reason)
+                    cachedItems.Clone(changeSet);
+                },
+                onError: error =>
                 {
-                    case ChangeReason.Update:
-                        // ReSharper disable once InconsistentlySynchronizedField
-                        change.Previous.IfHasValue(t => _removeAction(t));
-                        break;
-
-                    case ChangeReason.Remove:
-                        // ReSharper disable once InconsistentlySynchronizedField
-                        _removeAction(change.Current);
-                        break;
-                }
-            });
-        cache.Clone(changes);
+                    observer.OnError(error);
+
+                    ProcessFinalization(cachedItems);
+                },
+                onCompleted: () =>
+                {
+                    observer.OnCompleted();
+
+                    ProcessFinalization(cachedItems);
+                }));
+        });
+
+    private static void ProcessFinalization(Dictionary<TKey, TObject> cachedItems)
+    {
+        foreach (var pair in cachedItems)
+            (pair.Value as IDisposable)?.Dispose();
+
+        cachedItems.Clear();
     }
 }
diff --git a/src/DynamicData/Cache/Internal/MergeMany.cs b/src/DynamicData/Cache/Internal/MergeMany.cs
index 69f23fd5e..3233f3c1d 100644
--- a/src/DynamicData/Cache/Internal/MergeMany.cs
+++ b/src/DynamicData/Cache/Internal/MergeMany.cs
@@ -64,7 +64,15 @@ private sealed class SubscriptionCounter : IDisposable
 
         public void Finally() => CheckCompleted();
 
-        public void Dispose() => _subject.Dispose();
+        public void Dispose()
+        {
+            if (Interlocked.Exchange(ref _subscriptionCount, 0) != 0)
+            {
+                _subject.OnCompleted();
+            }
+
+            _subject.Dispose();
+        }
 
         private void CheckCompleted()
         {
diff --git a/src/DynamicData/Cache/Internal/SubscribeMany.cs b/src/DynamicData/Cache/Internal/SubscribeMany.cs
index ac2cc1796..fe7a5e11f 100644
--- a/src/DynamicData/Cache/Internal/SubscribeMany.cs
+++ b/src/DynamicData/Cache/Internal/SubscribeMany.cs
@@ -2,6 +2,7 @@
 // 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;
 
@@ -36,7 +37,13 @@ public IObservable<IChangeSet<TObject, TKey>> Run() => Observable.Create<IChange
             observer =>
             {
                 var published = _source.Publish();
-                var subscriptions = published.Transform((t, k) => _subscriptionFactory(t, k)).DisposeMany().Subscribe();
+                var subscriptions = published
+                    .Transform((t, k) => _subscriptionFactory(t, k))
+                    .DisposeMany()
+                    .SubscribeSafe(Observer.Create<IChangeSet<IDisposable, TKey>>(
+                        onNext: static _ => { },
+                        onError: observer.OnError,
+                        onCompleted: static () => { }));
 
                 return new CompositeDisposable(subscriptions, published.SubscribeSafe(observer), published.Connect());
             });
diff --git a/src/DynamicData/Cache/ObservableCacheEx.cs b/src/DynamicData/Cache/ObservableCacheEx.cs
index 48d0de534..cc9bb9813 100644
--- a/src/DynamicData/Cache/ObservableCacheEx.cs
+++ b/src/DynamicData/Cache/ObservableCacheEx.cs
@@ -1141,8 +1141,8 @@ public static IObservable<IChangeSet<TObject, TKey>> DeferUntilLoaded<TObject, T
     /// <summary>
     /// Disposes each item when no longer required.
     ///
-    /// Individual items are disposed when removed or replaced. All items
-    /// are disposed when the stream is disposed.
+    /// Individual items are disposed after removal or replacement changes have been sent downstream.
+    /// All items previously-published on the stream are disposed after the stream finalizes.
     /// </summary>
     /// <typeparam name="TObject">The type of the object.</typeparam>
     /// <typeparam name="TKey">The type of the key.</typeparam>
@@ -1158,13 +1158,7 @@ public static IObservable<IChangeSet<TObject, TKey>> DisposeMany<TObject, TKey>(
             throw new ArgumentNullException(nameof(source));
         }
 
-        return new DisposeMany<TObject, TKey>(
-            source,
-            t =>
-            {
-                var d = t as IDisposable;
-                d?.Dispose();
-            }).Run();
+        return new DisposeMany<TObject, TKey>(source).Run();
     }
 
     /// <summary>
diff --git a/src/DynamicData/EnumerableEx.cs b/src/DynamicData/EnumerableEx.cs
index c30953f43..93aa43f76 100644
--- a/src/DynamicData/EnumerableEx.cs
+++ b/src/DynamicData/EnumerableEx.cs
@@ -4,6 +4,7 @@
 
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
+using DynamicData.Cache.Internal;
 
 namespace DynamicData;
 
@@ -29,6 +30,11 @@ public static IObservable<IChangeSet<TObject, TKey>> AsObservableChangeSet<TObje
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(source);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (source is null)
         {
             throw new ArgumentNullException(nameof(source));
@@ -38,6 +44,7 @@ public static IObservable<IChangeSet<TObject, TKey>> AsObservableChangeSet<TObje
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Observable.Create<IChangeSet<TObject, TKey>>(
             obs =>
@@ -66,10 +73,14 @@ public static IObservable<IChangeSet<TObject, TKey>> AsObservableChangeSet<TObje
     public static IObservable<IChangeSet<TObject>> AsObservableChangeSet<TObject>(this IEnumerable<TObject> source, bool completable = false)
         where TObject : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(source);
+#else
         if (source is null)
         {
             throw new ArgumentNullException(nameof(source));
         }
+#endif
 
         return Observable.Create<IChangeSet<TObject>>(
             obs =>
diff --git a/src/DynamicData/List/Internal/DisposeMany.cs b/src/DynamicData/List/Internal/DisposeMany.cs
new file mode 100644
index 000000000..2bfececb4
--- /dev/null
+++ b/src/DynamicData/List/Internal/DisposeMany.cs
@@ -0,0 +1,76 @@
+// 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.Reactive;
+using System.Reactive.Linq;
+
+namespace DynamicData.List.Internal;
+
+internal sealed class DisposeMany<T>
+    where T : notnull
+{
+    private readonly IObservable<IChangeSet<T>> _source;
+
+    public DisposeMany(IObservable<IChangeSet<T>> source)
+        => _source = source;
+
+    public IObservable<IChangeSet<T>> Run()
+        => Observable.Create<IChangeSet<T>>(observer =>
+        {
+            var cachedItems = new List<T>();
+
+            return _source.SubscribeSafe(Observer.Create<IChangeSet<T>>(
+                onNext: changeSet =>
+                {
+                    observer.OnNext(changeSet);
+
+                    foreach (var change in changeSet)
+                    {
+                        switch (change.Reason)
+                        {
+                            case ListChangeReason.Clear:
+                                foreach (var item in cachedItems)
+                                    (item as IDisposable)?.Dispose();
+                                break;
+
+                            case ListChangeReason.Remove:
+                                (change.Item.Current as IDisposable)?.Dispose();
+                                break;
+
+                            case ListChangeReason.RemoveRange:
+                                foreach (var item in change.Range)
+                                    (item as IDisposable)?.Dispose();
+                                break;
+
+                            case ListChangeReason.Replace:
+                                if (change.Item.Previous.HasValue)
+                                    (change.Item.Previous.Value as IDisposable)?.Dispose();
+                                break;
+                        }
+                    }
+
+                    cachedItems.Clone(changeSet);
+                },
+                onError: error =>
+                {
+                    observer.OnError(error);
+
+                    ProcessFinalization(cachedItems);
+                },
+                onCompleted: () =>
+                {
+                    observer.OnCompleted();
+
+                    ProcessFinalization(cachedItems);
+                }));
+        });
+
+    private static void ProcessFinalization(List<T> cachedItems)
+    {
+        foreach (var item in cachedItems)
+            (item as IDisposable)?.Dispose();
+
+        cachedItems.Clear();
+    }
+}
diff --git a/src/DynamicData/List/Internal/SubscribeMany.cs b/src/DynamicData/List/Internal/SubscribeMany.cs
index 5e105ece7..2389d78ff 100644
--- a/src/DynamicData/List/Internal/SubscribeMany.cs
+++ b/src/DynamicData/List/Internal/SubscribeMany.cs
@@ -2,6 +2,7 @@
 // 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;
 
@@ -18,7 +19,13 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
             observer =>
             {
                 var shared = _source.Publish();
-                var subscriptions = shared.Transform(t => _subscriptionFactory(t)).DisposeMany().Subscribe();
+                var subscriptions = shared
+                    .Transform(t => _subscriptionFactory(t))
+                    .DisposeMany()
+                    .SubscribeSafe(Observer.Create<IChangeSet<IDisposable>>(
+                        onNext: static _ => { },
+                        onError: observer.OnError,
+                        onCompleted: static () => { }));
 
                 return new CompositeDisposable(subscriptions, shared.SubscribeSafe(observer), shared.Connect());
             });
diff --git a/src/DynamicData/List/ObservableListEx.cs b/src/DynamicData/List/ObservableListEx.cs
index aa2301329..d38e91775 100644
--- a/src/DynamicData/List/ObservableListEx.cs
+++ b/src/DynamicData/List/ObservableListEx.cs
@@ -594,20 +594,21 @@ public static IObservable<IChangeSet<T>> DeferUntilLoaded<T>(this IObservableLis
     /// <summary>
     /// Disposes each item when no longer required.
     ///
-    /// Individual items are disposed when removed or replaced. All items
-    /// are disposed when the stream is disposed.
+    /// Individual items are disposed after removal or replacement changes have been sent downstream.
+    /// All items previously-published on the stream are disposed after the stream finalizes.
     /// </summary>
     /// <typeparam name="T">The type of the object.</typeparam>
     /// <param name="source">The source.</param>
     /// <returns>A continuation of the original stream.</returns>
     /// <exception cref="System.ArgumentNullException">source.</exception>
     public static IObservable<IChangeSet<T>> DisposeMany<T>(this IObservable<IChangeSet<T>> source)
-        where T : notnull => source.OnItemRemoved(
-            t =>
-            {
-                var d = t as IDisposable;
-                d?.Dispose();
-            });
+        where T : notnull
+    {
+        if (source is null)
+            throw new ArgumentNullException(nameof(source));
+
+        return new DisposeMany<T>(source).Run();
+    }
 
     /// <summary>
     /// Selects distinct values from the source, using the specified value selector.
diff --git a/src/DynamicData/ObservableChangeSet.cs b/src/DynamicData/ObservableChangeSet.cs
index 8147e38ab..ebe523165 100644
--- a/src/DynamicData/ObservableChangeSet.cs
+++ b/src/DynamicData/ObservableChangeSet.cs
@@ -4,6 +4,7 @@
 
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
+using DynamicData.Cache.Internal;
 
 namespace DynamicData;
 
@@ -24,6 +25,11 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
@@ -33,6 +39,7 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Create(
             cache =>
@@ -55,6 +62,11 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
@@ -64,6 +76,7 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Observable.Create<IChangeSet<TObject, TKey>>(
             observer =>
@@ -96,6 +109,11 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
@@ -105,6 +123,7 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Create(async (cache, _) => await subscribe(cache).ConfigureAwait(false), keySelector);
     }
@@ -121,6 +140,11 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
@@ -130,6 +154,7 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Observable.Create<IChangeSet<TObject, TKey>>(
             async (observer, ct) =>
@@ -163,6 +188,11 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
@@ -172,6 +202,7 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Create((cache, _) => subscribe(cache), keySelector);
     }
@@ -188,6 +219,11 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
@@ -197,6 +233,7 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Observable.Create<IChangeSet<TObject, TKey>>(
             async (observer, ct) =>
@@ -239,6 +276,11 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
@@ -248,6 +290,7 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Observable.Create<IChangeSet<TObject, TKey>>(
             async observer =>
@@ -280,6 +323,11 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         where TObject : notnull
         where TKey : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+
+        ArgumentNullException.ThrowIfNull(keySelector);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
@@ -289,6 +337,7 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
         {
             throw new ArgumentNullException(nameof(keySelector));
         }
+#endif
 
         return Observable.Create<IChangeSet<TObject, TKey>>(
             async (observer, ct) =>
@@ -318,10 +367,14 @@ public static IObservable<IChangeSet<TObject, TKey>> Create<TObject, TKey>(Func<
     public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Action> subscribe)
         where T : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
         }
+#endif
 
         return Create<T>(
             list =>
@@ -340,10 +393,14 @@ public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Action>
     public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, IDisposable> subscribe)
         where T : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
         }
+#endif
 
         return Observable.Create<IChangeSet<T>>(
             observer =>
@@ -381,10 +438,14 @@ public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, IDisposa
     public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Task<IDisposable>> subscribe)
         where T : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
         }
+#endif
 
         return Create<T>((list, _) => subscribe(list));
     }
@@ -398,10 +459,14 @@ public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Task<IDi
     public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, CancellationToken, Task<IDisposable>> subscribe)
         where T : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
         }
+#endif
 
         return Observable.Create<IChangeSet<T>>(
             async (observer, ct) =>
@@ -442,10 +507,14 @@ public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Cancella
     public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Task<Action>> subscribe)
         where T : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
         }
+#endif
 
         return Create<T>(async (list, _) => await subscribe(list).ConfigureAwait(false));
     }
@@ -459,10 +528,14 @@ public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Task<Act
     public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, CancellationToken, Task<Action>> subscribe)
         where T : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
         }
+#endif
 
         return Observable.Create<IChangeSet<T>>(
             async (observer, ct) =>
@@ -502,10 +575,14 @@ public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Cancella
     public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Task> subscribe)
         where T : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
         }
+#endif
 
         return Observable.Create<IChangeSet<T>>(
             async observer =>
@@ -535,10 +612,14 @@ public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, Task> su
     public static IObservable<IChangeSet<T>> Create<T>(Func<ISourceList<T>, CancellationToken, Task> subscribe)
         where T : notnull
     {
+#if NET6_0_OR_GREATER
+        ArgumentNullException.ThrowIfNull(subscribe);
+#else
         if (subscribe is null)
         {
             throw new ArgumentNullException(nameof(subscribe));
         }
+#endif
 
         return Observable.Create<IChangeSet<T>>(
             async (observer, ct) =>
diff --git a/src/DynamicData/Platforms/net45/PFilter.cs b/src/DynamicData/Platforms/net45/PFilter.cs
index b1fec41cf..18b888d40 100644
--- a/src/DynamicData/Platforms/net45/PFilter.cs
+++ b/src/DynamicData/Platforms/net45/PFilter.cs
@@ -3,9 +3,6 @@
 // See the LICENSE file in the project root for full license information.
 
 #if P_LINQ
-using System;
-using System.Collections.Generic;
-using System.Linq;
 using System.Reactive.Linq;
 
 using DynamicData.Cache.Internal;
diff --git a/src/DynamicData/Platforms/net45/PSubscribeMany.cs b/src/DynamicData/Platforms/net45/PSubscribeMany.cs
index edd4852ee..6a018bd78 100644
--- a/src/DynamicData/Platforms/net45/PSubscribeMany.cs
+++ b/src/DynamicData/Platforms/net45/PSubscribeMany.cs
@@ -3,7 +3,6 @@
 // See the LICENSE file in the project root for full license information.
 
 #if P_LINQ
-using System;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 
diff --git a/src/DynamicData/Platforms/net45/PTransform.cs b/src/DynamicData/Platforms/net45/PTransform.cs
index 170d877a6..c155e80ff 100644
--- a/src/DynamicData/Platforms/net45/PTransform.cs
+++ b/src/DynamicData/Platforms/net45/PTransform.cs
@@ -3,9 +3,6 @@
 // See the LICENSE file in the project root for full license information.
 
 #if P_LINQ
-using System;
-using System.Collections.Generic;
-using System.Linq;
 using System.Reactive.Linq;
 
 using DynamicData.Kernel;
diff --git a/src/DynamicData/Platforms/net45/ParallelEx.cs b/src/DynamicData/Platforms/net45/ParallelEx.cs
index 5011814d1..13cb6cb41 100644
--- a/src/DynamicData/Platforms/net45/ParallelEx.cs
+++ b/src/DynamicData/Platforms/net45/ParallelEx.cs
@@ -3,10 +3,6 @@
 // See the LICENSE file in the project root for full license information.
 
 #if P_LINQ
-using System;
-using System.Collections.Generic;
-using System.Linq;
-
 // ReSharper disable once CheckNamespace
 namespace DynamicData.PLinq
 {
@@ -17,36 +13,20 @@ internal static class ParallelEx
     {
         internal static ParallelQuery<Change<TObject, TKey>> Parallelise<TObject, TKey>(this IChangeSet<TObject, TKey> source, ParallelisationOptions option)
             where TObject : notnull
-            where TKey : notnull
-        {
-            switch (option.Type)
+            where TKey : notnull => option.Type switch
             {
-                case ParallelType.Parallelise:
-                    return source.AsParallel();
-
-                case ParallelType.Ordered:
-                    return source.AsParallel().AsOrdered();
-
-                default:
-                    throw new ArgumentException("Should not parallelise!  Call ShouldParallelise() first");
-            }
-        }
+                ParallelType.Parallelise => source.AsParallel(),
+                ParallelType.Ordered => source.AsParallel().AsOrdered(),
+                _ => throw new ArgumentException("Should not parallelise!  Call ShouldParallelise() first"),
+            };
 
         internal static ParallelQuery<KeyValuePair<TKey, TObject>> Parallelise<TObject, TKey>(this IEnumerable<KeyValuePair<TKey, TObject>> source, ParallelisationOptions option)
-            where TKey : notnull
-        {
-            switch (option.Type)
+            where TKey : notnull => option.Type switch
             {
-                case ParallelType.Parallelise:
-                    return source.AsParallel();
-
-                case ParallelType.Ordered:
-                    return source.AsParallel().AsOrdered();
-
-                default:
-                    throw new ArgumentException("Should not parallelise!  Call ShouldParallelise() first");
-            }
-        }
+                ParallelType.Parallelise => source.AsParallel(),
+                ParallelType.Ordered => source.AsParallel().AsOrdered(),
+                _ => throw new ArgumentException("Should not parallelise!  Call ShouldParallelise() first"),
+            };
 
         internal static IEnumerable<T> Parallelise<T>(this IEnumerable<T> source, ParallelisationOptions option)
         {
@@ -81,16 +61,10 @@ internal static IEnumerable<T> Parallelise<T>(this IEnumerable<T> source, Parall
 
         internal static bool ShouldParallelise<TObject, TKey>(this IChangeSet<TObject, TKey> source, ParallelisationOptions option)
             where TObject : notnull
-            where TKey : notnull
-        {
-            return (option.Type == ParallelType.Parallelise || option.Type == ParallelType.Ordered) && (option.Threshold >= 0 && source.Count >= option.Threshold);
-        }
+            where TKey : notnull => (option.Type == ParallelType.Parallelise || option.Type == ParallelType.Ordered) && (option.Threshold >= 0 && source.Count >= option.Threshold);
 
         internal static bool ShouldParallelise<TObject, TKey>(this IEnumerable<KeyValuePair<TKey, TObject>> source, ParallelisationOptions option)
-            where TKey : notnull
-        {
-            return (option.Type == ParallelType.Parallelise || option.Type == ParallelType.Ordered) && (option.Threshold >= 0 && source.Skip(option.Threshold).Any());
-        }
+            where TKey : notnull => (option.Type == ParallelType.Parallelise || option.Type == ParallelType.Ordered) && (option.Threshold >= 0 && source.Skip(option.Threshold).Any());
     }
 }
 
diff --git a/src/DynamicData/Platforms/net45/ParallelOperators.cs b/src/DynamicData/Platforms/net45/ParallelOperators.cs
index 29222fcf7..b33258f20 100644
--- a/src/DynamicData/Platforms/net45/ParallelOperators.cs
+++ b/src/DynamicData/Platforms/net45/ParallelOperators.cs
@@ -3,8 +3,6 @@
 // See the LICENSE file in the project root for full license information.
 
 #if P_LINQ
-using System;
-
 using DynamicData.Kernel;
 
 // ReSharper disable once CheckNamespace