diff --git a/src/DynamicData.Tests/Cache/AndFixture.cs b/src/DynamicData.Tests/Cache/AndFixture.cs index 0f0ec2ad0..18a7b84ba 100644 --- a/src/DynamicData.Tests/Cache/AndFixture.cs +++ b/src/DynamicData.Tests/Cache/AndFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; @@ -7,6 +7,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -106,4 +108,34 @@ public void UpdatingOneSourceOnlyProducesNoResults() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = ObservableCacheEx.And(first, second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + + [Fact] + public void DeliversAnErrorFromAnySource() + { + Exception? error = null; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = ObservableCacheEx.And(first, second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/BatchIfFixture.cs b/src/DynamicData.Tests/Cache/BatchIfFixture.cs index d88bd2f03..5957feab2 100644 --- a/src/DynamicData.Tests/Cache/BatchIfFixture.cs +++ b/src/DynamicData.Tests/Cache/BatchIfFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Reactive.Linq; using System.Reactive.Subjects; @@ -9,6 +9,9 @@ using Microsoft.Reactive.Testing; using Xunit; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; namespace DynamicData.Tests.Cache; @@ -141,6 +144,53 @@ public void ResultsWillBeReceivedIfNotPaused() _results.Messages.Count.Should().Be(1, "Should be 1 update"); } + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.BatchIf(Observable.Return(false), Scheduler.Immediate).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void FlushesHeldChangesBeforeCompleting() + { + var received = 0; + var completed = false; + + using var source = new Subject>(); + using var pause = new BehaviorSubject(true); + using var subscription = source.BatchIf(pause, Scheduler.Immediate).Subscribe(_ => received++, () => completed = true); + + source.OnNext(new ChangeSet { new(ChangeReason.Add, "a", new Person("a", 1)) }); + source.OnCompleted(); + + received.Should().Be(1, "changes held back by the pause would otherwise be lost"); + completed.Should().BeTrue(); + } + + [Fact] + public void FailsWhenThePauseSelectorFails() + { + var expectedError = new Exception("Test Exception"); + var actualError = default(Exception); + var completed = false; + + using var source = new Subject>(); + using var pause = new Subject(); + using var subscription = source.BatchIf(pause, Scheduler.Immediate).Subscribe(_ => { }, error => actualError = error, () => completed = true); + + pause.OnError(expectedError); + + actualError.Should().BeSameAs(expectedError, "a failure of the pause selector belongs to the subscriber, not to whichever thread happened to raise it"); + completed.Should().BeFalse("the pause selector failed, it did not complete"); + } + [Fact] public void PauseSelectorOnlyStartsUnpaused() { diff --git a/src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs b/src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs index 1e19ba079..8e26af3bf 100644 --- a/src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs +++ b/src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Linq; using DynamicData.Tests.Domain; @@ -6,6 +6,10 @@ using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -61,4 +65,30 @@ public void SkipInitialDoesNotReturnTheFirstBatchOfData() updateReceived.Should().BeTrue(); deferStream.Dispose(); } + + [Fact] + public void DeferUntilLoadedCompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.DeferUntilLoaded().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void SkipInitialCompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.SkipInitial().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("SkipInitial is built on DeferUntilLoaded"); + } } diff --git a/src/DynamicData.Tests/Cache/ExceptFixture.cs b/src/DynamicData.Tests/Cache/ExceptFixture.cs index 17c4996e8..18fbb4341 100644 --- a/src/DynamicData.Tests/Cache/ExceptFixture.cs +++ b/src/DynamicData.Tests/Cache/ExceptFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using DynamicData.Tests.Domain; @@ -6,6 +6,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -80,4 +82,34 @@ public void UpdatingOneSourceOnlyProducesResult() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Except(second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + + [Fact] + public void DeliversAnErrorFromAnySource() + { + Exception? error = null; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Except(second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/GroupFixture.cs b/src/DynamicData.Tests/Cache/GroupFixture.cs index c0139eed5..0054a920c 100644 --- a/src/DynamicData.Tests/Cache/GroupFixture.cs +++ b/src/DynamicData.Tests/Cache/GroupFixture.cs @@ -9,6 +9,9 @@ using FluentAssertions; using Xunit; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -264,4 +267,71 @@ public class GroupViewModel public ReadOnlyObservableCollection Entries => _entries; } + + [Fact] + public void CompletesWhenNoRegrouperIsSupplied() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent regrouper can never fire and so must not hold the result open"); + } + + [Fact] + public void DeliversTheErrorWithoutThrowing() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + + [Fact] + public void DeliversTheErrorWhenAResultGroupSourceIsSupplied() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source + .Group(p => p.Age, Observable.Never>()) + .Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + + [Fact] + public void DeliversTerminalEventsFromTheResultGroupSource() + { + var completed = false; + + using (var source = new Subject>()) + using (var groups = new Subject>()) + using (source.Group(p => p.Age, groups).Subscribe(_ => { }, () => completed = true)) + { + groups.OnCompleted(); + } + + completed.Should().BeTrue("no group can appear once the result group source is finished"); + + Exception? error = null; + + using (var source = new Subject>()) + using (var groups = new Subject>()) + using (source.Group(p => p.Age, groups).Subscribe(_ => { }, ex => error = ex, () => { })) + { + groups.OnError(new InvalidOperationException("boom")); + } + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/GroupImmutableFixture.cs b/src/DynamicData.Tests/Cache/GroupImmutableFixture.cs index 8640bccb7..05c5c0692 100644 --- a/src/DynamicData.Tests/Cache/GroupImmutableFixture.cs +++ b/src/DynamicData.Tests/Cache/GroupImmutableFixture.cs @@ -7,6 +7,10 @@ using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -177,4 +181,17 @@ public void UpdatesArePermissible() var group = _results.Data.Items[0]; group.Count.Should().Be(2); } + + [Fact] + public void CompletesWhenNoRegrouperIsSupplied() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.GroupWithImmutableState(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent regrouper can never fire and so must not hold the result open"); + } } diff --git a/src/DynamicData.Tests/Cache/InnerJoinFixture.cs b/src/DynamicData.Tests/Cache/InnerJoinFixture.cs index bd5ffc054..9d51f4048 100644 --- a/src/DynamicData.Tests/Cache/InnerJoinFixture.cs +++ b/src/DynamicData.Tests/Cache/InnerJoinFixture.cs @@ -1,10 +1,16 @@ -using System; +using System; using DynamicData.Tests.Utilities; using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -457,4 +463,19 @@ public override bool Equals(object? obj) public override string ToString() => $"{Key}: {Device} ({MetaData})"; } + + [Fact] + public void InnerJoinManyCompletesWhenBothSidesComplete() + { + var completed = false; + + using var left = new Subject>(); + using var subscription = left + .InnerJoinMany(Observable.Empty>(), p => p.Name, (_, person, _) => person) + .Subscribe(_ => { }, () => completed = true); + + left.OnCompleted(); + + completed.Should().BeTrue("the grouping it is built on must not hold the result open"); + } } diff --git a/src/DynamicData.Tests/Cache/MergeManyChangeSetsParentUpdateFixture.cs b/src/DynamicData.Tests/Cache/MergeManyChangeSetsParentUpdateFixture.cs new file mode 100644 index 000000000..608f0be64 --- /dev/null +++ b/src/DynamicData.Tests/Cache/MergeManyChangeSetsParentUpdateFixture.cs @@ -0,0 +1,43 @@ +using System; +using System.Linq; + +using DynamicData.Tests.Domain; +using DynamicData.Tests.Utilities; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.Cache; + +public class MergeManyChangeSetsParentUpdateFixture +{ + [Fact] + public void ParentUpdateDoesNotEmitDuplicateAdds() + { + using var owners = new SourceCache(o => o.Id); + + var owner = new AnimalOwner("Owner"); + owner.Animals.AddRange([ + new Animal("A1", "Type", AnimalFamily.Mammal), + new Animal("A2", "Type", AnimalFamily.Mammal), + ]); + owners.AddOrUpdate(owner); + + using var subscription = owners.Connect() + .MergeManyChangeSets(o => o.Animals.Connect()) + .Transform(a => new Person(a.Name, a.Name.Length)) + .AddKey(static person => person.Name) + .ValidateChangeSets(static person => person.Name) + .RecordCacheItems(out var results); + + results.Error.Should().BeNull("the initial subscription should be valid"); + + // Re-adding the same instance is an Update on the parent, which swaps the child subscription. + owners.AddOrUpdate(owner); + + results.Error.Should().BeNull("a parent update must not re-add children that are already present"); + results.RecordedItemsByKey.Should().HaveCount(2, "the owner still has exactly two animals"); + } +} + diff --git a/src/DynamicData.Tests/Cache/MergeManyFixture.cs b/src/DynamicData.Tests/Cache/MergeManyFixture.cs index 553bf452e..3f6b962a0 100644 --- a/src/DynamicData.Tests/Cache/MergeManyFixture.cs +++ b/src/DynamicData.Tests/Cache/MergeManyFixture.cs @@ -1,10 +1,14 @@ -using System; +using System; using System.Reactive.Linq; using System.Reactive.Subjects; using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -83,4 +87,19 @@ public void InvokeObservable(bool value) _changed.OnNext(value); } } + + [Fact] + public void DeliversAnErrorRaisedByAChild() + { + Exception? error = null; + + using var source = new SourceCache(p => p.Name); + using var child = new Subject(); + using var subscription = source.Connect().MergeMany(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.AddOrUpdate(new Person("a", 1)); + child.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType("Merge propagates a failure from any inner stream"); + } } diff --git a/src/DynamicData.Tests/Cache/MergeManyItemsFixture.cs b/src/DynamicData.Tests/Cache/MergeManyItemsFixture.cs index e2a6be748..01ce186a8 100644 --- a/src/DynamicData.Tests/Cache/MergeManyItemsFixture.cs +++ b/src/DynamicData.Tests/Cache/MergeManyItemsFixture.cs @@ -1,10 +1,14 @@ -using System; +using System; using System.Reactive.Linq; using System.Reactive.Subjects; using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -95,4 +99,45 @@ public void InvokeObservable(bool value) _changed.OnNext(value); } } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.MergeManyItems(_ => Observable.Empty()).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void StaysOpenWhenOnlyAChildCompletes() + { + var completed = false; + + using var source = new SourceCache(p => p.Name); + using var subscription = source.Connect().MergeManyItems(_ => Observable.Return(1)).Subscribe(_ => { }, () => completed = true); + + source.AddOrUpdate(new Person("a", 1)); + + completed.Should().BeFalse("one child finishing does not finish the merge"); + } + + [Fact] + public void DeliversAnErrorRaisedByAChild() + { + Exception? error = null; + + using var source = new SourceCache(p => p.Name); + using var child = new Subject(); + using var subscription = source.Connect().MergeManyItems(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.AddOrUpdate(new Person("a", 1)); + child.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs b/src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs index e6987e846..abbc3750e 100644 --- a/src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs +++ b/src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs @@ -82,7 +82,7 @@ public void SingleItemCompleteWillNotMergedStream() } [Fact] - public void SingleItemFailWillNotFailMergedStream() + public void SingleItemFailWillFailMergedStream() { var failed = false; var stream = _source.Connect().MergeMany((o, key) => o.Observable).Subscribe(_ => { }, ex => failed = true); @@ -94,7 +94,7 @@ public void SingleItemFailWillNotFailMergedStream() stream.Dispose(); - failed.Should().BeFalse(); + failed.Should().BeTrue("Merge propagates a failure from any inner stream"); } /// @@ -144,7 +144,7 @@ public void MergedStreamCompletesWhenSourceAndItemsComplete() /// Stream completes even if one of the children fails. /// [Fact] - public void MergedStreamCompletesIfLastItemFails() + public void MergedStreamFailsIfLastItemFails() { var receivedError = default(Exception); var streamCompleted = false; @@ -159,9 +159,9 @@ public void MergedStreamCompletesIfLastItemFails() _source.Dispose(); item.FailObservable(new Exception("Test exception")); - receivedError.Should().Be(default); sourceCompleted.Should().BeTrue(); - streamCompleted.Should().BeTrue(); + receivedError.Should().NotBeNull("Merge propagates a failure from any inner stream"); + streamCompleted.Should().BeFalse("a failure and a completion are mutually exclusive"); } /// diff --git a/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs b/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs index 6a6a33a38..ca31d171f 100644 --- a/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs +++ b/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs @@ -7,6 +7,9 @@ using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -91,4 +94,47 @@ public void SetToLoaded() status.Should().Be(ConnectionStatus.Loaded, "Status should be ConnectionStatus.Loaded"); subscription.Dispose(); } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var statuses = new List(); + var completed = false; + + using var source = new Subject(); + using var subscription = source.MonitorStatus().Subscribe(statuses.Add, () => completed = true); + + source.OnNext(1); + source.OnCompleted(); + + completed.Should().BeTrue("the status stream is finished once the source is"); + statuses.Should().Equal(ConnectionStatus.Pending, ConnectionStatus.Loaded, ConnectionStatus.Completed); + } + + [Fact] + public void DeliversTheErrorAfterReportingIt() + { + var statuses = new List(); + Exception? error = null; + + using var source = new Subject(); + using var subscription = source.MonitorStatus().Subscribe(statuses.Add, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + statuses.Should().EndWith(ConnectionStatus.Errored); + } + + [Fact] + public void ReportsAStatusWhenTheSourceIsAlreadyFinished() + { + var statuses = new List(); + var completed = false; + + using var subscription = Observable.Empty().MonitorStatus().Subscribe(statuses.Add, () => completed = true); + + completed.Should().BeTrue("a terminal event arriving during subscription must not be lost"); + statuses.Should().Equal(ConnectionStatus.Pending, ConnectionStatus.Completed); + } } diff --git a/src/DynamicData.Tests/Cache/MultiSourceSerializationFixture.cs b/src/DynamicData.Tests/Cache/MultiSourceSerializationFixture.cs new file mode 100644 index 000000000..bebeed4bd --- /dev/null +++ b/src/DynamicData.Tests/Cache/MultiSourceSerializationFixture.cs @@ -0,0 +1,279 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reactive; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Reactive.Threading.Tasks; +using System.Threading; +using System.Threading.Tasks; + +using DynamicData.Binding; +using DynamicData.Tests.Domain; +using DynamicData.Tests.Utilities; + +using FluentAssertions; + +using Xunit; + +using Randomizer = Bogus.Randomizer; + +namespace DynamicData.Tests.Cache; + +/// +/// Serialization coverage for the cache operators that take more than one input. A single input is +/// serialized by whatever feeds it, so these are the operators where two producers can reach the +/// same observer at once, and where a missing gate shows up as overlapping delivery rather than as +/// a wrong value. Every pipeline is checked both for overlap and for change set integrity, since a +/// torn read of shared operator state usually surfaces as a structurally invalid change set rather +/// than as a visible race. +/// +public class MultiSourceSerializationFixture +{ + private const int Seed = 0x1146; + + private static readonly TimeSpan DrainTimeout = TimeSpan.FromMinutes(2); + + [Fact] + public async Task OrDeliversSeriallyWhileBothSourcesAreWritten() + { + using var left = new SourceCache(p => p.Name); + using var right = new SourceCache(p => p.Name); + + await RunAsync( + left.Connect().Or(right.Connect()), + writers: + [ + randomizer => left.AddOrUpdate(new Person("L" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => right.AddOrUpdate(new Person("R" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => left.Remove("L" + randomizer.Int(1, 20)), + ], + complete: () => + { + left.Dispose(); + right.Dispose(); + }); + } + + [Fact] + public async Task SortDeliversSeriallyWhileTheComparerChanges() + { + using var source = new SourceCache(p => p.Name); + using var comparers = new Subject>(); + using var resort = new Subject(); + + var byAge = SortExpressionComparer.Ascending(p => p.Age); + var byName = SortExpressionComparer.Ascending(p => p.Name); + + await RunAsync( + source.Connect().Sort(comparers, resort).Transform(static person => person), + writers: + [ + randomizer => source.AddOrUpdate(new Person("P" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => comparers.OnNext(randomizer.Bool() ? byAge : byName), + _ => resort.OnNext(Unit.Default), + ], + complete: () => + { + comparers.OnCompleted(); + resort.OnCompleted(); + source.Dispose(); + }, + seedAction: () => comparers.OnNext(byAge)); + } + + [Fact] + public async Task GroupOnDeliversSeriallyWhileRegroupingConcurrently() + { + using var source = new SourceCache(p => p.Name); + using var regroup = new Subject(); + + await RunAsync( + source.Connect().Group(p => p.Age % 5, regroup).Transform(group => new Person(group.Key.ToString(), group.Key)).ChangeKey(static person => person.Name), + writers: + [ + randomizer => source.AddOrUpdate(new Person("P" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => source.Remove("P" + randomizer.Int(1, 20)), + _ => regroup.OnNext(Unit.Default), + ], + complete: () => + { + regroup.OnCompleted(); + source.Dispose(); + }); + } + + [Fact] + public async Task BatchIfDeliversSeriallyWhilePausingConcurrently() + { + using var source = new SourceCache(p => p.Name); + using var pause = new Subject(); + + await RunAsync( + source.Connect().BatchIf(pause, initialPauseState: false, timeOut: null), + writers: + [ + randomizer => source.AddOrUpdate(new Person("P" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => source.Remove("P" + randomizer.Int(1, 20)), + randomizer => pause.OnNext(randomizer.Bool()), + ], + complete: () => + { + // Leave the batch open, so anything held back has to be flushed on completion. + pause.OnNext(false); + pause.OnCompleted(); + source.Dispose(); + }); + } + + [Fact] + public async Task TransformWithForcedTransformDeliversSeriallyWhileForcing() + { + using var source = new SourceCache(p => p.Name); + using var force = new Subject>(); + + await RunAsync( + source.Connect().Transform(p => new PersonWithGender(p, p.Age % 2 == 0 ? "M" : "F"), forceTransform: force).Transform(p => new Person(p.Name, p.Age)), + writers: + [ + randomizer => source.AddOrUpdate(new Person("P" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => source.Remove("P" + randomizer.Int(1, 20)), + _ => force.OnNext(static _ => true), + ], + complete: () => + { + force.OnCompleted(); + source.Dispose(); + }); + } + + [Fact] + public async Task DynamicOrDeliversSeriallyWhileBothSourcesAreWritten() + { + using var left = new SourceCache(p => p.Name); + using var right = new SourceCache(p => p.Name); + using var sources = new SourceList>>(); + + sources.Add(left.Connect()); + sources.Add(right.Connect()); + + await RunAsync( + sources.Or(), + writers: + [ + randomizer => left.AddOrUpdate(new Person("L" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => right.AddOrUpdate(new Person("R" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => left.Remove("L" + randomizer.Int(1, 20)), + ], + complete: () => + { + // Children first, then the list of sources: the combined result ends only once the + // sources themselves and the list feeding them have all finished. + left.Dispose(); + right.Dispose(); + sources.Dispose(); + }); + } + + [Fact] + public async Task MergeManyChangeSetsDeliversSeriallyWhileChildrenAreWritten() + { + using var owners = new SourceCache(o => o.Id); + var created = new List(); + + for (var i = 0; i < 5; i++) + { + var owner = new AnimalOwner("Owner" + i); + created.Add(owner); + owners.AddOrUpdate(owner); + } + + // The child collections are lists, which allow duplicates, so names are handed out from a + // counter. Otherwise keying the merged result collides for reasons that have nothing to do + // with how the writers interleave. + var nextName = 0; + + await RunAsync( + owners.Connect().MergeManyChangeSets(o => o.Animals.Connect()).Transform(a => new Person(a.Name, a.Name.Length)).AddKey(static person => person.Name), + writers: + [ + randomizer => created[randomizer.Int(0, created.Count - 1)].Animals.Add(new Animal("A" + Interlocked.Increment(ref nextName), "Type", AnimalFamily.Mammal)), + randomizer => + { + var owner = created[randomizer.Int(0, created.Count - 1)]; + var animals = owner.Animals.Items.ToArray(); + if (animals.Length > 0) + { + owner.Animals.Remove(animals[randomizer.Int(0, animals.Length - 1)]); + } + }, + randomizer => created[randomizer.Int(0, created.Count - 1)].Animals.Add(new Animal("B" + Interlocked.Increment(ref nextName), "Type", AnimalFamily.Bird)), + ], + complete: () => + { + // The merged result completes only once the parent and every child have, so the child + // lists have to be disposed as well as the owner cache. + foreach (var owner in created) + { + owner.Dispose(); + } + + owners.Dispose(); + }); + } + + /// + /// Drives every writer from its own thread against a shared barrier, then waits for the pipeline + /// to drain and asserts that it ended cleanly. A serialization failure surfaces as an + /// UnsynchronizedNotificationException on the terminal notification rather than as a + /// wrong count, so the assertion is on how the sequence ended. + /// + private static async Task RunAsync( + IObservable> pipeline, + Action[] writers, + Action complete, + Action? seedAction = null) + { + var randomizer = new Randomizer(Seed); + var iterations = randomizer.Int(150, 250); + + var published = pipeline + .ValidateSynchronization() + .ValidateChangeSets(static person => person.Name) + + // Holding each notification briefly makes an overlapping delivery observable instead of + // something that has to be caught inside a window of a few instructions. + .Do(static _ => Thread.SpinWait(500)) + .Publish(); + + var terminal = published.Materialize().LastAsync().ToTask(); + using var subscription = published.Connect(); + + seedAction?.Invoke(); + + using var barrier = new Barrier(writers.Length + 1); + + var tasks = writers.Select((writer, index) => Task.Run(() => + { + var threadRandomizer = new Randomizer(Seed + index + 1); + barrier.SignalAndWait(); + + for (var i = 0; i < iterations; i++) + { + writer(threadRandomizer); + } + })).ToArray(); + + barrier.SignalAndWait(); + await Task.WhenAll(tasks); + + complete(); + + (await Task.WhenAny(terminal, Task.Delay(DrainTimeout))).Should().BeSameAs(terminal, "the pipeline should drain rather than deadlock"); + + var lastNotification = await terminal; + + lastNotification.Exception.Should().BeNull("notifications must not overlap, and every change set must be structurally valid, however the writers interleave"); + } +} + diff --git a/src/DynamicData.Tests/Cache/OrFixture.cs b/src/DynamicData.Tests/Cache/OrFixture.cs index 720d3f0b4..da9d891f9 100644 --- a/src/DynamicData.Tests/Cache/OrFixture.cs +++ b/src/DynamicData.Tests/Cache/OrFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; @@ -7,6 +7,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -95,4 +97,34 @@ public void UpdatingOneSourceOnlyProducesResult() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Or(second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + + [Fact] + public void DeliversAnErrorFromAnySource() + { + Exception? error = null; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Or(second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/SortFixture.cs b/src/DynamicData.Tests/Cache/SortFixture.cs index 0f228b173..238beeed7 100644 --- a/src/DynamicData.Tests/Cache/SortFixture.cs +++ b/src/DynamicData.Tests/Cache/SortFixture.cs @@ -1,4 +1,4 @@ -#region +#region using System; using System.Collections.Generic; @@ -13,6 +13,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Concurrency; +using System.Reactive.Linq; #endregion @@ -1038,4 +1040,30 @@ public class Comparer : IComparer public int Compare(ViewModel? x, ViewModel? y) => StringComparer.OrdinalIgnoreCase.Compare(x?.Name, y?.Name); } } + + [Fact] + public void CompletesWhenGivenAComparerObservable() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.Sort(Observable.Return(_comparer)).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent resort signal can never fire and so must not hold the result open"); + } + + [Fact] + public void CompletesWhenGivenAResorter() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.Sort(_comparer, Observable.Never().Take(0)).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent comparer stream can never fire and so must not hold the result open"); + } } diff --git a/src/DynamicData.Tests/Cache/TerminalEventFixture.cs b/src/DynamicData.Tests/Cache/TerminalEventFixture.cs new file mode 100644 index 000000000..46d670c2e --- /dev/null +++ b/src/DynamicData.Tests/Cache/TerminalEventFixture.cs @@ -0,0 +1,72 @@ +using System; +using System.Reactive.Concurrency; +using System.Reactive.Subjects; + +using DynamicData.Tests.Domain; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.Cache; + +/// +/// Terminal-event behaviour for operators that subscribe internally without exposing the +/// subscription. A missing error handler on one of those internal subscriptions does not fail +/// loudly: the exception is rethrown wherever delivery happened to be, and the subscriber is left +/// believing the sequence ended normally. +/// +public class TerminalEventFixture +{ + [Fact] + public void WatchFailsWhenTheSourceFails() + { + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + var actualError = default(Exception); + var isCompleted = false; + using var subscription = cache.Watch("Name1").Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true); + + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + actualError.Should().BeSameAs(expectedError, "a watcher should be told when the source fails"); + isCompleted.Should().BeFalse("the source failed, it did not complete"); + } + + [Fact] + public void WatchCompletesWhenTheSourceCompletes() + { + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + var actualError = default(Exception); + var isCompleted = false; + using var subscription = cache.Watch("Name1").Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true); + + source.OnCompleted(); + + isCompleted.Should().BeTrue("the source completed"); + actualError.Should().BeNull("no error occurred"); + } + + [Fact] + public void LimitSizeToCompletesWhenTheSourceIsDisposed() + { + // The eviction stream is driven by source.Connect(). A plain SourceCache has no upstream that + // can fail, so only the completion path is reachable here, but the handler that carries it is + // the same one that now carries a failure for any ISourceCache implementation that does fail. + // Completion also travels the scheduler now rather than firing ahead of whatever it has queued. + var source = new SourceCache(p => p.Key); + + var actualError = default(Exception); + var isCompleted = false; + using var subscription = source.LimitSizeTo(10, Scheduler.Immediate).Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true); + + source.Dispose(); + + isCompleted.Should().BeTrue("the source completed"); + actualError.Should().BeNull("no error occurred"); + } +} diff --git a/src/DynamicData.Tests/Cache/TransformTreeFixture.cs b/src/DynamicData.Tests/Cache/TransformTreeFixture.cs index 33bf44573..b50656b21 100644 --- a/src/DynamicData.Tests/Cache/TransformTreeFixture.cs +++ b/src/DynamicData.Tests/Cache/TransformTreeFixture.cs @@ -6,6 +6,9 @@ using FluentAssertions; using Xunit; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -305,4 +308,30 @@ public override bool Equals(object? obj) public override string ToString() => $"Name: {Name}, Id: {Id}, BossId: {BossId}"; } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.TransformToTree(p => p.Name).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("the intermediate caches must not swallow the terminal event"); + } + + [Fact] + public void DeliversTheError() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source.TransformToTree(p => p.Name).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/XorFixture.cs b/src/DynamicData.Tests/Cache/XorFixture.cs index 8a320430a..f76229aa5 100644 --- a/src/DynamicData.Tests/Cache/XorFixture.cs +++ b/src/DynamicData.Tests/Cache/XorFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using DynamicData.Tests.Domain; @@ -6,6 +6,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -91,4 +93,34 @@ public void UpdatingOneSourceOnlyProducesResult() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Xor(second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + + [Fact] + public void DeliversAnErrorFromAnySource() + { + Exception? error = null; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Xor(second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData/Binding/BindPaged.cs b/src/DynamicData/Binding/BindPaged.cs index 1cbe39d35..5c02c4b04 100644 --- a/src/DynamicData/Binding/BindPaged.cs +++ b/src/DynamicData/Binding/BindPaged.cs @@ -46,15 +46,45 @@ private IObservable> UseContextSortOptions() => // making the comparedChanged observable to fire. Probably a deadlock var changesSubject = new Subject>(); var comparerSubject = new ReplaySubject>(1); + var bound = false; + + // Until the first element has supplied the binding options there is nothing standing between the + // source and the observer, so terminal events have to reach the observer directly. + void Fail(Exception error) + { + if (bound) + { + changesSubject.OnError(error); + } + else + { + observer.OnError(error); + } + } + + void Finish() + { + if (bound) + { + changesSubject.OnCompleted(); + } + else + { + observer.OnCompleted(); + } + } // once we have the initial values, publish as normal. var subsequent = shared .Skip(1) - .Subscribe(changesWithContext => - { - comparerSubject.OnNext(changesWithContext.Context.Comparer); - changesSubject.OnNext(changesWithContext); - }); + .Subscribe( + changesWithContext => + { + comparerSubject.OnNext(changesWithContext.Context.Comparer); + changesSubject.OnNext(changesWithContext); + }, + Fail, + Finish); // extract binding options from the page context var initial = shared @@ -72,9 +102,12 @@ private IObservable> UseContextSortOptions() => .SortAndBind(targetList, comparerSubject.DistinctUntilChanged(), extractedOptions) .SubscribeSafe(observer); + bound = true; + comparerSubject.OnNext(changesWithContext.Context.Comparer); changesSubject.OnNext(changesWithContext); - }); + }, + static _ => { }); return new CompositeDisposable(initial, subscriber, subsequent, shared.Connect()); }); diff --git a/src/DynamicData/Binding/BindVirtualized.cs b/src/DynamicData/Binding/BindVirtualized.cs index 37a3e1739..97d2b5d5b 100644 --- a/src/DynamicData/Binding/BindVirtualized.cs +++ b/src/DynamicData/Binding/BindVirtualized.cs @@ -44,15 +44,45 @@ private IObservable> UseVirtualSortOptions() => // making the comparedChanged observable to fire. Probably a deadlock var changesSubject = new Subject>(); var comparerSubject = new ReplaySubject>(1); + var bound = false; + + // Until the first element has supplied the binding options there is nothing standing between the + // source and the observer, so terminal events have to reach the observer directly. + void Fail(Exception error) + { + if (bound) + { + changesSubject.OnError(error); + } + else + { + observer.OnError(error); + } + } + + void Finish() + { + if (bound) + { + changesSubject.OnCompleted(); + } + else + { + observer.OnCompleted(); + } + } // once we have the initial values, publish as normal. var subsequent = shared .Skip(1) - .Subscribe(changesWithContext => - { - comparerSubject.OnNext(changesWithContext.Context.Comparer); - changesSubject.OnNext(changesWithContext); - }); + .Subscribe( + changesWithContext => + { + comparerSubject.OnNext(changesWithContext.Context.Comparer); + changesSubject.OnNext(changesWithContext); + }, + Fail, + Finish); // extract binding options from the virtual context var initial = shared @@ -70,9 +100,12 @@ private IObservable> UseVirtualSortOptions() => .SortAndBind(targetList, comparerSubject.DistinctUntilChanged(), extractedOptions) .SubscribeSafe(observer); + bound = true; + comparerSubject.OnNext(changesWithContext.Context.Comparer); changesSubject.OnNext(changesWithContext); - }); + }, + static _ => { }); return new CompositeDisposable(initial, subscriber, subsequent, shared.Connect()); }); diff --git a/src/DynamicData/Binding/SortAndBind.cs b/src/DynamicData/Binding/SortAndBind.cs index 0bb5fe208..c5b230dea 100644 --- a/src/DynamicData/Binding/SortAndBind.cs +++ b/src/DynamicData/Binding/SortAndBind.cs @@ -70,11 +70,13 @@ public SortAndBind(IObservable> source, // Create a new sort applicator each time. var latestComparer = comparerChanged.SynchronizeSafe(queue) - .Subscribe(comparer => - { - sortApplicator = new SortApplicator(_cache, target, comparer, options); - sortApplicator.ApplySort(); - }); + .Subscribe( + comparer => + { + sortApplicator = new SortApplicator(_cache, target, comparer, options); + sortApplicator.ApplySort(); + }, + observer.OnError); // Listen to changes and apply the sorting var subscriber = source.SynchronizeSafe(queue) diff --git a/src/DynamicData/Cache/Internal/BatchIf.cs b/src/DynamicData/Cache/Internal/BatchIf.cs index 76af20a61..7988ba6e1 100644 --- a/src/DynamicData/Cache/Internal/BatchIf.cs +++ b/src/DynamicData/Cache/Internal/BatchIf.cs @@ -55,7 +55,8 @@ IDisposable IntervalFunction() => { paused = true; } - }); + }, + observer.OnError); if (intervalTimer is not null) { @@ -83,9 +84,11 @@ IDisposable IntervalFunction() => { paused = false; ResumeAction(); - }); + }, + observer.OnError); } - }); + }, + observer.OnError); var publisher = _source.SynchronizeSafe(queue).Subscribe( changes => @@ -97,6 +100,13 @@ IDisposable IntervalFunction() => { ResumeAction(); } + }, + observer.OnError, + () => + { + // Anything still held back would otherwise be lost, so flush before finishing. + ResumeAction(); + observer.OnCompleted(); }); return new CompositeDisposable(publisher, pausedHandler, timeoutDisposer, intervalTimerDisposer, queue); diff --git a/src/DynamicData/Cache/Internal/Combiner.cs b/src/DynamicData/Cache/Internal/Combiner.cs index c7734512d..aa3a7e295 100644 --- a/src/DynamicData/Cache/Internal/Combiner.cs +++ b/src/DynamicData/Cache/Internal/Combiner.cs @@ -2,8 +2,11 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. +using System.Reactive; using System.Reactive.Disposables; +using DynamicData.Internal; + namespace DynamicData.Cache.Internal; /// @@ -23,8 +26,25 @@ internal sealed class Combiner(CombineOperator type, Action> _sourceCaches = []; - public IDisposable Subscribe(IObservable>[] source) + public IDisposable Subscribe(IObservable>[] source, Action onError, Action onCompleted) { + // Merging semantics: the result finishes only once every source has. + var pending = source.Length; + if (pending == 0) + { + onCompleted(); + return Disposable.Empty; + } + + // Each source updates shared state under _locker, but delivery has to be serialized too: + // without this, two sources can compute their notifications, leave the lock, and then both + // be inside updatedCallback at the same time. The queue takes the notification while the + // lock is held and drains it after the lock is released, so deliveries stay ordered and + // one at a time without a subscriber being able to block a producer. + var queue = new DeliveryQueue>( + _locker, + Observer.Create(updatedCallback, onError, onCompleted)); + // subscribe var disposable = new CompositeDisposable(); lock (_locker) @@ -34,11 +54,25 @@ public IDisposable Subscribe(IObservable>[] source) foreach (var pair in source.Zip(_sourceCaches, (item, cache) => new { Item = item, Cache = cache })) { - var subscription = pair.Item.Subscribe(updates => Update(pair.Cache, updates)); + var subscription = pair.Item.Subscribe( + updates => Update(queue, pair.Cache, updates), + queue.OnError, + () => + { + if (Interlocked.Decrement(ref pending) == 0) + { + queue.OnCompleted(); + } + }); + disposable.Add(subscription); } } + // Queue last: the subscriptions are torn down first, so any terminal notification still in + // flight is delivered through a queue that is still running. + disposable.Add(queue); + return disposable; } @@ -73,22 +107,19 @@ private bool MatchesConstraint(TKey key) } } - private void Update(Cache cache, IChangeSet updates) + private void Update(DeliveryQueue> queue, Cache cache, IChangeSet updates) { - ChangeSet notifications; + using var scope = queue.AcquireLock(); - lock (_locker) - { - // update cache for the individual source - cache.Clone(updates); + // update cache for the individual source + cache.Clone(updates); - // update combined - notifications = UpdateCombined(updates); - } + // update combined + var notifications = UpdateCombined(updates); if (notifications.Count != 0) { - updatedCallback(notifications); + scope.EnqueueNext(notifications); } } diff --git a/src/DynamicData/Cache/Internal/DynamicCombiner.cs b/src/DynamicData/Cache/Internal/DynamicCombiner.cs index 41c7e3a8d..2793efafd 100644 --- a/src/DynamicData/Cache/Internal/DynamicCombiner.cs +++ b/src/DynamicData/Cache/Internal/DynamicCombiner.cs @@ -39,7 +39,9 @@ public IObservable> Run() => Observable.Create> Run() => Observable.Create { }, static _ => { }); // when an list is added or removed, need to var sourceChanged = sharedLists.WhereReasonsAre(ListChangeReason.Add, ListChangeReason.AddRange).ForEachItemChange( @@ -77,7 +80,8 @@ public IObservable> Run() => Observable.Create { }, static _ => { }); return new CompositeDisposable(sourceLists, allChanges, removedItem, sourceChanged, sharedLists.Connect(), queue); }); diff --git a/src/DynamicData/Cache/Internal/GroupOn.cs b/src/DynamicData/Cache/Internal/GroupOn.cs index fd0936fe7..075c9ab78 100644 --- a/src/DynamicData/Cache/Internal/GroupOn.cs +++ b/src/DynamicData/Cache/Internal/GroupOn.cs @@ -15,7 +15,9 @@ internal sealed class GroupOn(IObservable _groupSelectorKey = groupSelectorKey ?? throw new ArgumentNullException(nameof(groupSelectorKey)); - private readonly IObservable _regrouper = regrouper ?? Observable.Never(); + // An absent regrouper means no regroup signal will ever arrive. Never would say one still might, + // which leaves the merge below unable to complete when the source does. + private readonly IObservable _regrouper = regrouper ?? Observable.Empty(); private readonly IObservable> _source = source ?? throw new ArgumentNullException(nameof(source)); @@ -25,13 +27,14 @@ public IObservable> Run() => Observabl var queue = new SharedDeliveryQueue(); var grouper = new Grouper(_groupSelectorKey); - var groups = _source.SynchronizeSafe(queue).Finally(observer.OnCompleted).Select(grouper.Update).Where(changes => changes.Count != 0); + var groups = _source.SynchronizeSafe(queue).Select(grouper.Update).Where(changes => changes.Count != 0); var regroup = _regrouper.SynchronizeSafe(queue).Select(_ => grouper.Regroup()).Where(changes => changes.Count != 0); var published = groups.Merge(regroup).Publish(); var subscriber = published.SubscribeSafe(observer); - var disposer = published.DisposeMany().Subscribe(); + + var disposer = published.DisposeMany().Subscribe(static _ => { }, static _ => { }); var connected = published.Connect(); diff --git a/src/DynamicData/Cache/Internal/GroupOnImmutable.cs b/src/DynamicData/Cache/Internal/GroupOnImmutable.cs index ba33db031..3ee2ee0d2 100644 --- a/src/DynamicData/Cache/Internal/GroupOnImmutable.cs +++ b/src/DynamicData/Cache/Internal/GroupOnImmutable.cs @@ -15,7 +15,9 @@ internal sealed class GroupOnImmutable(IObservable _groupSelectorKey = groupSelectorKey ?? throw new ArgumentNullException(nameof(groupSelectorKey)); - private readonly IObservable _regrouper = regrouper ?? Observable.Never(); + // An absent regrouper means no regroup signal will ever arrive. Never would say one still might, + // which leaves the merge below unable to complete when the source does. + private readonly IObservable _regrouper = regrouper ?? Observable.Empty(); private readonly IObservable> _source = source ?? throw new ArgumentNullException(nameof(source)); diff --git a/src/DynamicData/Cache/Internal/MergeMany.cs b/src/DynamicData/Cache/Internal/MergeMany.cs index fe2eecc75..fa3a00ac3 100644 --- a/src/DynamicData/Cache/Internal/MergeMany.cs +++ b/src/DynamicData/Cache/Internal/MergeMany.cs @@ -46,9 +46,9 @@ public IObservable Run() => Observable.Create( Interlocked.Increment(ref counter.Value); return _observableSelector(t, key) .Finally(() => CheckCompleted(counter, queue)) - .Subscribe(queue.OnNext, static _ => { }); + .Subscribe(queue.OnNext, queue.OnError); }) - .Subscribe(static _ => { }, observer.OnError)); + .Subscribe(static _ => { }, queue.OnError)); }); private static void CheckCompleted(StrongBox counter, DeliveryQueue queue) diff --git a/src/DynamicData/Cache/Internal/MergeManyItems.cs b/src/DynamicData/Cache/Internal/MergeManyItems.cs index 26b85614b..ccf6050d3 100644 --- a/src/DynamicData/Cache/Internal/MergeManyItems.cs +++ b/src/DynamicData/Cache/Internal/MergeManyItems.cs @@ -31,5 +31,10 @@ public MergeManyItems(IObservable> source, Func observableSelector(t); } - public IObservable> Run() => Observable.Create>(observer => _source.SubscribeMany((t, v) => _observableSelector(t, v).Select(z => new ItemWithValue(t, z)).SubscribeSafe(observer)).Subscribe()); + // MergeMany already tracks the parent and every child subscription so that the result finishes only once + // all of them have. Reusing it keeps a child completing from terminating the whole stream. + public IObservable> Run() => + new MergeMany>( + _source, + (t, v) => _observableSelector(t, v).Select(z => new ItemWithValue(t, z))).Run(); } diff --git a/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs b/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs index bd7b18f4d..c97a8f28d 100644 --- a/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs +++ b/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs @@ -46,11 +46,17 @@ protected override void ParentOnNext(IChangeSet> Run() => Observable.Create result is not null).Select(x => x!).SubscribeSafe(observer); } - var comparerChanged = (_comparerChangedObservable ?? Observable.Never>()).SynchronizeSafe(queue).Select(sorter.Sort); + // An absent comparer or resort signal will never fire, so it must not hold the merge open. + var comparerChanged = (_comparerChangedObservable ?? Observable.Empty>()).SynchronizeSafe(queue).Select(sorter.Sort); - var sortAgain = (_resorter ?? Observable.Never()).SynchronizeSafe(queue).Select(_ => sorter.Sort()); + var sortAgain = (_resorter ?? Observable.Empty()).SynchronizeSafe(queue).Select(_ => sorter.Sort()); var dataChanged = _source.SynchronizeSafe(queue).Select(sorter.Sort); diff --git a/src/DynamicData/Cache/Internal/SpecifiedGrouper.cs b/src/DynamicData/Cache/Internal/SpecifiedGrouper.cs index 863f2d6d6..493bb1c9a 100644 --- a/src/DynamicData/Cache/Internal/SpecifiedGrouper.cs +++ b/src/DynamicData/Cache/Internal/SpecifiedGrouper.cs @@ -23,8 +23,9 @@ public IObservable> Run() => Observabl { var queue = new SharedDeliveryQueue(); - // create source group cache - var sourceGroups = _source.SynchronizeSafe(queue).Group(_groupSelector).DisposeMany().AsObservableCache(); + // create source group cache. The observer is fed from the result group source below, so the + // source's own terminal events would otherwise never reach it. + var sourceGroups = _source.SynchronizeSafe(queue).Do(static _ => { }, observer.OnError, observer.OnCompleted).Group(_groupSelector).DisposeMany().AsObservableCache(); // create parent groups var parentGroups = _resultGroupSource.SynchronizeSafe(queue).Transform( @@ -53,7 +54,7 @@ public IObservable> Run() => Observabl { groupToUpdate.Value.Update(updater => updater.Clone(updates)); } - })).DisposeMany().Subscribe(); + })).DisposeMany().Subscribe(static _ => { }, static _ => { }); var notifier = parentGroups.Connect().Select( x => diff --git a/src/DynamicData/Cache/Internal/StatusMonitor.cs b/src/DynamicData/Cache/Internal/StatusMonitor.cs index 445074b63..84c9dfc0c 100644 --- a/src/DynamicData/Cache/Internal/StatusMonitor.cs +++ b/src/DynamicData/Cache/Internal/StatusMonitor.cs @@ -1,60 +1,17 @@ -// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved. +// 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.Reactive.Disposables; using System.Reactive.Linq; -using System.Reactive.Subjects; namespace DynamicData.Cache.Internal; internal sealed class StatusMonitor(IObservable source) { - public IObservable Run() => Observable.Create( - observer => - { - var statusSubject = new Subject(); - var status = ConnectionStatus.Pending; - - void Error(Exception ex) - { - status = ConnectionStatus.Errored; - statusSubject.OnNext(status); - observer.OnError(ex); - } - - void Completion() - { - if (status == ConnectionStatus.Errored) - { - return; - } - - status = ConnectionStatus.Completed; - statusSubject.OnNext(status); - } - - void Updated() - { - if (status != ConnectionStatus.Pending) - { - return; - } - - status = ConnectionStatus.Loaded; - statusSubject.OnNext(status); - } - - var monitor = source.Subscribe(_ => Updated(), Error, Completion); - - var subscriber = statusSubject.StartWith(status).DistinctUntilChanged().SubscribeSafe(observer); - - return Disposable.Create( - () => - { - statusSubject.OnCompleted(); - monitor.Dispose(); - subscriber.Dispose(); - }); - }); + public IObservable Run() => + source.Select(static _ => ConnectionStatus.Loaded) + .Concat(Observable.Return(ConnectionStatus.Completed)) + .Catch(static error => Observable.Return(ConnectionStatus.Errored).Concat(Observable.Throw(error))) + .StartWith(ConnectionStatus.Pending) + .DistinctUntilChanged(); } diff --git a/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs b/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs index c7c95aa66..5955ec0f8 100644 --- a/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs +++ b/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs @@ -20,7 +20,7 @@ public IObservable> Run() => Observable.Create(); - var cacheLoader = shared.Subscribe(changes => cache.Clone(changes)); + var cacheLoader = shared.Subscribe(changes => cache.Clone(changes), static _ => { }); // create change set of items where force refresh is applied var refresher = forceTransform.SynchronizeSafe(queue).Select(selector => CaptureChanges(cache, selector)).Select(changes => new ChangeSet(changes)).NotEmpty(); diff --git a/src/DynamicData/Cache/Internal/TreeBuilder.cs b/src/DynamicData/Cache/Internal/TreeBuilder.cs index bf379ef99..c86134104 100644 --- a/src/DynamicData/Cache/Internal/TreeBuilder.cs +++ b/src/DynamicData/Cache/Internal/TreeBuilder.cs @@ -27,7 +27,8 @@ public IObservable, TKey>> Run() => Observable.Cr var queue = new SharedDeliveryQueue(); var reFilterObservable = new BehaviorSubject(Unit.Default); - var allData = _source.SynchronizeSafe(queue).AsObservableCache(); + // Terminal events do not survive the intermediate caches below, so relay them to the observer. + var allData = _source.SynchronizeSafe(queue).Do(static _ => { }, observer.OnError, observer.OnCompleted).AsObservableCache(); // for each object we need a node which provides // a structure to set the parent and children @@ -195,7 +196,7 @@ void UpdateChildren(Node parentNode) } reFilterObservable.OnNext(Unit.Default); - }).DisposeMany().Subscribe(); + }).DisposeMany().Subscribe(static _ => { }, static _ => { }); var filter = _predicateChanged.SynchronizeSafe(queue).CombineLatest(reFilterObservable, (predicate, _) => predicate); var result = allNodes.Connect().Filter(filter).SubscribeSafe(observer); diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 60642ee29..d40700e90 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -284,7 +284,7 @@ private IObservable> CreateWatchObservable(TKey key) => ? _changes.SkipWhile(_ => Volatile.Read(ref _currentDeliveryVersion) <= snapshotVersion) : _changes; - return changes.Finally(observer.OnCompleted).Subscribe( + return changes.Subscribe( changes => { foreach (var change in changes) @@ -295,7 +295,9 @@ private IObservable> CreateWatchObservable(TKey key) => observer.OnNext(change); } } - }); + }, + observer.OnError, + observer.OnCompleted); }); /// diff --git a/src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs b/src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs index e8733432c..3596227e7 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs @@ -72,7 +72,7 @@ public static IObservable> BatchIf(this /// UpdateBuffered while paused; forwarded immediately while active. /// RemoveBuffered while paused; forwarded immediately while active. /// RefreshBuffered while paused; forwarded immediately while active. - /// OnErrorBuffered data is lost. + /// OnErrorBuffered data is lost. An error from terminates the output the same way a source error does. /// OnCompletedAny remaining buffered data is flushed before completion. /// /// Worth noting: If the source completes while paused, buffered data IS flushed before OnCompleted. However, if the source errors while paused, buffered data is lost. diff --git a/src/DynamicData/Cache/ObservableCacheEx.Combine.cs b/src/DynamicData/Cache/ObservableCacheEx.Combine.cs index eb46b86b3..450d5a5d4 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.Combine.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.Combine.cs @@ -1,4 +1,4 @@ -// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved. +// 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. @@ -89,12 +89,11 @@ void UpdateAction(IChangeSet updates) try { var combiner = new Combiner(type, UpdateAction); - subscriber = combiner.Subscribe([.. sources]); + subscriber = combiner.Subscribe([.. sources], observer.OnError, observer.OnCompleted); } catch (Exception ex) { observer.OnError(ex); - observer.OnCompleted(); } return subscriber; @@ -119,7 +118,6 @@ void UpdateAction(IChangeSet updates) catch (Exception ex) { observer.OnError(ex); - observer.OnCompleted(); } } @@ -130,12 +128,11 @@ void UpdateAction(IChangeSet updates) list.Insert(0, source); var combiner = new Combiner(type, UpdateAction); - subscriber = combiner.Subscribe([.. list]); + subscriber = combiner.Subscribe([.. list], observer.OnError, observer.OnCompleted); } catch (Exception ex) { observer.OnError(ex); - observer.OnCompleted(); } return subscriber; diff --git a/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs b/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs index 810f118d0..ea832e226 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs @@ -88,7 +88,7 @@ public static IObservable>> LimitSizeTo< long orderItemWasAdded = -1; var sizeLimiter = new SizeLimiter(sizeLimit); - return source.Connect().Finally(observer.OnCompleted).ObserveOn(scheduler ?? GlobalConfig.DefaultScheduler).Transform((t, v) => new ExpirableItem(t, v, DateTime.Now, Interlocked.Increment(ref orderItemWasAdded))).Select(sizeLimiter.CloneAndReturnExpiredOnly).Where(expired => expired.Length != 0).Subscribe( + return source.Connect().ObserveOn(scheduler ?? GlobalConfig.DefaultScheduler).Transform((t, v) => new ExpirableItem(t, v, DateTime.Now, Interlocked.Increment(ref orderItemWasAdded))).Select(sizeLimiter.CloneAndReturnExpiredOnly).Where(expired => expired.Length != 0).Subscribe( toRemove => { try @@ -100,7 +100,9 @@ public static IObservable>> LimitSizeTo< { observer.OnError(ex); } - }); + }, + observer.OnError, + observer.OnCompleted); }); } } diff --git a/src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs b/src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs index b3c290d34..90e0c88e4 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs @@ -49,7 +49,8 @@ public static partial class ObservableCacheEx /// UpdateDisposes the previous child subscription and creates a new one for the updated item. /// RemoveDisposes the child subscription for the removed item. /// RefreshNo effect on subscriptions. The child observable continues unchanged. - /// OnErrorErrors from child observables are silently swallowed (the child is unsubscribed). Errors from the source changeset stream terminate the merged output. + /// OnErrorAn error from a child observable, or from the source changeset stream, terminates the merged output. + /// OnCompletedThe output completes once the source changeset stream has completed and every active child observable has completed. A child completing on its own does not end the merged output. /// /// Worth noting: The output is a plain , not a changeset stream. If you need merged changesets, use instead. /// diff --git a/src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs b/src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs index 5fcbb9ed0..347f8ec32 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs @@ -148,7 +148,7 @@ public static IObservable> MergeManyCh /// /// /// EventBehavior - /// OnErrorAn error from the source (parent) stream or from any child changeset stream terminates the entire output. Unlike , child errors are NOT swallowed. + /// OnErrorAn error from the source (parent) stream or from any child changeset stream terminates the entire output. /// OnCompletedThe output completes when the source (parent) stream completes and all active child changeset streams have also completed. /// /// diff --git a/src/DynamicData/Cache/ObservableCacheEx.PopulateInto.cs b/src/DynamicData/Cache/ObservableCacheEx.PopulateInto.cs index d5ce99d73..ce49086f4 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.PopulateInto.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.PopulateInto.cs @@ -43,7 +43,7 @@ public static partial class ObservableCacheEx /// UpdateThe item is updated in the destination cache via AddOrUpdate. /// RemoveThe item is removed from the destination cache. /// RefreshA Refresh is issued on the destination cache for the item. - /// OnErrorThe subscription is terminated. The destination cache is not rolled back. + /// OnErrorThe subscription ends and the destination cache is not rolled back. Since this method is a sink with no downstream observer, the exception is rethrown by Rx on whichever thread delivered it. Subscribe to directly if you need to handle the failure. /// OnCompletedThe subscription ends. The destination cache retains all items. /// ///