diff --git a/.github/workflows/ci-build.yml b/.github/workflows/ci-build.yml
index ebafe1cf8..e5f9f3baf 100644
--- a/.github/workflows/ci-build.yml
+++ b/.github/workflows/ci-build.yml
@@ -21,7 +21,7 @@ jobs:
lfs: true
- name: Setup .NET (With cache)
- uses: actions/setup-dotnet@v5.0.1
+ uses: actions/setup-dotnet@v5.3.0
with:
dotnet-version: |
6.0.x
@@ -37,12 +37,20 @@ jobs:
**/global.json
**/nuget.config
- - name: NBGV
+ - name: Install (or update) nbgv tool
+ run: dotnet tool update --global nbgv
+
+ - name: Set NBGV cloud variables
+ run: nbgv cloud -a
+
+ - name: Expose NBGV version as step outputs
id: nbgv
- uses: dotnet/nbgv@v0.5.1
- with:
- setAllVars: true
-
+ shell: pwsh
+ run: |
+ $ErrorActionPreference = 'Stop'
+ "SemVer2=$env:NBGV_SemVer2" >> $env:GITHUB_OUTPUT
+ "PrereleaseVersion=$env:NBGV_PrereleaseVersion" >> $env:GITHUB_OUTPUT
+
- name: NuGet Restore
run: dotnet restore DynamicData.sln
working-directory: src
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 643ae46fd..50f8371d1 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -40,7 +40,7 @@ jobs:
Write-Host "OK: publishing from '$env:REF_NAME'."
- name: Setup .NET (With cache)
- uses: actions/setup-dotnet@v5.0.1
+ uses: actions/setup-dotnet@v5.3.0
with:
dotnet-version: |
6.0.x
@@ -56,11 +56,19 @@ jobs:
**/global.json
**/nuget.config
- - name: NBGV
+ - name: Install (or update) nbgv tool
+ run: dotnet tool update --global nbgv
+
+ - name: Set NBGV cloud variables
+ run: nbgv cloud -a
+
+ - name: Expose NBGV version as step outputs
id: nbgv
- uses: dotnet/nbgv@v0.5.1
- with:
- setAllVars: true
+ shell: pwsh
+ run: |
+ $ErrorActionPreference = 'Stop'
+ "SemVer2=$env:NBGV_SemVer2" >> $env:GITHUB_OUTPUT
+ "PrereleaseVersion=$env:NBGV_PrereleaseVersion" >> $env:GITHUB_OUTPUT
- name: Verify version matches branch policy
shell: pwsh
@@ -126,21 +134,24 @@ jobs:
if ($LASTEXITCODE -ne 0) { throw "dotnet nuget push failed for $($pkg.Name) (exit $LASTEXITCODE)." }
}
- - name: Changelog
- uses: glennawatson/ChangeLog@0464dd89b26f61fecf24b41d675f8ffdb11c4c3f # v1
- id: changelog
+ - name: Install GitReleaseNoteGenerator
+ run: dotnet tool install -g GitReleaseNoteGenerator
+
+ - name: Generate release notes
+ env:
+ GITHUB_TOKEN: ${{ github.token }}
+ RELEASE_VERSION: ${{ steps.nbgv.outputs.SemVer2 }}
+ shell: pwsh
+ run: git-release-notes --release-version "$env:RELEASE_VERSION" --output-file release-notes.md
- name: Create GitHub Release
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
TAG: ${{ steps.nbgv.outputs.SemVer2 }}
IS_PRERELEASE: ${{ steps.nbgv.outputs.PrereleaseVersion != '' }}
- BODY: ${{ steps.changelog.outputs.commitLog }}
shell: pwsh
run: |
$ErrorActionPreference = 'Stop'
- $notesPath = Join-Path $env:RUNNER_TEMP 'release-notes.md'
- Set-Content -Path $notesPath -Value $env:BODY -Encoding utf8 -NoNewline
- $cmd = @('release', 'create', $env:TAG, '--title', $env:TAG, '--notes-file', $notesPath, '--target', $env:GITHUB_SHA)
+ $cmd = @('release', 'create', $env:TAG, '--title', $env:TAG, '--notes-file', 'release-notes.md', '--target', $env:GITHUB_SHA)
if ($env:IS_PRERELEASE -eq 'true') { $cmd += '--prerelease' }
gh @cmd
diff --git a/src/Directory.Build.targets b/src/Directory.Build.targets
index 347ab888d..d341fbbbd 100644
--- a/src/Directory.Build.targets
+++ b/src/Directory.Build.targets
@@ -3,6 +3,13 @@
$(AssemblyName) ($(TargetFramework))
+
+
+
+ $([System.Text.RegularExpressions.Regex]::Replace(%(Filename), '\.[^\.]+$', '.cs'))
+
+
+
$(DefineConstants);P_LINQ;SUPPORTS_BINDINGLIST
diff --git a/src/DynamicData.Benchmarks/Cache/Sum_Cache.cs b/src/DynamicData.Benchmarks/Cache/Sum_Cache.cs
new file mode 100644
index 000000000..008192d9b
--- /dev/null
+++ b/src/DynamicData.Benchmarks/Cache/Sum_Cache.cs
@@ -0,0 +1,134 @@
+using System;
+using System.Collections.Generic;
+using System.Reactive.Subjects;
+
+using BenchmarkDotNet.Attributes;
+
+using DynamicData.Aggregation;
+
+namespace DynamicData.Benchmarks.Cache;
+
+[MemoryDiagnoser]
+[MarkdownExporterAttribute.GitHub]
+public class Sum_Cache
+{
+ private IReadOnlyList> _addChangeSets = null!;
+ private IReadOnlyList> _replaceChangeSets = null!;
+ private IReadOnlyList> _removeChangeSets = null!;
+ private IReadOnlyList> _refreshChangeSets = null!;
+
+ private IChangeSet- _seedAfterAdds = null!;
+ private IChangeSet
- _seedAfterReplaces = null!;
+
+ [Params(100, 500, 1_000, 10_000)]
+ public int Count { get; set; }
+
+ [GlobalSetup]
+ public void Setup()
+ {
+ var source = new ChangeAwareCache
- (capacity: Count);
+ var items = new Item[Count + 1];
+
+ var addChangeSets = new List>(capacity: Count);
+ for (var id = 1; id <= Count; ++id)
+ {
+ var item = new Item()
+ {
+ Id = id,
+ Value = id
+ };
+ items[id] = item;
+ source.Add(item, key: id);
+ addChangeSets.Add(source.CaptureChanges());
+ }
+ _addChangeSets = addChangeSets;
+
+ var addedItems = (Item[])items.Clone();
+
+ var replaceChangeSets = new List>(capacity: Count);
+ for (var id = 1; id <= Count; ++id)
+ {
+ var replacement = new Item()
+ {
+ Id = id,
+ Value = id * 2
+ };
+ items[id] = replacement;
+ source.AddOrUpdate(replacement, key: id);
+ replaceChangeSets.Add(source.CaptureChanges());
+ }
+ _replaceChangeSets = replaceChangeSets;
+
+ var refreshChangeSets = new List>(capacity: Count);
+ for (var id = 1; id <= Count; ++id)
+ {
+ // Mutate in place, then refresh - the scenario stateless aggregation cannot currently observe.
+ items[id].Value += 1;
+ source.Refresh(id);
+ refreshChangeSets.Add(source.CaptureChanges());
+ }
+ _refreshChangeSets = refreshChangeSets;
+
+ var removeChangeSets = new List>(capacity: Count);
+ for (var id = 1; id <= Count; ++id)
+ {
+ source.Remove(id);
+ removeChangeSets.Add(source.CaptureChanges());
+ }
+ _removeChangeSets = removeChangeSets;
+
+ // Replaces, refreshes, and removes only form a valid sequence for an operator that has already
+ // seen the items they refer to, so each of those runs gets seeded with the population as it stood
+ // beforehand. Collapsing the seed into a single change set keeps its cost off the measurement as
+ // far as possible: replaces follow on from the items that were added, while refreshes and removes
+ // follow on from the items that replaced them.
+ _seedAfterAdds = BuildSeed(addedItems);
+ _seedAfterReplaces = BuildSeed(items);
+ }
+
+ [Benchmark]
+ public void Adds() => Run(seed: null, _addChangeSets);
+
+ [Benchmark]
+ public void Replaces() => Run(_seedAfterAdds, _replaceChangeSets);
+
+ [Benchmark]
+ public void Refreshes() => Run(_seedAfterReplaces, _refreshChangeSets);
+
+ [Benchmark]
+ public void Removes() => Run(_seedAfterReplaces, _removeChangeSets);
+
+ private static IChangeSet
- BuildSeed(Item[] items)
+ {
+ var seed = new ChangeAwareCache
- (capacity: items.Length - 1);
+
+ for (var id = 1; id < items.Length; ++id)
+ seed.Add(items[id], key: id);
+
+ return seed.CaptureChanges();
+ }
+
+ private static void Run(IChangeSet
- ? seed, IReadOnlyList> changeSets)
+ {
+ using var source = new Subject>();
+
+ using var subscription = source
+ .Sum(static item => item.Value)
+ .Subscribe();
+
+ if (seed is not null)
+ source.OnNext(seed);
+
+ foreach (var changeSet in changeSets)
+ source.OnNext(changeSet);
+
+ source.OnCompleted();
+ }
+
+ private sealed class Item
+ {
+ public required int Id { get; init; }
+
+ public int Value { get; set; }
+ }
+}
diff --git a/src/DynamicData.Benchmarks/List/Sum_List.cs b/src/DynamicData.Benchmarks/List/Sum_List.cs
new file mode 100644
index 000000000..87a90c97e
--- /dev/null
+++ b/src/DynamicData.Benchmarks/List/Sum_List.cs
@@ -0,0 +1,131 @@
+using System;
+using System.Collections.Generic;
+using System.Reactive.Subjects;
+
+using BenchmarkDotNet.Attributes;
+
+using DynamicData.Aggregation;
+
+namespace DynamicData.Benchmarks.List;
+
+[MemoryDiagnoser]
+[MarkdownExporterAttribute.GitHub]
+public class Sum_List
+{
+ private IReadOnlyList> _addChangeSets = null!;
+ private IReadOnlyList> _replaceChangeSets = null!;
+ private IReadOnlyList> _removeChangeSets = null!;
+ private IReadOnlyList> _refreshChangeSets = null!;
+
+ private IChangeSet
- _seedAfterAdds = null!;
+ private IChangeSet
- _seedAfterReplaces = null!;
+
+ [Params(100, 500, 1_000, 10_000)]
+ public int Count { get; set; }
+
+ [GlobalSetup]
+ public void Setup()
+ {
+ var source = new ChangeAwareList
- (capacity: Count);
+ var items = new Item[Count];
+
+ var addChangeSets = new List>(capacity: Count);
+ for (var index = 0; index < Count; ++index)
+ {
+ items[index] = new Item()
+ {
+ Id = index + 1,
+ Value = index + 1
+ };
+ source.Add(items[index]);
+ addChangeSets.Add(source.CaptureChanges());
+ }
+ _addChangeSets = addChangeSets;
+
+ var addedItems = (Item[])items.Clone();
+
+ var replaceChangeSets = new List>(capacity: Count);
+ for (var index = 0; index < Count; ++index)
+ {
+ items[index] = new Item()
+ {
+ Id = index + 1,
+ Value = (index + 1) * 2
+ };
+ source[index] = items[index];
+ replaceChangeSets.Add(source.CaptureChanges());
+ }
+ _replaceChangeSets = replaceChangeSets;
+
+ var refreshChangeSets = new List>(capacity: Count);
+ for (var index = 0; index < Count; ++index)
+ {
+ // Mutate in place, then refresh - the scenario stateless aggregation cannot currently observe.
+ items[index].Value += 1;
+ source.RefreshAt(index);
+ refreshChangeSets.Add(source.CaptureChanges());
+ }
+ _refreshChangeSets = refreshChangeSets;
+
+ var removeChangeSets = new List>(capacity: Count);
+ for (var id = 1; id <= Count; ++id)
+ {
+ source.RemoveAt(source.Count - 1);
+ removeChangeSets.Add(source.CaptureChanges());
+ }
+ _removeChangeSets = removeChangeSets;
+
+ // Replaces, refreshes, and removes only form a valid sequence for an operator that has already
+ // seen the items they refer to, so each of those runs gets seeded with the population as it stood
+ // beforehand. Collapsing the seed into a single change set keeps its cost off the measurement as
+ // far as possible: replaces follow on from the items that were added, while refreshes and removes
+ // follow on from the items that replaced them.
+ _seedAfterAdds = BuildSeed(addedItems);
+ _seedAfterReplaces = BuildSeed(items);
+ }
+
+ [Benchmark]
+ public void Adds() => Run(seed: null, _addChangeSets);
+
+ [Benchmark]
+ public void Replaces() => Run(_seedAfterAdds, _replaceChangeSets);
+
+ [Benchmark]
+ public void Refreshes() => Run(_seedAfterReplaces, _refreshChangeSets);
+
+ [Benchmark]
+ public void Removes() => Run(_seedAfterReplaces, _removeChangeSets);
+
+ private static IChangeSet
- BuildSeed(Item[] items)
+ {
+ var seed = new ChangeAwareList
- (capacity: items.Length);
+
+ seed.AddRange(items);
+
+ return seed.CaptureChanges();
+ }
+
+ private static void Run(IChangeSet
- ? seed, IReadOnlyList> changeSets)
+ {
+ using var source = new Subject>();
+
+ using var subscription = source
+ .Sum(static item => item.Value)
+ .Subscribe();
+
+ if (seed is not null)
+ source.OnNext(seed);
+
+ foreach (var changeSet in changeSets)
+ source.OnNext(changeSet);
+
+ source.OnCompleted();
+ }
+
+ private sealed class Item
+ {
+ public required int Id { get; init; }
+
+ public int Value { get; set; }
+ }
+}
diff --git a/src/DynamicData.Tests/API/ApiApprovalTests.DynamicDataTests.DotNet8_0.verified.txt b/src/DynamicData.Tests/API/ApiApprovalTests.DynamicDataTests.DotNet8_0.verified.txt
index 0fbeaf4f4..92fa2bb0d 100644
--- a/src/DynamicData.Tests/API/ApiApprovalTests.DynamicDataTests.DotNet8_0.verified.txt
+++ b/src/DynamicData.Tests/API/ApiApprovalTests.DynamicDataTests.DotNet8_0.verified.txt
@@ -1,6 +1,4 @@
-[assembly: System.Runtime.CompilerServices.InternalsVisibleTo("DynamicData.Profile")]
-[assembly: System.Runtime.CompilerServices.InternalsVisibleTo("DynamicData.ReactiveUI")]
-[assembly: System.Runtime.CompilerServices.InternalsVisibleTo("DynamicData.Tests")]
+[assembly: System.Runtime.CompilerServices.InternalsVisibleTo("DynamicData.Tests")]
[assembly: System.Runtime.Versioning.TargetFramework(".NETCoreApp,Version=v8.0", FrameworkDisplayName=".NET 8.0")]
namespace DynamicData.Aggregation
{
@@ -1152,19 +1150,22 @@ namespace DynamicData
public static System.IObservable> Batch(this System.IObservable> source, System.TimeSpan timeSpan, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector)
+ where TObject : notnull
+ where TKey : notnull { }
public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
- public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState = false, System.Reactive.Concurrency.IScheduler? scheduler = null)
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
- public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, System.TimeSpan? timeOut = default, System.Reactive.Concurrency.IScheduler? scheduler = null)
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, System.TimeSpan? timeOut, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
- public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState = false, System.IObservable? timer = null, System.Reactive.Concurrency.IScheduler? scheduler = null)
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState, System.IObservable? timer, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
- public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState = false, System.TimeSpan? timeOut = default, System.Reactive.Concurrency.IScheduler? scheduler = null)
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState, System.TimeSpan? timeOut, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
public static System.IObservable> Bind<[System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembers(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.All)] TObject, TKey>(this System.IObservable>> source, System.Collections.Generic.IList targetList)
@@ -3110,4 +3111,4 @@ namespace DynamicData.Tests
public void Dispose() { }
protected virtual void Dispose(bool isDisposing) { }
}
-}
\ No newline at end of file
+}
diff --git a/src/DynamicData.Tests/API/ApiApprovalTests.DynamicDataTests.DotNet9_0.verified.txt b/src/DynamicData.Tests/API/ApiApprovalTests.DynamicDataTests.DotNet9_0.verified.txt
index ff1f94916..5e852c6d4 100644
--- a/src/DynamicData.Tests/API/ApiApprovalTests.DynamicDataTests.DotNet9_0.verified.txt
+++ b/src/DynamicData.Tests/API/ApiApprovalTests.DynamicDataTests.DotNet9_0.verified.txt
@@ -1,6 +1,4 @@
-[assembly: System.Runtime.CompilerServices.InternalsVisibleTo("DynamicData.Profile")]
-[assembly: System.Runtime.CompilerServices.InternalsVisibleTo("DynamicData.ReactiveUI")]
-[assembly: System.Runtime.CompilerServices.InternalsVisibleTo("DynamicData.Tests")]
+[assembly: System.Runtime.CompilerServices.InternalsVisibleTo("DynamicData.Tests")]
[assembly: System.Runtime.Versioning.TargetFramework(".NETCoreApp,Version=v9.0", FrameworkDisplayName=".NET 9.0")]
namespace DynamicData.Aggregation
{
@@ -1150,19 +1148,22 @@ namespace DynamicData
public static System.IObservable> Batch(this System.IObservable> source, System.TimeSpan timeSpan, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector)
+ where TObject : notnull
+ where TKey : notnull { }
public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
- public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, System.TimeSpan? timeOut = default, System.Reactive.Concurrency.IScheduler? scheduler = null)
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, System.TimeSpan? timeOut, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
- public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState = false, System.Reactive.Concurrency.IScheduler? scheduler = null)
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
- public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState = false, System.IObservable? timer = null, System.Reactive.Concurrency.IScheduler? scheduler = null)
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState, System.IObservable? timer, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
- public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState = false, System.TimeSpan? timeOut = default, System.Reactive.Concurrency.IScheduler? scheduler = null)
+ public static System.IObservable> BatchIf(this System.IObservable> source, System.IObservable pauseIfTrueSelector, bool initialPauseState, System.TimeSpan? timeOut, System.Reactive.Concurrency.IScheduler? scheduler = null)
where TObject : notnull
where TKey : notnull { }
public static System.IObservable> Bind<[System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembers(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.All)] TObject, TKey>(this System.IObservable>> source, System.Collections.Generic.IList targetList)
@@ -1964,6 +1965,14 @@ namespace DynamicData
where TDestination : notnull
where TSource : notnull
where TKey : notnull { }
+ public static System.IObservable> TransformAsync(this System.IObservable> source, System.Func, TKey, System.Threading.CancellationToken, System.Threading.Tasks.Task> transformFactory, DynamicData.TransformAsyncOptions options)
+ where TDestination : notnull
+ where TSource : notnull
+ where TKey : notnull { }
+ public static System.IObservable> TransformAsync(this System.IObservable> source, System.Func, TKey, System.Threading.CancellationToken, System.Threading.Tasks.Task> transformFactory, System.IObservable>? forceTransform = null)
+ where TDestination : notnull
+ where TSource : notnull
+ where TKey : notnull { }
public static System.IObservable> TransformImmutable(this System.IObservable> source, System.Func transformFactory)
where TDestination : notnull
where TSource : notnull
@@ -2108,6 +2117,14 @@ namespace DynamicData
where TDestination : notnull
where TSource : notnull
where TKey : notnull { }
+ public static System.IObservable> TransformSafeAsync(this System.IObservable> source, System.Func, TKey, System.Threading.CancellationToken, System.Threading.Tasks.Task> transformFactory, System.Action> errorHandler, DynamicData.TransformAsyncOptions options)
+ where TDestination : notnull
+ where TSource : notnull
+ where TKey : notnull { }
+ public static System.IObservable> TransformSafeAsync(this System.IObservable> source, System.Func, TKey, System.Threading.CancellationToken, System.Threading.Tasks.Task> transformFactory, System.Action> errorHandler, System.IObservable>? forceTransform = null)
+ where TDestination : notnull
+ where TSource : notnull
+ where TKey : notnull { }
public static System.IObservable, TKey>> TransformToTree(this System.IObservable> source, System.Func pivotOn, System.IObservable, bool>>? predicateChanged = null)
where TObject : class
where TKey : notnull { }
@@ -2481,6 +2498,9 @@ namespace DynamicData
public static System.IObservable> TransformAsync(this System.IObservable> source, System.Func, int, System.Threading.Tasks.Task> transformFactory, bool transformOnRefresh = false)
where TSource : notnull
where TDestination : notnull { }
+ public static System.IObservable> TransformAsync(this System.IObservable> source, System.Func, int, System.Threading.CancellationToken, System.Threading.Tasks.Task> transformFactory, bool transformOnRefresh = false)
+ where TSource : notnull
+ where TDestination : notnull { }
public static System.IObservable> TransformMany(this System.IObservable> source, System.Func> manySelector, System.Collections.Generic.IEqualityComparer? equalityComparer = null)
where TDestination : notnull
where TSource : notnull { }
diff --git a/src/DynamicData.Tests/AggregationTests/SumFixture.ForCache.cs b/src/DynamicData.Tests/AggregationTests/SumFixture.ForCache.cs
new file mode 100644
index 000000000..078a8f1b0
--- /dev/null
+++ b/src/DynamicData.Tests/AggregationTests/SumFixture.ForCache.cs
@@ -0,0 +1,517 @@
+using System;
+
+using DynamicData.Aggregation;
+using DynamicData.Tests.Domain;
+using DynamicData.Tests.Utilities;
+
+using FluentAssertions;
+
+using Xunit;
+
+namespace DynamicData.Tests.AggregationTests;
+
+public partial class SumFixture
+{
+ public class ForCache
+ {
+ [Theory]
+ [InlineData(1, 10)]
+ [InlineData(3, 60)]
+ public void ItemsAreAdded_SumReflectsAllItems(int itemCount, int expectedSum)
+ {
+ var ages = new[] { 10, 20, 30 };
+ using var source = new TestSourceCache(p => p.Name);
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().BeEmpty("no items have been added to the source");
+
+ // UUT Action
+ for (var i = 0; i < itemCount; i++)
+ {
+ source.AddOrUpdate(new Person(((char)('A' + i)).ToString(), ages[i]));
+ }
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().HaveCount(itemCount, "each AddOrUpdate should produce a new sum emission");
+ results.RecordedValues[^1].Should().Be(expectedSum, $"the sum of the first {itemCount} ages should be {expectedSum}");
+ }
+
+ [Theory]
+ [InlineData("A", 50)]
+ [InlineData("B", 40)]
+ [InlineData("C", 30)]
+ public void ItemIsRemoved_SumReflectsRemoval(string keyToRemove, int expectedSum)
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items")
+ .Which.Should().Be(60, "the sum of ages 10 + 20 + 30 is 60");
+
+ // UUT Action
+ source.Remove(keyToRemove);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().HaveCount(2, "one additional sum value should have been emitted after the removal");
+ results.RecordedValues[^1].Should().Be(expectedSum, $"removing '{keyToRemove}' should leave a sum of {expectedSum}");
+ }
+
+ [Fact]
+ public void ItemIsUpdated_SumReflectsNewValue()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items")
+ .Which.Should().Be(30, "the sum of ages 10 + 20 is 30");
+
+ // UUT Action: update "B" from age 20 to age 50 (same key, new value)
+ source.AddOrUpdate(new Person("B", 50));
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().HaveCount(2, "one additional sum value should have been emitted after the update");
+ results.RecordedValues[^1].Should().Be(60, "updating 'B' from 20 to 50 should change the sum from 30 to 60");
+ }
+
+ [Fact]
+ public void MultipleChangesInBatch_SingleSumEmitted()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.RecordedValues.Should().BeEmpty("no items have been added to the source");
+
+ // UUT Action: add 3 items in a single batch
+ source.Edit(updater =>
+ {
+ updater.AddOrUpdate(new Person("A", 10));
+ updater.AddOrUpdate(new Person("B", 20));
+ updater.AddOrUpdate(new Person("C", 30));
+ });
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().ContainSingle("a batched edit should produce exactly one sum emission")
+ .Which.Should().Be(60, "the sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void SourceIsEmpty_NoSumEmitted()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().BeEmpty("no items were added so no sum values should have been emitted");
+ }
+
+ [Fact]
+ public void AllItemsRemoved_SumReturnsToZero()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items")
+ .Which.Should().Be(60, "the sum of ages 10 + 20 + 30 is 60");
+
+ // UUT Action: remove all items in a single batch
+ source.Edit(updater => updater.Clear());
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().HaveCount(2, "one additional sum value should have been emitted after clearing");
+ results.RecordedValues[^1].Should().Be(0, "all items were removed so the sum should return to zero");
+ }
+
+ [Fact]
+ public void SourceCompletesAfterEmitting_CompletionPropagates()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing the pre-existing item")
+ .Which.Should().Be(10, "the sum of a single age of 10 is 10");
+
+ // UUT Action
+ source.Complete();
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeTrue("the source has completed");
+ }
+
+ [Fact]
+ public void SourceCompletesWithoutEmitting_CompletionPropagates()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.RecordedValues.Should().BeEmpty("no items were added to the source");
+
+ // UUT Action
+ source.Complete();
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeTrue("the source has completed");
+ results.RecordedValues.Should().BeEmpty("no items were added so no sum values should have been emitted");
+ }
+
+ [Fact]
+ public void SourceCompletesImmediately_InitialSumAndCompletionPropagate()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+ source.Complete();
+
+ // UUT Construction: source is already completed, with pre-existing items.
+ // Subscription should produce both an initial sum and a completion, synchronously.
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeTrue("the source was already completed at the time of subscription");
+ results.RecordedValues.Should().ContainSingle("an initial sum value should still be emitted, even when the source completes immediately upon subscription")
+ .Which.Should().Be(60, "the sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void SourceCompletesImmediatelyWithoutEmitting_CompletionPropagates()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.Complete();
+
+ // UUT Construction: source is already completed, with no items.
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeTrue("the source was already completed at the time of subscription");
+ results.RecordedValues.Should().BeEmpty("no items were added so no sum values should have been emitted");
+ }
+
+ [Fact]
+ public void SourceErrorsAfterEmitting_ErrorPropagates()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing the pre-existing item");
+
+ // UUT Action
+ var error = new Exception("Test error");
+ source.SetError(error);
+
+ results.Error.Should().BeSameAs(error, "the error from the source should propagate to the subscriber");
+ results.HasCompleted.Should().BeFalse("an error is not a completion");
+ }
+
+ [Fact]
+ public void SourceErrorsWithoutEmitting_ErrorPropagates()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.RecordedValues.Should().BeEmpty("no items were added to the source");
+
+ // UUT Action
+ var error = new Exception("Test error");
+ source.SetError(error);
+
+ results.Error.Should().BeSameAs(error, "the error from the source should propagate to the subscriber");
+ results.HasCompleted.Should().BeFalse("an error is not a completion");
+ results.RecordedValues.Should().BeEmpty("no items were added so no sum values should have been emitted");
+ }
+
+ [Fact]
+ public void SourceFailsImmediately_ErrorPropagates()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ var error = new Exception("Test error");
+ source.SetError(error);
+
+ // UUT Construction: source is already in error state.
+ // The error should propagate synchronously upon subscription.
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeSameAs(error, "the error from the source should propagate to the subscriber immediately upon subscription");
+ results.HasCompleted.Should().BeFalse("an error is not a completion");
+ }
+
+ [Fact]
+ public void NullableValuesAreTreatedAsZero()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", new int?(10), "F", null));
+ source.AddOrUpdate(new Person("B", null, "F", null));
+ source.AddOrUpdate(new Person("C", new int?(30), "F", null));
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(p => p.AgeNullable)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items")
+ .Which.Should().Be(40, "null values should be treated as zero, so the sum should be 10 + 0 + 30 = 40");
+ }
+
+ [Theory]
+ [InlineData(new[] { 10, 20, 30 }, 60)]
+ [InlineData(new[] { int.MaxValue }, int.MaxValue)]
+ [InlineData(new[] { int.MinValue }, int.MinValue)]
+ [InlineData(new[] { int.MaxValue, -1 }, int.MaxValue - 1)]
+ [InlineData(new[] { int.MinValue, 1 }, int.MinValue + 1)]
+ public void ItemsAreAdded_SumIsCorrect_ForInt(int[] ages, int expectedSum)
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ for (var i = 0; i < ages.Length; i++)
+ {
+ source.AddOrUpdate(new Person(((char)('A' + i)).ToString(), ages[i]));
+ }
+
+ using var subscription = source.Connect()
+ .Sum(p => p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(expectedSum, $"the int sum of [{string.Join(", ", ages)}] is {expectedSum}");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableInt()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", new int?(10), "F", null));
+ source.AddOrUpdate(new Person("B", new int?(20), "F", null));
+ source.AddOrUpdate(new Person("C", new int?(30), "F", null));
+
+ using var subscription = source.Connect()
+ .Sum(p => p.AgeNullable)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60, "the nullable int sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForLong()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ using var subscription = source.Connect()
+ .Sum(p => (long)p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60L, "the long sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableLong()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ using var subscription = source.Connect()
+ .Sum(p => (long?)p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60L, "the nullable long sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForDouble()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ using var subscription = source.Connect()
+ .Sum(p => (double)p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60.0, "the double sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableDouble()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ using var subscription = source.Connect()
+ .Sum(p => (double?)p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60.0, "the nullable double sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForDecimal()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ using var subscription = source.Connect()
+ .Sum(p => (decimal)p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60M, "the decimal sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableDecimal()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ using var subscription = source.Connect()
+ .Sum(p => (decimal?)p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60M, "the nullable decimal sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForFloat()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ using var subscription = source.Connect()
+ .Sum(p => (float)p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60F, "the float sum of ages 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableFloat()
+ {
+ using var source = new TestSourceCache(p => p.Name);
+
+ source.AddOrUpdate(new Person("A", 10));
+ source.AddOrUpdate(new Person("B", 20));
+ source.AddOrUpdate(new Person("C", 30));
+
+ using var subscription = source.Connect()
+ .Sum(p => (float?)p.Age)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60F, "the nullable float sum of ages 10 + 20 + 30 is 60");
+ }
+ }
+}
diff --git a/src/DynamicData.Tests/AggregationTests/SumFixture.ForList.cs b/src/DynamicData.Tests/AggregationTests/SumFixture.ForList.cs
new file mode 100644
index 000000000..9e9108602
--- /dev/null
+++ b/src/DynamicData.Tests/AggregationTests/SumFixture.ForList.cs
@@ -0,0 +1,413 @@
+using System;
+using System.Linq;
+
+using DynamicData.Aggregation;
+using DynamicData.Tests.Utilities;
+
+using FluentAssertions;
+
+using Xunit;
+
+namespace DynamicData.Tests.AggregationTests;
+
+public partial class SumFixture
+{
+ public class ForList
+ {
+ [Theory]
+ [InlineData(1, 10)]
+ [InlineData(3, 60)]
+ public void ItemsAreAdded_SumReflectsAllItems(int itemCount, int expectedSum)
+ {
+ var items = new[] { 10, 20, 30 };
+ using var source = new TestSourceList();
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().BeEmpty("no items have been added to the source");
+
+ // UUT Action
+ source.AddRange(items.Take(itemCount));
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().ContainSingle("an AddRange produces a single changeset")
+ .Which.Should().Be(expectedSum, $"the sum of the first {itemCount} items should be {expectedSum}");
+ }
+
+ [Theory]
+ [InlineData(0, 50)]
+ [InlineData(1, 40)]
+ [InlineData(2, 30)]
+ public void ItemIsRemoved_SumReflectsRemoval(int removalIndex, int expectedSum)
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items")
+ .Which.Should().Be(60, "the sum of items 10 + 20 + 30 is 60");
+
+ // UUT Action
+ source.RemoveAt(removalIndex);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().HaveCount(2, "one additional sum value should have been emitted after the removal");
+ results.RecordedValues[^1].Should().Be(expectedSum, $"removing item at index {removalIndex} should leave a sum of {expectedSum}");
+ }
+
+ [Fact]
+ public void ItemIsReplaced_SumReflectsReplacement()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items")
+ .Which.Should().Be(60, "the sum of items 10 + 20 + 30 is 60");
+
+ // UUT Action: replace item at index 1 (value 20) with 50
+ source.ReplaceAt(1, 50);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().HaveCount(2, "one additional sum value should have been emitted after the replacement");
+ results.RecordedValues[^1].Should().Be(90, "replacing 20 with 50 should change the sum from 60 to 90");
+ }
+
+ [Fact]
+ public void ItemsAreCleared_SumReturnsToZero()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items")
+ .Which.Should().Be(60, "the sum of items 10 + 20 + 30 is 60");
+
+ // UUT Action
+ source.Clear();
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().HaveCount(2, "one additional sum value should have been emitted after clearing");
+ results.RecordedValues[^1].Should().Be(0, "all items were removed so the sum should return to zero");
+ }
+
+ [Fact]
+ public void SourceIsEmpty_NoSumEmitted()
+ {
+ using var source = new TestSourceList();
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().BeEmpty("no items were added so no sum values should have been emitted");
+ }
+
+ [Fact]
+ public void SourceCompletesAfterEmitting_CompletionPropagates()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeFalse("the source can still publish notifications");
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items");
+
+ // UUT Action
+ source.Complete();
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeTrue("the source has completed");
+ }
+
+ [Fact]
+ public void SourceCompletesWithoutEmitting_CompletionPropagates()
+ {
+ using var source = new TestSourceList();
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.RecordedValues.Should().BeEmpty("no items were added to the source");
+
+ // UUT Action
+ source.Complete();
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeTrue("the source has completed");
+ results.RecordedValues.Should().BeEmpty("no items were added so no sum values should have been emitted");
+ }
+
+ [Fact]
+ public void SourceCompletesImmediately_InitialSumAndCompletionPropagate()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+ source.Complete();
+
+ // UUT Construction: source is already completed, with pre-existing items.
+ // Subscription should produce both an initial sum and a completion, synchronously.
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeTrue("the source was already completed at the time of subscription");
+ results.RecordedValues.Should().ContainSingle("an initial sum value should still be emitted, even when the source completes immediately upon subscription")
+ .Which.Should().Be(60, "the sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void SourceCompletesImmediatelyWithoutEmitting_CompletionPropagates()
+ {
+ using var source = new TestSourceList();
+
+ source.Complete();
+
+ // UUT Construction: source is already completed, with no items.
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.HasCompleted.Should().BeTrue("the source was already completed at the time of subscription");
+ results.RecordedValues.Should().BeEmpty("no items were added so no sum values should have been emitted");
+ }
+
+ [Fact]
+ public void SourceErrorsAfterEmitting_ErrorPropagates()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ // UUT Construction
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.RecordedValues.Should().ContainSingle("one changeset was published containing all pre-existing items");
+
+ // UUT Action
+ var error = new Exception("Test error");
+ source.SetError(error);
+
+ results.Error.Should().BeSameAs(error, "the error from the source should propagate to the subscriber");
+ results.HasCompleted.Should().BeFalse("an error is not a completion");
+ }
+
+ [Fact]
+ public void SourceFailsImmediately_ErrorPropagates()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+ var error = new Exception("Test error");
+ source.SetError(error);
+
+ // UUT Construction: source is already in error state.
+ // The error should propagate synchronously upon subscription.
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .ValidateSynchronization()
+ .RecordValues(out var results);
+
+ results.Error.Should().BeSameAs(error, "the error from the source should propagate to the subscriber immediately upon subscription");
+ results.HasCompleted.Should().BeFalse("an error is not a completion");
+ }
+
+ [Theory]
+ [InlineData(new[] { 10, 20, 30 }, 60)]
+ [InlineData(new[] { int.MaxValue }, int.MaxValue)]
+ [InlineData(new[] { int.MinValue }, int.MinValue)]
+ [InlineData(new[] { int.MaxValue, -1 }, int.MaxValue - 1)]
+ [InlineData(new[] { int.MinValue, 1 }, int.MinValue + 1)]
+ public void ItemsAreAdded_SumIsCorrect_ForInt(int[] values, int expectedSum)
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(values);
+
+ using var subscription = source.Connect()
+ .Sum(x => x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(expectedSum, $"the int sum of [{string.Join(", ", values)}] is {expectedSum}");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableInt()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (int?)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60, "the nullable int sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForLong()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (long)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60L, "the long sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableLong()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (long?)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60L, "the nullable long sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForDouble()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (double)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60.0, "the double sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableDouble()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (double?)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60.0, "the nullable double sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForDecimal()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (decimal)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60M, "the decimal sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableDecimal()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (decimal?)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60M, "the nullable decimal sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForFloat()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (float)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60F, "the float sum of items 10 + 20 + 30 is 60");
+ }
+
+ [Fact]
+ public void ItemsAreAdded_SumIsCorrect_ForNullableFloat()
+ {
+ using var source = new TestSourceList();
+
+ source.AddRange(new[] { 10, 20, 30 });
+
+ using var subscription = source.Connect()
+ .Sum(x => (float?)x)
+ .RecordValues(out var results);
+
+ results.RecordedValues[^1].Should().Be(60F, "the nullable float sum of items 10 + 20 + 30 is 60");
+ }
+ }
+}
diff --git a/src/DynamicData.Tests/AggregationTests/SumFixture.cs b/src/DynamicData.Tests/AggregationTests/SumFixture.cs
deleted file mode 100644
index 6fecbcabc..000000000
--- a/src/DynamicData.Tests/AggregationTests/SumFixture.cs
+++ /dev/null
@@ -1,227 +0,0 @@
-using System;
-
-using DynamicData.Aggregation;
-using DynamicData.Tests.Domain;
-
-using FluentAssertions;
-
-using Xunit;
-
-namespace DynamicData.Tests.AggregationTests;
-
-public class SumFixture : IDisposable
-{
- private readonly SourceCache _source;
-
- public SumFixture() => _source = new SourceCache(p => p.Name);
-
- [Fact]
- public void AddedItemsContributeToSum()
- {
- var sum = 0;
- double dev = 0;
-
- var accumulator = _source.Connect().Sum(p => p.Age).Subscribe(x => sum = x);
- var deviation = _source.Connect().StdDev(p => p.Age, (int)0).Subscribe(x => dev = x);
-
- _source.AddOrUpdate(new Person("A", 10));
- _source.AddOrUpdate(new Person("B", 20));
- _source.AddOrUpdate(new Person("C", 30));
-
- sum.Should().Be(60, "Accumulated value should be 60");
- dev.Should().Be(7.0710678118654755, "");
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumLong()
- {
- long sum = 0;
- double dev = 0;
-
- var accumulator = _source.Connect().Sum(p => Convert.ToInt64(p.Age)).Subscribe(x => sum = x);
- var deviation = _source.Connect().StdDev(p => p.Age, (long)0).Subscribe(x => dev = x);
-
- _source.AddOrUpdate(new Person("A", 10));
- _source.AddOrUpdate(new Person("B", 20));
- _source.AddOrUpdate(new Person("C", 30));
-
- sum.Should().Be(60, "Accumulated value should be 60");
- dev.Should().Be(7.0710678118654755, "");
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumFloat()
- {
- float sum = 0;
- double dev = 0;
-
- var accumulator = _source.Connect().Sum(p => Convert.ToSingle(p.Age)).Subscribe(x => sum = x);
- var deviation = _source.Connect().StdDev(p => p.Age, (float)0).Subscribe(x => dev = x);
-
- _source.AddOrUpdate(new Person("A", 10));
- _source.AddOrUpdate(new Person("B", 20));
- _source.AddOrUpdate(new Person("C", 30));
-
- sum.Should().Be(60, "Accumulated value should be 60");
- dev.Should().Be(7.0710678118654755, "");
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumDouble()
- {
- double sum = 0;
- double dev = 0;
-
- var accumulator = _source.Connect().Sum(p => Convert.ToDouble(p.Age)).Subscribe(x => sum = x);
- var deviation = _source.Connect().StdDev(p => p.Age, (double)0).Subscribe(x => dev = x);
-
- _source.AddOrUpdate(new Person("A", 10));
- _source.AddOrUpdate(new Person("B", 20));
- _source.AddOrUpdate(new Person("C", 30));
-
- sum.Should().Be(60, "Accumulated value should be 60");
- dev.Should().Be(7.0710678118654755, "");
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumDecimal()
- {
- decimal sum = 0;
- decimal dev = 0;
-
- var accumulator = _source.Connect().Sum(p => Convert.ToDecimal(p.Age)).Subscribe(x => sum = x);
- var deviation = _source.Connect().StdDev(p => p.Age, (decimal)0).Subscribe(x => dev = x);
-
- _source.AddOrUpdate(new Person("A", 10));
- _source.AddOrUpdate(new Person("B", 20));
- _source.AddOrUpdate(new Person("C", 30));
-
- sum.Should().Be(60, "Accumulated value should be 60");
- dev.Should().Be(7.0710678118654752440084436210M, "");
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumNullable()
- {
- var sum = 0;
-
- var accumulator = _source.Connect().Sum(p => p.AgeNullable).Subscribe(x => sum = x);
-
- _source.AddOrUpdate(new Person("A", new int?(10), "F", null));
- _source.AddOrUpdate(new Person("B", new int?(20), "F", null));
- _source.AddOrUpdate(new Person("C", new int?(30), "F", null));
-
- sum.Should().Be(60, "Accumulated value should be 60");
-
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumLongNullable()
- {
- long sum = 0;
-
- var accumulator = _source.Connect().Sum(p => (long?)(p.AgeNullable.HasValue ? Convert.ToInt64(p.AgeNullable) : default)).Subscribe(x => sum = x);
-
- _source.AddOrUpdate(new Person("A", new int?(10), "F", null));
- _source.AddOrUpdate(new Person("B", new int?(20), "F", null));
- _source.AddOrUpdate(new Person("C", new int?(30), "F", null));
-
- sum.Should().Be(60, "Accumulated value should be 60");
-
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumFloatNullable()
- {
- float sum = 0;
-
- var accumulator = _source.Connect().Sum(p => (float?)(p.AgeNullable.HasValue ? Convert.ToSingle(p.AgeNullable) : default)).Subscribe(x => sum = x);
-
- _source.AddOrUpdate(new Person("A", new int?(10), "F", null));
- _source.AddOrUpdate(new Person("B", new int?(20), "F", null));
- _source.AddOrUpdate(new Person("C", new int?(30), "F", null));
-
- sum.Should().Be(60, "Accumulated value should be 60");
-
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumDoubleNullable()
- {
- double sum = 0;
-
- var accumulator = _source.Connect().Sum(p => (double?)(p.AgeNullable.HasValue ? Convert.ToDouble(p.AgeNullable) : default)).Subscribe(x => sum = x);
-
- _source.AddOrUpdate(new Person("A", new int?(10), "F", null));
- _source.AddOrUpdate(new Person("B", new int?(20), "F", null));
- _source.AddOrUpdate(new Person("C", new int?(30), "F", null));
-
- sum.Should().Be(60, "Accumulated value should be 60");
-
- accumulator.Dispose();
- }
-
- [Fact]
- public void AddedItemsContributeToSumDecimalNullable()
- {
- decimal sum = 0;
-
- var accumulator = _source.Connect().Sum(p => (decimal?)(p.AgeNullable.HasValue ? Convert.ToDecimal(p.AgeNullable) : default)).Subscribe(x => sum = x);
-
- _source.AddOrUpdate(new Person("A", new int?(10), "F", null));
- _source.AddOrUpdate(new Person("B", new int?(20), "F", null));
- _source.AddOrUpdate(new Person("C", new int?(30), "F", null));
-
- sum.Should().Be(60, "Accumulated value should be 60");
-
- accumulator.Dispose();
- }
-
- public void Dispose() => _source.Dispose();
-
- [Fact]
- public void InlineChangeReEvaluatesTotals()
- {
- var sum = 0;
-
- var somepropChanged = _source.Connect().WhenValueChanged(p => p.Age);
-
- var accumulator = _source.Connect().Sum(p => p.Age).InvalidateWhen(somepropChanged).Subscribe(x => sum = x);
-
- var personb = new Person("B", 5);
- _source.AddOrUpdate(new Person("A", 10));
- _source.AddOrUpdate(personb);
- _source.AddOrUpdate(new Person("C", 30));
-
- sum.Should().Be(45, "Sum should be 45 after inline change");
-
- personb.Age = 20;
-
- sum.Should().Be(60, "Sum should be 60 after inline change");
- accumulator.Dispose();
- }
-
- [Fact]
- public void RemoveProduceCorrectResult()
- {
- var sum = 0;
-
- var accumulator = _source.Connect().Sum(p => p.Age).Subscribe(x => sum = x);
-
- _source.AddOrUpdate(new Person("A", 10));
- _source.AddOrUpdate(new Person("B", 20));
- _source.AddOrUpdate(new Person("C", 30));
-
- _source.Remove("A");
- sum.Should().Be(50, "Accumulated value should be 50 after remove");
- accumulator.Dispose();
- }
-}
diff --git a/src/DynamicData.Tests/Binding/WhenPropertyChangedBehaviorFixture.cs b/src/DynamicData.Tests/Binding/WhenPropertyChangedBehaviorFixture.cs
new file mode 100644
index 000000000..718df9537
--- /dev/null
+++ b/src/DynamicData.Tests/Binding/WhenPropertyChangedBehaviorFixture.cs
@@ -0,0 +1,309 @@
+// Copyright (c) 2011-2025 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.Collections.Generic;
+using System.ComponentModel;
+using System.Linq;
+using DynamicData.Binding;
+using DynamicData.Tests.Utilities;
+using FluentAssertions;
+
+using Xunit;
+
+namespace DynamicData.Tests.Binding;
+
+///
+/// Single-threaded contract tests for :
+/// handler attachment ordering, no-dedup semantics, deep-chain re-walks on swaps.
+///
+public sealed class WhenPropertyChangedBehaviorFixture
+{
+ [Fact]
+ public void Shallow_NotifyInitialFalse_SubscribesHandlerBeforeReturning()
+ {
+ // notifyOnInitialValue=false: Subscribe must return only after the PropertyChanged handler
+ // is attached. A setter that fires immediately after Subscribe returns must reach the
+ // observer.
+ var model = new TestModel { Value = 10 };
+ var emissions = new List();
+
+ using var sub = model.WhenPropertyChanged(m => m.Value, notifyOnInitialValue: false)
+ .Subscribe(pv => emissions.Add(pv.Value));
+
+ model.Value = 20;
+
+ emissions.Should().Equal(new[] { 20 });
+ }
+
+ [Fact]
+ public void Shallow_NotifyInitialTrue_DoesNotDedupSameValuedEvents()
+ {
+ var model = new TestModel { Value = 10 };
+ var emissions = new List();
+
+ using var sub = model.WhenPropertyChanged(m => m.Value, notifyOnInitialValue: true)
+ .Subscribe(pv => emissions.Add(pv.Value));
+
+ model.Value = 10;
+ model.Value = 10;
+ model.Value = 10;
+
+ emissions.Should().Equal(new[] { 10, 10, 10, 10 });
+ }
+
+ [Fact]
+ public void Shallow_NotifyInitialFalse_DoesNotDedupSameValuedEvents()
+ {
+ var model = new TestModel { Value = 10 };
+ var emissions = new List();
+
+ using var sub = model.WhenPropertyChanged(m => m.Value, notifyOnInitialValue: false)
+ .Subscribe(pv => emissions.Add(pv.Value));
+
+ model.Value = 42;
+ model.Value = 42;
+
+ emissions.Should().Equal(new[] { 42, 42 });
+ }
+
+ [Fact]
+ public void DeepChain_NotifyInitialTrue_DoesNotDedupSameValuedEvents()
+ {
+ var parent = new ParentModel { Child = new ChildModel { Age = 1 } };
+ var emissions = new List();
+
+ using var sub = parent.WhenPropertyChanged(p => p.Child!.Age, notifyOnInitialValue: true)
+ .Subscribe(pv => emissions.Add(pv.Value));
+
+ parent.Child!.Age = 1;
+ parent.Child!.Age = 1;
+ parent.Child!.Age = 1;
+
+ emissions.Should().Equal(new[] { 1, 1, 1, 1 });
+ }
+
+ [Fact]
+ public void DeepChain_NotifyInitialFalse_DoesNotDedupSameValuedEvents()
+ {
+ var parent = new ParentModel { Child = new ChildModel { Age = 1 } };
+ var emissions = new List();
+
+ using var sub = parent.WhenPropertyChanged(p => p.Child!.Age, notifyOnInitialValue: false)
+ .Subscribe(pv => emissions.Add(pv.Value));
+
+ parent.Child!.Age = 7;
+ parent.Child!.Age = 7;
+
+ emissions.Should().Equal(new[] { 7, 7 });
+ }
+
+ [Fact]
+ public void DeepChain_PostSwap_LeafEventOnNewChild_Captured()
+ {
+ // After parent.Child is reassigned, the leaf-level subscription must be re-attached
+ // against the new child. A subsequent leaf mutation on the new child must be captured.
+ var parent = new ParentModel { Child = new ChildModel { Age = 10 } };
+ var emissions = new List();
+
+ using var sub = parent.WhenPropertyChanged(p => p.Child!.Age, notifyOnInitialValue: true)
+ .Subscribe(pv => emissions.Add(pv.Value));
+
+ var newChild = new ChildModel { Age = 20 };
+ parent.Child = newChild;
+ newChild.Age = 30;
+
+ emissions.Should().Equal(new[] { 10, 20, 30 });
+ }
+
+ [Fact]
+ public void DeepChain_MidChainSwap_DeeperLevelsRetargetCorrectly()
+ {
+ // Mid-chain swap on a 4-level chain. When level 3 is reassigned, the leaf subscription
+ // must re-attach against the new subtree; events on the old subtree must be ignored
+ // (its notifier subscription was disposed).
+ var l1 = new Level1
+ {
+ Child = new Level2
+ {
+ Child = new Level3
+ {
+ Child = new Level4 { Leaf = 10 },
+ },
+ },
+ };
+
+ var emissions = new List();
+ using var sub = l1.WhenPropertyChanged(x => x.Child!.Child!.Child!.Leaf, notifyOnInitialValue: true)
+ .Subscribe(pv => emissions.Add(pv.Value));
+
+ emissions.Should().Equal(new[] { 10 }, "initial emission");
+
+ var originalLeaf = l1.Child!.Child!.Child!;
+
+ var newL4 = new Level4 { Leaf = 20 };
+ l1.Child!.Child!.Child = newL4;
+
+ emissions.Should().Equal(new[] { 10, 20 }, "mid-chain swap emits the new leaf value");
+
+ newL4.Leaf = 30;
+ emissions.Should().Equal(new[] { 10, 20, 30 }, "leaf event on new subtree is captured");
+
+ originalLeaf.Leaf = 999;
+ emissions.Should().Equal(new[] { 10, 20, 30 }, "leaf event on detached subtree is ignored");
+ }
+
+ // https://github.com/reactivemarbles/DynamicData/issues/1149
+ [Fact]
+ public void ExpressionContainsImplicitInterfaceCast()
+ {
+ var child = new ChildModel()
+ {
+ Age = 10
+ };
+
+ using var subscription = ObserveAge(child)
+ .RecordValues(out var results);
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.RecordedValues.Should().ContainSingle("the initial value of the observed expression should have been published");
+ results.RecordedValues[0].Should().Be(child.Age, "the initial value of the observed expression should have been published");
+
+ ++child.Age;
+
+ results.Error.Should().BeNull("no errors should have occurred");
+ results.RecordedValues.Skip(1).Should().ContainSingle("the value of the observed expression changed once");
+ results.RecordedValues[1].Should().Be(child.Age, "the correct value should have been published");
+
+ static IObservable ObserveAge(T source)
+ where T : IHasAge
+ => source.WhenValueChanged(source => source.Age);
+ }
+
+ private interface IHasAge
+ : INotifyPropertyChanged
+ {
+ int Age { get; }
+ }
+
+ private sealed class TestModel : INotifyPropertyChanged
+ {
+ private int _value;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public int Value
+ {
+ get => _value;
+ set
+ {
+ _value = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Value)));
+ }
+ }
+ }
+
+ private sealed class ParentModel : INotifyPropertyChanged
+ {
+ private ChildModel? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public ChildModel? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class ChildModel
+ : IHasAge
+ {
+ private int _age;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public int Age
+ {
+ get => _age;
+ set
+ {
+ _age = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Age)));
+ }
+ }
+ }
+
+ private sealed class Level1 : INotifyPropertyChanged
+ {
+ private Level2? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public Level2? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class Level2 : INotifyPropertyChanged
+ {
+ private Level3? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public Level3? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class Level3 : INotifyPropertyChanged
+ {
+ private Level4? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public Level4? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class Level4 : INotifyPropertyChanged
+ {
+ private int _leaf;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public int Leaf
+ {
+ get => _leaf;
+ set
+ {
+ _leaf = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Leaf)));
+ }
+ }
+ }
+}
diff --git a/src/DynamicData.Tests/Binding/WhenPropertyChangedRaceFixture.cs b/src/DynamicData.Tests/Binding/WhenPropertyChangedRaceFixture.cs
new file mode 100644
index 000000000..9da900efe
--- /dev/null
+++ b/src/DynamicData.Tests/Binding/WhenPropertyChangedRaceFixture.cs
@@ -0,0 +1,474 @@
+// Copyright (c) 2011-2025 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.Collections.Generic;
+using System.ComponentModel;
+using System.Linq;
+using System.Reactive;
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using DynamicData.Binding;
+using DynamicData.Tests.Utilities;
+
+using FluentAssertions;
+
+using Xunit;
+
+namespace DynamicData.Tests.Binding;
+
+///
+/// Multi-threaded race tests for .
+/// Each test forces concurrency between the operator's subscribe call (or chain re-walk) and one or more
+/// notifiers firing on other threads.
+///
+public sealed class WhenPropertyChangedRaceFixture
+{
+ private static readonly TimeSpan ConditionTimeout = TimeSpan.FromSeconds(30);
+
+ [Fact]
+ public async Task DeepChain_ConcurrentParentSwap_LeafEventOnWinnerNotDropped()
+ {
+ // Two threads concurrently swap parent.Child. After both swaps complete, a leaf mutation
+ // on the current child must be captured. SharedDeliveryQueue serialises the level-0
+ // signals on the drainer, so the final level-1 subscription always targets parent.Child's
+ // current value.
+ const int iterations = 50;
+ var losses = 0;
+
+ for (var iter = 0; iter < iterations; iter++)
+ {
+ var parent = new ParentModel { Child = new ChildModel { Age = 0 } };
+ var emissions = new List();
+
+ using var sub = parent.WhenPropertyChanged(p => p.Child!.Age, notifyOnInitialValue: false)
+ .Subscribe(pv => { lock (emissions) emissions.Add(pv.Value); });
+
+ var newChild1 = new ChildModel { Age = 1 };
+ var newChild2 = new ChildModel { Age = 2 };
+
+ using var barrier = new Barrier(2);
+ var taskA = Task.Run(() => { barrier.SignalAndWait(); parent.Child = newChild1; });
+ var taskB = Task.Run(() => { barrier.SignalAndWait(); parent.Child = newChild2; });
+ await Task.WhenAll(taskA, taskB).WaitAsync(ConditionTimeout);
+
+ var winner = parent.Child;
+ if (winner is null)
+ {
+ continue;
+ }
+
+ winner.Age = 99;
+
+ WaitForCondition(() => { lock (emissions) return emissions.Contains(99); });
+
+ lock (emissions)
+ {
+ if (!emissions.Contains(99))
+ {
+ losses++;
+ }
+ }
+ }
+
+ losses.Should().Be(0, $"out of {iterations} iterations, {losses} dropped the leaf event on the post-swap winner");
+ }
+
+ [Fact]
+ public async Task DeepChain_FiveLevels_AllLevelsMutatedConcurrently_FinalEmissionMatchesActual()
+ {
+ // Torture: five worker threads each mutating at a different level of a 5-level chain.
+ // Mutations that land on detached subtrees are ignored (their notifier subscriptions were
+ // disposed by ResubscribeFrom). Mutations on the live chain reach the drainer.
+ //
+ // Three invariants per iteration:
+ // (a) Rx contract: ValidateSynchronization catches any concurrent OnNext on the user
+ // observer (a SharedDeliveryQueue serialisation failure).
+ // (b) Value legality: every emission must be a value that some thread legitimately
+ // wrote.
+ // (c) Final consistency: after Task.WhenAll the drainer continues until the queue is
+ // empty. The last processed signal triggers a ReadCurrent against the now-frozen
+ // chain state, so emissions.Last() == ReadCurrent().
+ const int iterations = 50;
+ const int mutationsPerThread = 200;
+ var mismatches = 0;
+
+ for (var iter = 0; iter < iterations; iter++)
+ {
+ var root = NewDeepChain(0);
+ var emissions = new List();
+
+ using var sub = root.WhenPropertyChanged(r => r.Child!.Child!.Child!.Child!.Leaf, notifyOnInitialValue: true)
+ .ValidateSynchronization()
+ .Subscribe(pv => { lock (emissions) emissions.Add(pv.Value); });
+
+ using var barrier = new Barrier(5);
+ var iterSeed = iter * 10_000;
+ var tasks = new[]
+ {
+ Task.Run(() =>
+ {
+ barrier.SignalAndWait();
+ for (var i = 0; i < mutationsPerThread; i++)
+ {
+ root.Child = NewDeep2(iterSeed + 40_000 + i);
+ }
+ }),
+ Task.Run(() =>
+ {
+ barrier.SignalAndWait();
+ for (var i = 0; i < mutationsPerThread; i++)
+ {
+ var l2 = root.Child;
+ if (l2 is not null) l2.Child = NewDeep3(iterSeed + 30_000 + i);
+ }
+ }),
+ Task.Run(() =>
+ {
+ barrier.SignalAndWait();
+ for (var i = 0; i < mutationsPerThread; i++)
+ {
+ var l3 = root.Child?.Child;
+ if (l3 is not null) l3.Child = NewDeep4(iterSeed + 20_000 + i);
+ }
+ }),
+ Task.Run(() =>
+ {
+ barrier.SignalAndWait();
+ for (var i = 0; i < mutationsPerThread; i++)
+ {
+ var l4 = root.Child?.Child?.Child;
+ if (l4 is not null) l4.Child = new Deep5 { Leaf = iterSeed + 10_000 + i };
+ }
+ }),
+ Task.Run(() =>
+ {
+ barrier.SignalAndWait();
+ for (var i = 0; i < mutationsPerThread; i++)
+ {
+ var l5 = root.Child?.Child?.Child?.Child;
+ if (l5 is not null) l5.Leaf = i;
+ }
+ }),
+ };
+
+ await Task.WhenAll(tasks).WaitAsync(ConditionTimeout);
+
+ var actualFinal = root.Child!.Child!.Child!.Child!.Leaf;
+
+ WaitForCondition(() => { lock (emissions) return emissions.Count > 0 && emissions[^1] == actualFinal; });
+
+ var legal = new HashSet { 0 };
+ for (var i = 0; i < mutationsPerThread; i++)
+ {
+ legal.Add(i);
+ legal.Add(iterSeed + 10_000 + i);
+ legal.Add(iterSeed + 20_000 + i);
+ legal.Add(iterSeed + 30_000 + i);
+ legal.Add(iterSeed + 40_000 + i);
+ }
+
+ lock (emissions)
+ {
+ emissions.Should().NotBeEmpty($"iter {iter}: notifyOnInitialValue=true requires at least the initial emission");
+ emissions[0].Should().Be(0, $"iter {iter}: first emission must be the initial value");
+
+ var illegal = emissions.Where(v => !legal.Contains(v)).ToList();
+ illegal.Should().BeEmpty($"iter {iter}: every emission must be a value some thread wrote; saw {string.Join(",", illegal.Take(5))}");
+
+ if (emissions.Count == 0 || emissions[^1] != actualFinal)
+ {
+ mismatches++;
+ }
+ }
+ }
+
+ mismatches.Should().Be(0, $"out of {iterations} iterations, {mismatches} ended with the last emission not matching the actual final chain leaf");
+ }
+
+ [Fact(Skip = "AutoRefresh has a separate concurrency bug; tracked separately")]
+ public async Task AutoRefreshThenFilter_ConcurrentAddsAndPropertyActivation_AllItemsObserved()
+ {
+ // One adder thread sequentially adds items to the cache while a single flipper thread
+ // concurrently sets each item's Activated to true. Final filter contents must include
+ // every item (every item ends Activated=true).
+ //
+ // KeyedActivable's setter only raises PropertyChanged on actual value change, so a
+ // dropped false->true transition is unrecoverable.
+ //
+ // The race lives in AutoRefresh's internal Publish multicast: Sub 1 (Filter path)
+ // receives the Add and reads the property before Sub 2 (MergeMany) subscribes the
+ // per-item refresh handler. A concurrent flip landing in that gap is dropped. This
+ // is not a WhenPropertyChanged issue: AutoRefresh calls WhenPropertyChanged with
+ // notifyInitial=false, so the per-item subscribe attaches the handler immediately
+ // and has no internal race window.
+ const int iterations = 100;
+ const int itemCount = 200;
+
+ for (var iter = 0; iter < iterations; iter++)
+ {
+ using var cache = new SourceCache(x => x.Id);
+ var items = Enumerable.Range(0, itemCount).Select(i => new KeyedActivable(i)).ToList();
+
+ using var results = cache.Connect()
+ .AutoRefresh(x => x.Activated)
+ .Filter(x => x.Activated)
+ .AsAggregator();
+
+ using var barrier = new Barrier(2);
+
+ var adder = Task.Run(() =>
+ {
+ barrier.SignalAndWait();
+ foreach (var item in items) cache.AddOrUpdate(item);
+ });
+
+ var flipper = Task.Run(() =>
+ {
+ barrier.SignalAndWait();
+ foreach (var item in items) item.Activated = true;
+ });
+
+ await Task.WhenAll(adder, flipper).WaitAsync(ConditionTimeout);
+
+ var expected = items.Select(x => x.Id).ToHashSet();
+ WaitForCondition(() => results.Data.Keys.ToHashSet().SetEquals(expected));
+
+ var actual = results.Data.Keys.ToHashSet();
+ actual.Should().BeEquivalentTo(expected, $"iter {iter}: every item ends Activated=true and must appear in the filter (missing: {string.Join(",", expected.Except(actual))})");
+ results.Error.Should().BeNull($"iter {iter}: pipeline must not error");
+ }
+ }
+
+ [Fact(Skip = "AutoRefresh has a separate concurrency bug; tracked separately")]
+ public async Task AutoRefreshThenFilter_DualSubscribers_AllItemsObserved()
+ {
+ // Two independent cache subscribers running on the ThreadPool:
+ // Sub 1 (mutator): on every Add change, flips item.Activated to true
+ // Sub 2 (filter chain): AutoRefresh + Filter (filter = Activated)
+ // Items start with Activated=false (filtered out). The mutator flips every item, so
+ // the final filter contents must include every item.
+ //
+ // Same root cause as the single-flipper variant above: AutoRefresh's internal Publish
+ // multicasts the Add to the Filter path before MergeMany subscribes the per-item
+ // refresh handler. The mutator's flip can land in that gap and be dropped.
+ const int iterations = 100;
+ const int itemCount = 200;
+
+ for (var iter = 0; iter < iterations; iter++)
+ {
+ using var cache = new SourceCache(x => x.Id);
+ var items = Enumerable.Range(0, itemCount).Select(i => new KeyedActivable(i)).ToList();
+
+ using var mutator = cache.Connect()
+ .ObserveOn(TaskPoolScheduler.Default)
+ .Subscribe(changes =>
+ {
+ foreach (var change in changes)
+ {
+ if (change.Reason == ChangeReason.Add)
+ {
+ change.Current.Activated = true;
+ }
+ }
+ });
+
+ using var results = cache.Connect()
+ .ObserveOn(TaskPoolScheduler.Default)
+ .AutoRefresh(x => x.Activated)
+ .Filter(x => x.Activated)
+ .AsAggregator();
+
+ foreach (var item in items) cache.AddOrUpdate(item);
+
+ var expected = items.Select(x => x.Id).ToHashSet();
+ WaitForCondition(() => results.Data.Keys.ToHashSet().SetEquals(expected));
+
+ var actual = results.Data.Keys.ToHashSet();
+ actual.Should().BeEquivalentTo(expected, $"iter {iter}: every item was flipped to Activated=true by the mutator and must appear in the filter (missing: {string.Join(",", expected.Except(actual))})");
+ results.Error.Should().BeNull($"iter {iter}: pipeline must not error");
+ }
+ }
+
+ private static Deep1 NewDeepChain(int leaf) =>
+ new Deep1 { Child = NewDeep2(leaf) };
+
+ private static Deep2 NewDeep2(int leaf) =>
+ new Deep2 { Child = NewDeep3(leaf) };
+
+ private static Deep3 NewDeep3(int leaf) =>
+ new Deep3 { Child = NewDeep4(leaf) };
+
+ private static Deep4 NewDeep4(int leaf) =>
+ new Deep4 { Child = new Deep5 { Leaf = leaf } };
+
+ private static void WaitForCondition(Func condition, TimeSpan? timeout = null) =>
+ SpinWait.SpinUntil(condition, timeout ?? ConditionTimeout);
+
+ private sealed class Item : INotifyPropertyChanged
+ {
+ private int _value;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public int Id { get; init; }
+
+ public int Value
+ {
+ get => _value;
+ set
+ {
+ _value = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Value)));
+ }
+ }
+ }
+
+ private sealed class ParentModel : INotifyPropertyChanged
+ {
+ private ChildModel? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public ChildModel? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class ChildModel : INotifyPropertyChanged
+ {
+ private int _age;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public int Age
+ {
+ get => _age;
+ set
+ {
+ _age = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Age)));
+ }
+ }
+ }
+
+ private sealed class Deep1 : INotifyPropertyChanged
+ {
+ private Deep2? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public Deep2? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class Deep2 : INotifyPropertyChanged
+ {
+ private Deep3? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public Deep3? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class Deep3 : INotifyPropertyChanged
+ {
+ private Deep4? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public Deep4? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class Deep4 : INotifyPropertyChanged
+ {
+ private Deep5? _child;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public Deep5? Child
+ {
+ get => _child;
+ set
+ {
+ _child = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+ }
+ }
+ }
+
+ private sealed class Deep5 : INotifyPropertyChanged
+ {
+ private int _leaf;
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public int Leaf
+ {
+ get => _leaf;
+ set
+ {
+ _leaf = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Leaf)));
+ }
+ }
+ }
+
+ private sealed class KeyedActivable : INotifyPropertyChanged
+ {
+ private bool _activated;
+
+ public KeyedActivable(int id)
+ {
+ Id = id;
+ }
+
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ public int Id { get; }
+
+ public bool Activated
+ {
+ get => _activated;
+ set
+ {
+ if (_activated == value) return;
+ _activated = value;
+ PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Activated)));
+ }
+ }
+ }
+}
diff --git a/src/DynamicData.Tests/Cache/AutoRefreshFixture.Base.cs b/src/DynamicData.Tests/Cache/AutoRefreshFixture.Base.cs
new file mode 100644
index 000000000..ebdc26950
--- /dev/null
+++ b/src/DynamicData.Tests/Cache/AutoRefreshFixture.Base.cs
@@ -0,0 +1,617 @@
+using System;
+using System.Linq;
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+
+using Microsoft.Reactive.Testing;
+
+using FluentAssertions;
+using Xunit;
+
+using DynamicData.Tests.Utilities;
+
+namespace DynamicData.Tests.Cache;
+
+public static partial class AutoRefreshFixture
+{
+ public abstract class Base
+ {
+ [Fact]
+ public void ChangeSetBufferIsGiven_PropertyChangedNotificationsAreBufferedOnScheduler()
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ source.AddOrUpdate(new[] { item1, item2, item3 });
+
+ var scheduler = new TestScheduler();
+
+
+ // UUT Initialization
+ using var subscription = BuildUut(
+ source: source.Connect(),
+ changeSetBuffer: TimeSpan.FromSeconds(10),
+ scheduler: scheduler)
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "the initial changeset should propagate");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "3 items were added to the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (publish property change notification)
+ ++item2.Value;
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Should().BeEmpty("the property change notification should have been buffered");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (advance time, within buffer window)
+ scheduler.AdvanceTo(TimeSpan.FromSeconds(5).Ticks);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Should().BeEmpty("the buffer window has not yet ended");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (advance time, to buffer window)
+ scheduler.AdvanceTo(TimeSpan.FromSeconds(10).Ticks);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Count().Should().Be(1, "a buffer window expired");
+ results.RecordedChangeSets.Skip(1).First().Count.Should().Be(1, "1 item published a property change notification");
+ results.RecordedChangeSets.Skip(1).First().Refreshes.Should().Be(1, "1 item published a property change notification");
+ results.RecordedChangeSets.Skip(1).First().First().Current.Should().Be(item2, "item #2 published a property change notification");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "no items should have changed, within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (publish property change notification)
+ ++item1.Value;
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(2).Should().BeEmpty("the property change notification should have been buffered");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (advance time, within buffer window)
+ scheduler.AdvanceTo(TimeSpan.FromSeconds(15).Ticks);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(2).Should().BeEmpty("the buffer window has not yet ended");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (publish additional property change notification)
+ ++item3.Value;
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(2).Should().BeEmpty("the property change notification should have been buffered");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (advance time, to buffer window)
+ scheduler.AdvanceTo(TimeSpan.FromSeconds(20).Ticks);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(2).Count().Should().Be(1, "a buffer window expired");
+ results.RecordedChangeSets.Skip(2).First().Count.Should().Be(2, "2 items published a property change notification");
+ results.RecordedChangeSets.Skip(2).First().Refreshes.Should().Be(2, "2 items published a property change notification");
+ results.RecordedChangeSets.Skip(2).First().Select(change => change.Current).Should().BeEquivalentTo(new[] { item1, item3 }, "items #2 and #3 published property change notification");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "no items should have changed, within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (normal refresh)
+ source.Refresh(item2);
+
+ // Normal refreshes should not be buffered
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(3).Count().Should().Be(1, "one source operation was performed");
+ results.RecordedChangeSets.Skip(3).First().Count.Should().Be(1, "1 item was refreshed, within the source");
+ results.RecordedChangeSets.Skip(3).First().Refreshes.Should().Be(1, "1 item was refreshed, within the source");
+ results.RecordedChangeSets.Skip(3).First().First().Current.Should().Be(item2, "item #2 was refreshed, within the source");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "no items should have changed, within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+ }
+
+ [Fact]
+ public void ItemIsAdded_SubscribesToPropertyChanged()
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+ // UUT Initialization
+ using var subscription = BuildUut(source.Connect())
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Should().BeEmpty("no source operations were performed");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ source.AddOrUpdate(new[] { item1, item2, item3 });
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "one source operation was performed");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "3 items were added to the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+ item1.HasSubscriptions.Should().BeTrue("the PropertyChanged event should be subscribed to, for each added item");
+ item2.HasSubscriptions.Should().BeTrue("the PropertyChanged event should be subscribed to, for each added item");
+ item3.HasSubscriptions.Should().BeTrue("the PropertyChanged event should be subscribed to, for each added item");
+ }
+
+ [Fact]
+ public void ItemIsMoved_NotificationPropagates()
+ {
+ // Setup
+ using var source = new Subject>();
+
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ var items = new [] { item1, item2, item3 };
+
+ var initialChangeset = new ChangeSet
- ()
+ {
+ new Change
- (reason: ChangeReason.Add, key: item1.Id, current: item1, index: 0),
+ new Change
- (reason: ChangeReason.Add, key: item2.Id, current: item2, index: 1),
+ new Change
- (reason: ChangeReason.Add, key: item3.Id, current: item3, index: 2)
+ };
+
+ // UUT Initialization
+ using var subscription = BuildUut(source.Prepend(initialChangeset))
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "the initial changeset should propagate");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(items, "3 items were added to the source");
+ results.RecordedItemsSorted.Should().BeEquivalentTo(
+ items,
+ options => options.WithStrictOrdering(),
+ "item indexes should propagate");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action
+ source.OnNext(new ChangeSet
- ()
+ {
+ new Change
- (
+ key: item3.Id,
+ current: item3,
+ currentIndex: 0,
+ previousIndex: 2)
+ });
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Count().Should().Be(1, "one source operation was performed");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(items, "an item was moved within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+ results.RecordedItemsSorted.Should().BeEquivalentTo(
+ new[] { item3, item1, item2 },
+ options => options.WithStrictOrdering(),
+ "an item was moved within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+ }
+
+ [Fact]
+ public void ItemIsRefreshed_NotificationPropagates()
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ source.AddOrUpdate(new[] { item1, item2, item3 });
+
+
+ // UUT Initialization
+ using var subscription = BuildUut(source.Connect())
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "the initial changeset should propagate");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "3 items were added to the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+ // UUT Action
+ source.Refresh(item2);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Count().Should().Be(1, "one source operation was performed");
+ results.RecordedChangeSets.Skip(1).First().Count.Should().Be(1, "1 item was refreshed within the source");
+ results.RecordedChangeSets.Skip(1).First().Refreshes.Should().Be(1, "1 item was refreshed within the source");
+ results.RecordedChangeSets.Skip(1).First().First().Current.Should().Be(item2, "item #2 was refreshed within the source");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "no items were changed, within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+ }
+
+ [Fact]
+ public void ItemIsRemoved_UnsubscribesFromPropertyChanged()
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ source.AddOrUpdate(new[] { item1, item2, item3 });
+
+
+ // UUT Initialization
+ using var subscription = BuildUut(source.Connect())
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "the initial changeset should propagate");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "3 items were added to the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action
+ source.Remove(item2);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Count().Should().Be(1, "one source operation was performed");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "1 item was removed from the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+ item2.HasSubscriptions.Should().BeFalse("removing an item should trigger unsubscription from its reevaluator");
+ item1.HasSubscriptions.Should().BeTrue("the item was not removed from the source");
+ item3.HasSubscriptions.Should().BeTrue("the item was not removed from the source");
+ }
+
+ [Fact]
+ public void ItemIsUpdated_ReSubscribesToPropertyChanged()
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ source.AddOrUpdate(new[] { item1, item2, item3 });
+
+
+ // UUT Initialization
+ using var subscription = BuildUut(source.Connect())
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "the initial changeset should propagate");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "3 items were added to the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action
+ var item4 = new Item() { Id = 2 };
+ source.AddOrUpdate(item4);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Count().Should().Be(1, "one source operation was performed");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "1 item was replaced within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+ item2.HasSubscriptions.Should().BeFalse("replacing an item should trigger unsubscription from its reevaluator");
+ item4.HasSubscriptions.Should().BeTrue("adding an item should invoke its reevaluator and subscribe to it");
+ item1.HasSubscriptions.Should().BeTrue("the item was not removed from the source");
+ item3.HasSubscriptions.Should().BeTrue("the item was not removed from the source");
+ }
+
+ [Fact]
+ public void PropertyChangedOccurs_ItemRefreshes()
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ source.AddOrUpdate(new[] { item1, item2, item3 });
+
+
+ // UUT Initialization
+ using var subscription = BuildUut(source.Connect())
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "the initial changeset should propagate");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "3 items were added to the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action
+ ++item2.Value;
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Count().Should().Be(1, "1 item published a property change notification");
+ results.RecordedChangeSets.Skip(1).First().Count.Should().Be(1, "1 item published a property change notification");
+ results.RecordedChangeSets.Skip(1).First().Refreshes.Should().Be(1, "1 item published a property change notification");
+ results.RecordedChangeSets.Skip(1).First().First().Current.Should().Be(item2, "item #2 published a property change notification");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "no source operations were performed");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+ }
+
+ [Fact]
+ public void PropertyChangeThrottleIsGiven_PropertyChangedNotificationsAreThrottledByScheduler()
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ source.AddOrUpdate(new[] { item1, item2, item3 });
+
+ var scheduler = new TestScheduler();
+
+
+ // UUT Initialization
+ using var subscription = BuildUut(
+ source: source.Connect(),
+ propertyChangeThrottle: TimeSpan.FromSeconds(10),
+ scheduler: scheduler)
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "the initial changeset should propagate");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "3 items were added to the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (publish property change notification)
+ ++item2.Value;
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Should().BeEmpty("the throttle window has not yet ended");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (publish additional property change notification, immediately)
+ ++item2.Value;
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Should().BeEmpty("the throttle window has not yet ended");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (advance time to end of throttle window)
+ scheduler.AdvanceTo(TimeSpan.FromSeconds(10).Ticks);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(1).Count().Should().Be(1, "the throttle window ended");
+ results.RecordedChangeSets.Skip(1).First().Count.Should().Be(1, "1 item published property change notifications");
+ results.RecordedChangeSets.Skip(1).First().Refreshes.Should().Be(1, "1 item published property change notifications");
+ results.RecordedChangeSets.Skip(1).First().First().Current.Should().Be(item2, "item #2 published property change notifications");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "no items should have changed, within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (publish property change notification)
+ ++item2.Value;
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(2).Should().BeEmpty("the throttle window has not yet ended");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (publish additional property change notification, within throttle window)
+ scheduler.AdvanceTo(TimeSpan.FromSeconds(15).Ticks);
+ ++item2.Value;
+ scheduler.AdvanceBy(1);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(2).Should().BeEmpty("the throttle window has not yet ended");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (advance time to end of original throttle window)
+ scheduler.AdvanceTo(TimeSpan.FromSeconds(20).Ticks);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(2).Should().BeEmpty("the throttle window should have been extended");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+
+
+ // UUT Action (advance time to end of throttle window)
+ scheduler.AdvanceTo(TimeSpan.FromSeconds(25).Ticks);
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Skip(2).Count().Should().Be(1, "the throttle window ended");
+ results.RecordedChangeSets.Skip(2).First().Count.Should().Be(1, "1 item published property change notifications");
+ results.RecordedChangeSets.Skip(2).First().Refreshes.Should().Be(1, "1 item published property change notifications");
+ results.RecordedChangeSets.Skip(2).First().First().Current.Should().Be(item2, "item #2 published property change notifications");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "no items should have changed, within the source");
+ results.HasCompleted.Should().BeFalse("the source has not completed");
+ }
+
+ [Theory]
+ [InlineData(NotificationStrategy.Immediate)]
+ [InlineData(NotificationStrategy.Asynchronous)]
+ public void SourceCompletesWhenEmpty_CompletionPropagates(NotificationStrategy notificationStrategy)
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+
+ // UUT Initialization & Action
+ if (notificationStrategy is NotificationStrategy.Immediate)
+ source.Complete();
+
+ using var subscription = BuildUut(source.Connect())
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ if (notificationStrategy is NotificationStrategy.Asynchronous)
+ source.Complete();
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Should().BeEmpty("no source operations were performed");
+ results.HasCompleted.Should().BeTrue("all notification sources have completed");
+ }
+
+ [Theory]
+ [InlineData(NotificationStrategy.Immediate)]
+ [InlineData(NotificationStrategy.Asynchronous)]
+ public void SourceCompletesWhenNotEmpty_CompletionDoesNotPropagate(NotificationStrategy notificationStrategy)
+ {
+ // Setup
+ using var source = new TestSourceCache
- (Item.SelectId);
+
+ var item1 = new Item() { Id = 1 };
+ var item2 = new Item() { Id = 2 };
+ var item3 = new Item() { Id = 3 };
+
+ source.AddOrUpdate(new[] { item1, item2, item3 });
+
+
+ // UUT Initialization & Action (source completion)
+ if (notificationStrategy is NotificationStrategy.Immediate)
+ source.Complete();
+
+ using var subscription = BuildUut(source.Connect())
+ .ValidateSynchronization()
+ .ValidateChangeSets(Item.SelectId)
+ .RecordCacheItems(out var results);
+
+ if (notificationStrategy is NotificationStrategy.Asynchronous)
+ source.Complete();
+
+ results.Error.Should().BeNull();
+ results.RecordedChangeSets.Count.Should().Be(1, "the initial changeset should propagate");
+ results.RecordedItemsByKey.Values.Should().BeEquivalentTo(source.Items, "3 items were added to the source");
+ results.HasCompleted.Should().BeFalse("PropertyChanged events can still publish notifications");
+ }
+
+ [Theory]
+ [InlineData(NotificationStrategy.Immediate)]
+ [InlineData(NotificationStrategy.Asynchronous)]
+ public void SourceFails_ErrorPropagates(NotificationStrategy notificationStrategy)
+ {
+ // Setup
+ using var source = new TestSourceCache