From 6f16300a8db6c665f58574287efa846d73e1e158 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 07:35:43 -0700 Subject: [PATCH 1/7] Make the list operators honour the Rx observable contract The companion to the cache changes. The conformance sweep that found these ships with that PR; this one carries the list side of what it reported, plus a defect the sweep could not reach. MergeChangeSets(IComparer) called itself. A lone IComparer argument is an exact match for the method's own signature, so the call bound back to the method making it and exhausted the stack before returning. Naming the arguments reaches the intended overload. The IObservableList overload funnels into the same method and went the same way. Neither could be covered by the sweep, because a stack overflow cannot be caught. Sort, GroupOn and GroupOnImmutable fell back to Observable.Never for control signals the caller did not supply, then merged them with the data. Never says a signal might still arrive, so the merge could never complete. Empty is what is meant when nothing was supplied. QueryWhenChanged built its result by hand inside Observable.Create and forwarded only OnNext, dropping terminal events. Expressed as Defer and Select it keeps its per subscription list and propagates terminal events for free. ToCollection, ToSortedCollection, Maximum and Minimum are built on it and are fixed as a consequence. BufferIf subscribed without an error or completion handler. It now propagates them, and flushes anything still buffered by a pause first. The MergeManyChangeSets operators subscribe to their shared source twice, and the second subscription had no error handler, so Rx rethrew the source's error out of the subscription even though the first had already reported it. MergeMany discarded errors raised by its inner sequences. Rx's Merge propagates them, and DynamicCombiner inherits the fix. One existing test asserted the old behaviour and has been rewritten to the contract. And, Or, Except and Xor now complete once every source has and fail as soon as any one does. --- .../List/CombinerCompletionFixture.cs | 57 +++++ .../List/MergeManyFixture.cs | 6 +- .../List/MergeManyInnerErrorFixture.cs | 32 +++ .../List/OperatorCompletionFixture.cs | 203 ++++++++++++++++++ src/DynamicData/List/Internal/BufferIf.cs | 12 ++ src/DynamicData/List/Internal/Combiner.cs | 16 ++ .../List/Internal/DynamicCombiner.cs | 16 +- src/DynamicData/List/Internal/GroupOn.cs | 3 +- .../List/Internal/GroupOnImmutable.cs | 3 +- src/DynamicData/List/Internal/MergeMany.cs | 2 +- .../List/Internal/MergeManyCacheChangeSets.cs | 4 +- .../List/Internal/MergeManyListChangeSets.cs | 4 +- .../List/Internal/QueryWhenChanged.cs | 18 +- src/DynamicData/List/Internal/Sort.cs | 5 +- .../List/ObservableListEx.MergeChangeSets.cs | 4 +- 15 files changed, 361 insertions(+), 24 deletions(-) create mode 100644 src/DynamicData.Tests/List/CombinerCompletionFixture.cs create mode 100644 src/DynamicData.Tests/List/MergeManyInnerErrorFixture.cs create mode 100644 src/DynamicData.Tests/List/OperatorCompletionFixture.cs diff --git a/src/DynamicData.Tests/List/CombinerCompletionFixture.cs b/src/DynamicData.Tests/List/CombinerCompletionFixture.cs new file mode 100644 index 000000000..2f2ba3a69 --- /dev/null +++ b/src/DynamicData.Tests/List/CombinerCompletionFixture.cs @@ -0,0 +1,57 @@ +using System; +using System.Reactive.Linq; +using System.Reactive.Subjects; + +using DynamicData.Tests.Domain; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.List; + +/// +/// Terminal event behaviour for the combining operators. +/// +public class CombinerCompletionFixture +{ + [Fact] + public void CombinersCompleteWhenEverySourceCompletes() + { + foreach (var combine in new Func>, IObservable>, IObservable>>[] + { + static (a, b) => ObservableListEx.And(a, b), + static (a, b) => a.Or(b), + static (a, b) => a.Except(b), + static (a, b) => a.Xor(b), + }) + { + using var first = new Subject>(); + using var second = new Subject>(); + var completed = false; + + using var subscription = combine(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 CombinersDeliverErrorFromAnySource() + { + using var first = new Subject>(); + using var second = new Subject>(); + Exception? error = null; + + using var subscription = first.Or(second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + +} diff --git a/src/DynamicData.Tests/List/MergeManyFixture.cs b/src/DynamicData.Tests/List/MergeManyFixture.cs index 997e91f9f..57cfc3d6f 100644 --- a/src/DynamicData.Tests/List/MergeManyFixture.cs +++ b/src/DynamicData.Tests/List/MergeManyFixture.cs @@ -114,7 +114,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; @@ -129,9 +129,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/List/MergeManyInnerErrorFixture.cs b/src/DynamicData.Tests/List/MergeManyInnerErrorFixture.cs new file mode 100644 index 000000000..003525b6e --- /dev/null +++ b/src/DynamicData.Tests/List/MergeManyInnerErrorFixture.cs @@ -0,0 +1,32 @@ +using System; +using System.Reactive.Subjects; + +using DynamicData.Tests.Domain; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.List; + +/// +/// Merge propagates a failure from any inner stream, rather than discarding it. +/// +public class MergeManyInnerErrorFixture +{ + [Fact] + public void MergeManyDeliversErrorFromAChild() + { + using var source = new SourceList(); + using var child = new Subject(); + Exception? error = null; + + using var subscription = source.Connect().MergeMany(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.Add(new Person("a", 1)); + child.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType("a failing inner stream must not be silently discarded"); + } + +} diff --git a/src/DynamicData.Tests/List/OperatorCompletionFixture.cs b/src/DynamicData.Tests/List/OperatorCompletionFixture.cs new file mode 100644 index 000000000..2ea58a445 --- /dev/null +++ b/src/DynamicData.Tests/List/OperatorCompletionFixture.cs @@ -0,0 +1,203 @@ +using System; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; + +using DynamicData.Aggregation; +using DynamicData.Binding; +using DynamicData.Tests.Domain; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.List; + +/// +/// Terminal event behaviour for list operators which previously never delivered OnCompleted. +/// +public class OperatorCompletionFixture +{ + private static readonly IComparer ByName = SortExpressionComparer.Ascending(p => p.Name); + + private static ChangeSet OneAdd() => [new Change(ListChangeReason.Add, new Person("a", 1), 0)]; + + [Fact] + public void QueryWhenChangedCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.QueryWhenChanged().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void QueryWhenChangedDeliversError() + { + using var source = new Subject>(); + Exception? error = null; + + using var subscription = source.QueryWhenChanged().Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + + [Fact] + public void ToCollectionCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.ToCollection().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void ToSortedCollectionCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.ToSortedCollection(ByName).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void GroupOnCompletesWhenNoRegrouperIsSupplied() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.GroupOn(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 GroupWithImmutableStateCompletesWhenNoRegrouperIsSupplied() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.GroupWithImmutableState(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void SortCompletesWhenGivenAComparerObservable() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.Sort(Observable.Return(ByName)).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void BufferIfCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.BufferIf(Observable.Return(false), Scheduler.Immediate).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void BufferIfFlushesHeldChangesBeforeCompleting() + { + using var source = new Subject>(); + var received = 0; + var completed = false; + + using var subscription = source.BufferIf(Observable.Return(true), Scheduler.Immediate).Subscribe(_ => received++, () => completed = true); + + source.OnNext(OneAdd()); + source.OnCompleted(); + + received.Should().Be(1, "changes held back by the pause would otherwise be lost"); + completed.Should().BeTrue(); + } + + [Fact] + public void MaximumCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void MaximumDeliversErrorWithoutThrowing() + { + using var source = new Subject>(); + Exception? error = null; + + using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + + [Fact] + public void MergeManyChangeSetsDeliversErrorWithoutThrowing() + { + using var source = new Subject>(); + Exception? error = null; + + using var subscription = source + .MergeManyChangeSets(_ => Observable.Empty>(), EqualityComparer.Default) + .Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + + [Fact] + public void MergeChangeSetsWithComparerDoesNotRecurse() + { + using var sources = new SourceList>>(); + var completed = false; + + // This used to bind to itself and exhaust the stack before returning. + using var subscription = sources.Connect() + .MergeChangeSets(SortExpressionComparer.Ascending(p => p.Name)) + .Subscribe(_ => { }, () => completed = true); + + sources.Dispose(); + + completed.Should().BeTrue(); + } +} diff --git a/src/DynamicData/List/Internal/BufferIf.cs b/src/DynamicData/List/Internal/BufferIf.cs index 8f7d1993c..4e42b3b6a 100644 --- a/src/DynamicData/List/Internal/BufferIf.cs +++ b/src/DynamicData/List/Internal/BufferIf.cs @@ -72,6 +72,18 @@ public IObservable> Run() => Observable.Create>( { observer.OnNext(updates); } + }, + observer.OnError, + () => + { + // Anything still buffered would otherwise be lost, so flush before finishing. + if (buffer.Count > 0) + { + observer.OnNext(buffer); + buffer = []; + } + + observer.OnCompleted(); }); var connected = bufferSelector.Connect(); diff --git a/src/DynamicData/List/Internal/Combiner.cs b/src/DynamicData/List/Internal/Combiner.cs index 64882a3ba..f74e66941 100644 --- a/src/DynamicData/List/Internal/Combiner.cs +++ b/src/DynamicData/List/Internal/Combiner.cs @@ -27,6 +27,14 @@ public IObservable> Run() => Observable.Create>( var resultList = new ChangeAwareListWithRefCounts(); + // Merging semantics: the result finishes only once every source has. + var pending = _source.Count; + if (pending == 0) + { + observer.OnCompleted(); + return disposable; + } + lock (_locker) { var sourceLists = Enumerable.Range(0, _source.Count).Select(_ => new ReferenceCountTracker()).ToList(); @@ -44,6 +52,14 @@ public IObservable> Run() => Observable.Create>( { observer.OnNext(notifications); } + }, + observer.OnError, + () => + { + if (Interlocked.Decrement(ref pending) == 0) + { + observer.OnCompleted(); + } })); } } diff --git a/src/DynamicData/List/Internal/DynamicCombiner.cs b/src/DynamicData/List/Internal/DynamicCombiner.cs index bb3091378..1f196d3e4 100644 --- a/src/DynamicData/List/Internal/DynamicCombiner.cs +++ b/src/DynamicData/List/Internal/DynamicCombiner.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. @@ -40,7 +40,9 @@ public IObservable> Run() => Observable.Create>( { observer.OnNext(notifications); } - }); + }, + observer.OnError, + observer.OnCompleted); // When a list is removed, update all items that were in that list var removedItem = sourceLists.Connect().OnItemRemoved( @@ -63,7 +65,10 @@ public IObservable> Run() => Observable.Create>( observer.OnNext(notification2); } } - }).Subscribe(); + }) + // The merge subscription above reports errors to the observer. Without a handler here Rx + // would rethrow them out of the subscription as well. + .Subscribe(static _ => { }, static _ => { }); // When a list is added, update all items that are in that list var sourceChanged = sourceLists.Connect().WhereReasonsAre(ListChangeReason.Add, ListChangeReason.AddRange).ForEachItemChange( @@ -84,7 +89,10 @@ public IObservable> Run() => Observable.Create>( observer.OnNext(notification2); } } - }).Subscribe(); + }) + // The merge subscription above reports errors to the observer. Without a handler here Rx + // would rethrow them out of the subscription as well. + .Subscribe(static _ => { }, static _ => { }); return new CompositeDisposable(sourceLists, allChanges, removedItem, sourceChanged); }); diff --git a/src/DynamicData/List/Internal/GroupOn.cs b/src/DynamicData/List/Internal/GroupOn.cs index 6ea7deccd..620b05896 100644 --- a/src/DynamicData/List/Internal/GroupOn.cs +++ b/src/DynamicData/List/Internal/GroupOn.cs @@ -32,8 +32,9 @@ public IObservable>> Run() => Observable.C var grouper = shared.Select(changes => Process(groupings, groupCache, changes)); + // An absent regrouper never fires, so Empty rather than Never keeps the merge able to complete. var regrouperFunc = _regrouper is null ? - Observable.Never>>() : + Observable.Empty>>() : _regrouper.Synchronize(locker).CombineLatest(shared.ToCollection(), (_, collection) => Regroup(groupings, groupCache, collection)); var publisher = grouper.Merge(regrouperFunc).DisposeMany() // dispose removes as the grouping is disposable diff --git a/src/DynamicData/List/Internal/GroupOnImmutable.cs b/src/DynamicData/List/Internal/GroupOnImmutable.cs index 2b4a08be8..45a03aeeb 100644 --- a/src/DynamicData/List/Internal/GroupOnImmutable.cs +++ b/src/DynamicData/List/Internal/GroupOnImmutable.cs @@ -35,8 +35,9 @@ public IObservable>> Run() => Observabl var grouper = shared.Select(changes => Process(groupings, groupCache, changes)); + // An absent regrouper never fires, so Empty rather than Never keeps the merge able to complete. var reGroupFunc = _reGrouper is null ? - Observable.Never>>() : + Observable.Empty>>() : _reGrouper.Synchronize(locker).CombineLatest(shared.ToCollection(), (_, collection) => Regroup(groupings, groupCache, collection)); var publisher = grouper.Merge(reGroupFunc).NotEmpty().SubscribeSafe(observer); diff --git a/src/DynamicData/List/Internal/MergeMany.cs b/src/DynamicData/List/Internal/MergeMany.cs index 65b238068..40cb63e23 100644 --- a/src/DynamicData/List/Internal/MergeMany.cs +++ b/src/DynamicData/List/Internal/MergeMany.cs @@ -24,7 +24,7 @@ public IObservable Run() => Observable.Create( .SubscribeMany(t => { counter.Added(); - return _observableSelector(t).Synchronize(locker).Finally(() => counter.Finally()).Subscribe(observer.OnNext, _ => { }, () => { }); + return _observableSelector(t).Synchronize(locker).Finally(() => counter.Finally()).Subscribe(observer.OnNext, observer.OnError, () => { }); }) .Subscribe(_ => { }, observer.OnError, observer.OnCompleted); diff --git a/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs b/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs index 3b63ea231..1ab5ae85f 100644 --- a/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs +++ b/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs @@ -51,7 +51,9 @@ public IObservable> Run() => Observabl changeTracker.EmitChanges(observer); parentUpdate = false; }) - .Subscribe(); + // The subscription above already reports errors to the observer. Without a handler here Rx + // would rethrow them out of the subscription as well. + .Subscribe(static _ => { }, static _ => { }); return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove); }); diff --git a/src/DynamicData/List/Internal/MergeManyListChangeSets.cs b/src/DynamicData/List/Internal/MergeManyListChangeSets.cs index 482c4f46b..1767720dd 100644 --- a/src/DynamicData/List/Internal/MergeManyListChangeSets.cs +++ b/src/DynamicData/List/Internal/MergeManyListChangeSets.cs @@ -47,7 +47,9 @@ public IObservable> Run() => Observable.Create { }, static _ => { }); return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove); }); diff --git a/src/DynamicData/List/Internal/QueryWhenChanged.cs b/src/DynamicData/List/Internal/QueryWhenChanged.cs index ef58d4925..ab36511e6 100644 --- a/src/DynamicData/List/Internal/QueryWhenChanged.cs +++ b/src/DynamicData/List/Internal/QueryWhenChanged.cs @@ -11,14 +11,14 @@ internal sealed class QueryWhenChanged(IObservable> source) { private readonly IObservable> _source = source ?? throw new ArgumentNullException(nameof(source)); - public IObservable> Run() => Observable.Create>(observer => - { - var list = new List(); + public IObservable> Run() => Observable.Defer(() => + { + var list = new List(); - return _source.Subscribe(changes => - { - list.Clone(changes); - observer.OnNext(new ReadOnlyCollectionLight(list)); - }); - }); + return _source.Select(changes => + { + list.Clone(changes); + return (IReadOnlyCollection)new ReadOnlyCollectionLight(list); + }); + }); } diff --git a/src/DynamicData/List/Internal/Sort.cs b/src/DynamicData/List/Internal/Sort.cs index ff03e2e70..f366fe02d 100644 --- a/src/DynamicData/List/Internal/Sort.cs +++ b/src/DynamicData/List/Internal/Sort.cs @@ -10,8 +10,9 @@ namespace DynamicData.List.Internal; internal sealed class Sort(IObservable> source, IComparer? comparer, SortOptions sortOptions, IObservable? resort, IObservable>? comparerObservable, int resetThreshold) where T : notnull { - private readonly IObservable> _comparerObservable = comparerObservable ?? Observable.Never>(); - private readonly IObservable _resort = resort ?? Observable.Never(); + // An absent comparer or resort signal will never fire, so it must not hold the merge open. + private readonly IObservable> _comparerObservable = comparerObservable ?? Observable.Empty>(); + private readonly IObservable _resort = resort ?? Observable.Empty(); private readonly IObservable> _source = source ?? throw new ArgumentNullException(nameof(source)); private IComparer _comparer = comparer ?? Comparer.Default; diff --git a/src/DynamicData/List/ObservableListEx.MergeChangeSets.cs b/src/DynamicData/List/ObservableListEx.MergeChangeSets.cs index 4f58c05cc..42af3ba92 100644 --- a/src/DynamicData/List/ObservableListEx.MergeChangeSets.cs +++ b/src/DynamicData/List/ObservableListEx.MergeChangeSets.cs @@ -196,7 +196,9 @@ public static IObservable> MergeChangeSets is an exact match for this method's own + // signature, so an unnamed call binds here again and recurses until the stack is exhausted. + return source.MergeChangeSets(equalityComparer: null, comparer: comparer); } /// From a9f1501bdab37d3082ad5a71785ab98d56f5173f Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 09:22:56 -0700 Subject: [PATCH 2/7] Test the contract per operator instead of with a sweep Each behaviour now sits in the fixture for the operator it belongs to, where someone changing that operator will actually see it. QueryWhenChangedFixture, ToCollectionFixture, GroupOnFixture, GroupImmutableFixture, SortFixture, BufferFixture, MergeChangeSetsFixture, MergeManyChangeSetsFixture, MergeManyFixture, MaxFixture, and the four combiner fixtures. The recursion test uses int rather than Person because Bogus.Person and the domain Person are both in scope in that fixture. --- .../AggregationTests/MaxFixture.cs | 33 ++- src/DynamicData.Tests/List/AndFixture.cs | 37 +++- src/DynamicData.Tests/List/BufferFixture.cs | 35 ++- .../List/CombinerCompletionFixture.cs | 57 ----- src/DynamicData.Tests/List/ExceptFixture.cs | 37 +++- .../List/GroupImmutableFixture.cs | 17 ++ src/DynamicData.Tests/List/GroupOnFixture.cs | 21 +- .../List/MergeChangeSetsFixture.cs | 19 ++ .../List/MergeManyChangeSetsFixture.cs | 24 ++- .../List/MergeManyFixture.cs | 21 +- .../List/MergeManyInnerErrorFixture.cs | 32 --- .../List/OperatorCompletionFixture.cs | 203 ------------------ src/DynamicData.Tests/List/OrFixture.cs | 35 +++ .../List/QueryWhenChangedFixture.cs | 34 ++- src/DynamicData.Tests/List/SortFixture.cs | 18 ++ .../List/ToCollectionFixture.cs | 33 +++ src/DynamicData.Tests/List/XOrFixture.cs | 23 +- 17 files changed, 378 insertions(+), 301 deletions(-) delete mode 100644 src/DynamicData.Tests/List/CombinerCompletionFixture.cs delete mode 100644 src/DynamicData.Tests/List/MergeManyInnerErrorFixture.cs delete mode 100644 src/DynamicData.Tests/List/OperatorCompletionFixture.cs diff --git a/src/DynamicData.Tests/AggregationTests/MaxFixture.cs b/src/DynamicData.Tests/AggregationTests/MaxFixture.cs index 89f274619..56301895f 100644 --- a/src/DynamicData.Tests/AggregationTests/MaxFixture.cs +++ b/src/DynamicData.Tests/AggregationTests/MaxFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using DynamicData.Aggregation; using DynamicData.Tests.Domain; @@ -6,6 +6,11 @@ using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Binding; namespace DynamicData.Tests.AggregationTests; @@ -70,4 +75,30 @@ public void RemoveItems() result.Should().Be(20, "Max value should be 20 after remove"); accumulator.Dispose(); } + + [Fact] + public void MaximumCompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("Maximum is built on QueryWhenChanged"); + } + + [Fact] + public void MaximumDeliversTheErrorWithoutThrowing() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/List/AndFixture.cs b/src/DynamicData.Tests/List/AndFixture.cs index 1d18596ad..ed1de11f1 100644 --- a/src/DynamicData.Tests/List/AndFixture.cs +++ b/src/DynamicData.Tests/List/AndFixture.cs @@ -1,10 +1,15 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; using FluentAssertions; using Xunit; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Binding; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; @@ -97,4 +102,34 @@ public void StartingWithNonEmptySourceProducesNoResult() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = ObservableListEx.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 = ObservableListEx.And(first, second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/List/BufferFixture.cs b/src/DynamicData.Tests/List/BufferFixture.cs index e02330f98..e089f2d6c 100644 --- a/src/DynamicData.Tests/List/BufferFixture.cs +++ b/src/DynamicData.Tests/List/BufferFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Reactive.Linq; using DynamicData.Tests.Domain; @@ -8,6 +8,10 @@ using Microsoft.Reactive.Testing; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using System.Reactive.Subjects; +using DynamicData.Binding; namespace DynamicData.Tests.List; @@ -48,4 +52,33 @@ public void ResultsWillBeReceivedAfterClosingBuffer() _scheduler.AdvanceBy(TimeSpan.FromSeconds(61).Ticks); _results.Messages.Count.Should().Be(1, "Should be 1 update"); } + + [Fact] + public void BufferIfCompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.BufferIf(Observable.Return(false), Scheduler.Immediate).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void BufferIfFlushesHeldChangesBeforeCompleting() + { + var received = 0; + var completed = false; + + using var source = new Subject>(); + using var subscription = source.BufferIf(Observable.Return(true), Scheduler.Immediate).Subscribe(_ => received++, () => completed = true); + + source.OnNext(new ChangeSet { new Change(ListChangeReason.Add, new Person("a", 1), 0) }); + source.OnCompleted(); + + received.Should().Be(1, "changes held back by the pause would otherwise be lost"); + completed.Should().BeTrue(); + } } diff --git a/src/DynamicData.Tests/List/CombinerCompletionFixture.cs b/src/DynamicData.Tests/List/CombinerCompletionFixture.cs deleted file mode 100644 index 2f2ba3a69..000000000 --- a/src/DynamicData.Tests/List/CombinerCompletionFixture.cs +++ /dev/null @@ -1,57 +0,0 @@ -using System; -using System.Reactive.Linq; -using System.Reactive.Subjects; - -using DynamicData.Tests.Domain; - -using FluentAssertions; - -using Xunit; - -namespace DynamicData.Tests.List; - -/// -/// Terminal event behaviour for the combining operators. -/// -public class CombinerCompletionFixture -{ - [Fact] - public void CombinersCompleteWhenEverySourceCompletes() - { - foreach (var combine in new Func>, IObservable>, IObservable>>[] - { - static (a, b) => ObservableListEx.And(a, b), - static (a, b) => a.Or(b), - static (a, b) => a.Except(b), - static (a, b) => a.Xor(b), - }) - { - using var first = new Subject>(); - using var second = new Subject>(); - var completed = false; - - using var subscription = combine(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 CombinersDeliverErrorFromAnySource() - { - using var first = new Subject>(); - using var second = new Subject>(); - Exception? error = null; - - using var subscription = first.Or(second).Subscribe(_ => { }, ex => error = ex, () => { }); - - second.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType(); - } - -} diff --git a/src/DynamicData.Tests/List/ExceptFixture.cs b/src/DynamicData.Tests/List/ExceptFixture.cs index d74eff07b..f62eb507f 100644 --- a/src/DynamicData.Tests/List/ExceptFixture.cs +++ b/src/DynamicData.Tests/List/ExceptFixture.cs @@ -1,10 +1,15 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; using FluentAssertions; using Xunit; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Binding; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; @@ -106,4 +111,34 @@ public void NothingFromOther() } 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/List/GroupImmutableFixture.cs b/src/DynamicData.Tests/List/GroupImmutableFixture.cs index cbf674654..f640d7252 100644 --- a/src/DynamicData.Tests/List/GroupImmutableFixture.cs +++ b/src/DynamicData.Tests/List/GroupImmutableFixture.cs @@ -9,6 +9,10 @@ using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using DynamicData.Binding; namespace DynamicData.Tests.List; @@ -181,4 +185,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/List/GroupOnFixture.cs b/src/DynamicData.Tests/List/GroupOnFixture.cs index 61164d3c3..63828ab54 100644 --- a/src/DynamicData.Tests/List/GroupOnFixture.cs +++ b/src/DynamicData.Tests/List/GroupOnFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Linq; using DynamicData.Tests.Domain; @@ -6,6 +6,12 @@ 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.Binding; namespace DynamicData.Tests.List; @@ -74,4 +80,17 @@ public void UpdateWillChangeTheGroup() var firstGroup = _results.Data.Items[0].List.Items.ToArray(); firstGroup[0].Should().Be(amended, "Should be same person"); } + + [Fact] + public void CompletesWhenNoRegrouperIsSupplied() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.GroupOn(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/List/MergeChangeSetsFixture.cs b/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs index 253a70070..636ecf6b8 100644 --- a/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs +++ b/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs @@ -13,6 +13,8 @@ using Microsoft.Reactive.Testing; using Xunit; using System.Collections.Concurrent; +using System.Reactive.Subjects; +using DynamicData.Binding; namespace DynamicData.Tests.List; @@ -439,4 +441,21 @@ private Animal UseInsert(AnimalOwner owner) private IEnumerable>> GetEnumerableObservable() => _animalOwners.Select(owner => owner.Animals.Connect()); private IObservable>> GetObservableObservable() => GetEnumerableObservable().ToObservable(); + + [Fact] + public void WithAComparerDoesNotRecurse() + { + var completed = false; + + using var sources = new SourceList>>(); + + // This overload used to bind to itself and exhaust the stack before returning. + using var subscription = sources.Connect() + .MergeChangeSets(Comparer.Default) + .Subscribe(_ => { }, () => completed = true); + + sources.Dispose(); + + completed.Should().BeTrue(); + } } diff --git a/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs b/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs index ea8019ea4..118ae0111 100644 --- a/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs +++ b/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs @@ -1,7 +1,14 @@ -using System.Linq; +using System.Linq; using FluentAssertions; using Xunit; +using System; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Binding; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; @@ -52,4 +59,19 @@ public void MergeManyShouldWork() new[] { 2, 100, 10,11,12,13,14 }.Should().BeEquivalentTo(d.Items); } + + [Fact] + public void DeliversTheErrorWithoutThrowing() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source + .MergeManyChangeSets(_ => Observable.Empty>(), EqualityComparer.Default) + .Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/List/MergeManyFixture.cs b/src/DynamicData.Tests/List/MergeManyFixture.cs index 57cfc3d6f..c9a991041 100644 --- a/src/DynamicData.Tests/List/MergeManyFixture.cs +++ b/src/DynamicData.Tests/List/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.Concurrency; +using DynamicData.Binding; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; @@ -174,4 +178,19 @@ public void InvokeObservable(bool value) _changed.OnNext(value); } } + + [Fact] + public void DeliversAnErrorRaisedByAChild() + { + Exception? error = null; + + using var source = new SourceList(); + using var child = new Subject(); + using var subscription = source.Connect().MergeMany(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.Add(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/List/MergeManyInnerErrorFixture.cs b/src/DynamicData.Tests/List/MergeManyInnerErrorFixture.cs deleted file mode 100644 index 003525b6e..000000000 --- a/src/DynamicData.Tests/List/MergeManyInnerErrorFixture.cs +++ /dev/null @@ -1,32 +0,0 @@ -using System; -using System.Reactive.Subjects; - -using DynamicData.Tests.Domain; - -using FluentAssertions; - -using Xunit; - -namespace DynamicData.Tests.List; - -/// -/// Merge propagates a failure from any inner stream, rather than discarding it. -/// -public class MergeManyInnerErrorFixture -{ - [Fact] - public void MergeManyDeliversErrorFromAChild() - { - using var source = new SourceList(); - using var child = new Subject(); - Exception? error = null; - - using var subscription = source.Connect().MergeMany(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); - - source.Add(new Person("a", 1)); - child.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType("a failing inner stream must not be silently discarded"); - } - -} diff --git a/src/DynamicData.Tests/List/OperatorCompletionFixture.cs b/src/DynamicData.Tests/List/OperatorCompletionFixture.cs deleted file mode 100644 index 2ea58a445..000000000 --- a/src/DynamicData.Tests/List/OperatorCompletionFixture.cs +++ /dev/null @@ -1,203 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Reactive; -using System.Reactive.Concurrency; -using System.Reactive.Linq; -using System.Reactive.Subjects; - -using DynamicData.Aggregation; -using DynamicData.Binding; -using DynamicData.Tests.Domain; - -using FluentAssertions; - -using Xunit; - -namespace DynamicData.Tests.List; - -/// -/// Terminal event behaviour for list operators which previously never delivered OnCompleted. -/// -public class OperatorCompletionFixture -{ - private static readonly IComparer ByName = SortExpressionComparer.Ascending(p => p.Name); - - private static ChangeSet OneAdd() => [new Change(ListChangeReason.Add, new Person("a", 1), 0)]; - - [Fact] - public void QueryWhenChangedCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.QueryWhenChanged().Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void QueryWhenChangedDeliversError() - { - using var source = new Subject>(); - Exception? error = null; - - using var subscription = source.QueryWhenChanged().Subscribe(_ => { }, ex => error = ex, () => { }); - - source.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType(); - } - - [Fact] - public void ToCollectionCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.ToCollection().Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void ToSortedCollectionCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.ToSortedCollection(ByName).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void GroupOnCompletesWhenNoRegrouperIsSupplied() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.GroupOn(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 GroupWithImmutableStateCompletesWhenNoRegrouperIsSupplied() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.GroupWithImmutableState(p => p.Age).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void SortCompletesWhenGivenAComparerObservable() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.Sort(Observable.Return(ByName)).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void BufferIfCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.BufferIf(Observable.Return(false), Scheduler.Immediate).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void BufferIfFlushesHeldChangesBeforeCompleting() - { - using var source = new Subject>(); - var received = 0; - var completed = false; - - using var subscription = source.BufferIf(Observable.Return(true), Scheduler.Immediate).Subscribe(_ => received++, () => completed = true); - - source.OnNext(OneAdd()); - source.OnCompleted(); - - received.Should().Be(1, "changes held back by the pause would otherwise be lost"); - completed.Should().BeTrue(); - } - - [Fact] - public void MaximumCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void MaximumDeliversErrorWithoutThrowing() - { - using var source = new Subject>(); - Exception? error = null; - - using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { }); - - source.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType(); - } - - [Fact] - public void MergeManyChangeSetsDeliversErrorWithoutThrowing() - { - using var source = new Subject>(); - Exception? error = null; - - using var subscription = source - .MergeManyChangeSets(_ => Observable.Empty>(), EqualityComparer.Default) - .Subscribe(_ => { }, ex => error = ex, () => { }); - - source.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType(); - } - - [Fact] - public void MergeChangeSetsWithComparerDoesNotRecurse() - { - using var sources = new SourceList>>(); - var completed = false; - - // This used to bind to itself and exhaust the stack before returning. - using var subscription = sources.Connect() - .MergeChangeSets(SortExpressionComparer.Ascending(p => p.Name)) - .Subscribe(_ => { }, () => completed = true); - - sources.Dispose(); - - completed.Should().BeTrue(); - } -} diff --git a/src/DynamicData.Tests/List/OrFixture.cs b/src/DynamicData.Tests/List/OrFixture.cs index c4c973aa2..ec382921a 100644 --- a/src/DynamicData.Tests/List/OrFixture.cs +++ b/src/DynamicData.Tests/List/OrFixture.cs @@ -5,6 +5,11 @@ using FluentAssertions; using Xunit; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Binding; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; @@ -136,4 +141,34 @@ public void RemovedWhenNoLongerInEither() } 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/List/QueryWhenChangedFixture.cs b/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs index 49c5cd27e..ce93bea5e 100644 --- a/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs +++ b/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs @@ -1,10 +1,16 @@ -using System; +using System; using DynamicData.Tests.Domain; 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.Binding; namespace DynamicData.Tests.List; @@ -89,4 +95,30 @@ public void Dispose() _source.Dispose(); _results.Dispose(); } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.QueryWhenChanged().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void DeliversTheError() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source.QueryWhenChanged().Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/List/SortFixture.cs b/src/DynamicData.Tests/List/SortFixture.cs index 7f2a7dea3..4fc551216 100644 --- a/src/DynamicData.Tests/List/SortFixture.cs +++ b/src/DynamicData.Tests/List/SortFixture.cs @@ -8,6 +8,9 @@ using FluentAssertions; using Xunit; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Linq; namespace DynamicData.Tests.List; @@ -182,4 +185,19 @@ public void SortInitialBatch() actualResult.Should().BeEquivalentTo(expectedResult); } + + [Fact] + public void CompletesWhenGivenAComparerObservable() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source + .Sort(Observable.Return(SortExpressionComparer.Ascending(p => p.Name))) + .Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent resort signal can never fire and so must not hold the result open"); + } } diff --git a/src/DynamicData.Tests/List/ToCollectionFixture.cs b/src/DynamicData.Tests/List/ToCollectionFixture.cs index a46aa790f..1ab6e277d 100644 --- a/src/DynamicData.Tests/List/ToCollectionFixture.cs +++ b/src/DynamicData.Tests/List/ToCollectionFixture.cs @@ -3,6 +3,11 @@ using System.Reactive.Linq; using FluentAssertions; using Xunit; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Subjects; +using DynamicData.Binding; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; @@ -23,4 +28,32 @@ public void ToCollectionTest() res1?.Count.Should().Be(2); res2?.Count.Should().Be(2); } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.ToCollection().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("ToCollection is built on QueryWhenChanged"); + } + + [Fact] + public void ToSortedCollectionCompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source + .ToSortedCollection(SortExpressionComparer.Ascending(p => p.Name)) + .Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } } diff --git a/src/DynamicData.Tests/List/XOrFixture.cs b/src/DynamicData.Tests/List/XOrFixture.cs index 8f66b0d20..84ad95a8d 100644 --- a/src/DynamicData.Tests/List/XOrFixture.cs +++ b/src/DynamicData.Tests/List/XOrFixture.cs @@ -1,10 +1,15 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; using FluentAssertions; using Xunit; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Binding; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; @@ -108,4 +113,20 @@ public void RemovedWhenNoLongerInEither() } 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"); + } } From c67185e9bd41fd68411244c6bd690968f9e52adf Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 15:29:02 -0700 Subject: [PATCH 3/7] Deliver BufferIf pause-selector failures through the result observable The pause and resume subscriptions had no error handler, so a failure of the pause selector escaped on whichever thread raised it instead of reaching the subscriber. Collapsed into a single subscription, which gives the failure one unambiguous path out rather than two competing ones. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData.Tests/List/BufferFixture.cs | 17 +++++++++++ src/DynamicData/List/Internal/BufferIf.cs | 32 ++++++++++++--------- 2 files changed, 35 insertions(+), 14 deletions(-) diff --git a/src/DynamicData.Tests/List/BufferFixture.cs b/src/DynamicData.Tests/List/BufferFixture.cs index e089f2d6c..07aca16bd 100644 --- a/src/DynamicData.Tests/List/BufferFixture.cs +++ b/src/DynamicData.Tests/List/BufferFixture.cs @@ -81,4 +81,21 @@ public void BufferIfFlushesHeldChangesBeforeCompleting() received.Should().Be(1, "changes held back by the pause would otherwise be lost"); completed.Should().BeTrue(); } + + [Fact] + public void BufferIfFailsWhenThePauseSelectorFails() + { + 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.BufferIf(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"); + } } diff --git a/src/DynamicData/List/Internal/BufferIf.cs b/src/DynamicData/List/Internal/BufferIf.cs index 4e42b3b6a..cb51c99fa 100644 --- a/src/DynamicData/List/Internal/BufferIf.cs +++ b/src/DynamicData/List/Internal/BufferIf.cs @@ -31,21 +31,25 @@ public IObservable> Run() => Observable.Create>( var bufferSelector = Observable.Return(initialPauseState).Concat(_pauseIfTrueSelector.Merge(timeoutSubject)).ObserveOn(_scheduler).Synchronize(locker).Publish(); - var pause = bufferSelector.Where(state => state).Subscribe( - _ => + // Handled as a single subscription rather than two filtered ones, so that a failure + // of the pause selector has one unambiguous path to the observer instead of either + // arriving twice or, as before, escaping unhandled. + var pauseOrResume = bufferSelector.Subscribe( + state => { - paused = true; - - // add pause timeout if required - if (_timeOut != TimeSpan.Zero) + if (state) { - timeoutSubscriber.Disposable = Observable.Timer(_timeOut, _scheduler).Select(_ => false).SubscribeSafe(timeoutSubject); + paused = true; + + // add pause timeout if required + if (_timeOut != TimeSpan.Zero) + { + timeoutSubscriber.Disposable = Observable.Timer(_timeOut, _scheduler).Select(static _ => false).SubscribeSafe(timeoutSubject); + } + + return; } - }); - var resume = bufferSelector.Where(state => !state).Subscribe( - _ => - { paused = false; // publish changes and clear buffer @@ -59,7 +63,8 @@ public IObservable> Run() => Observable.Create>( // kill off timeout if required timeoutSubscriber.Disposable = Disposable.Empty; - }); + }, + observer.OnError); var updateSubscriber = _source.Synchronize(locker).Subscribe( updates => @@ -92,8 +97,7 @@ public IObservable> Run() => Observable.Create>( () => { connected.Dispose(); - pause.Dispose(); - resume.Dispose(); + pauseOrResume.Dispose(); updateSubscriber.Dispose(); timeoutSubject.OnCompleted(); timeoutSubscriber.Dispose(); From 7562b70bdc1b22e59dd702531f778d4ad97541f0 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 16:17:18 -0700 Subject: [PATCH 4/7] Update list operator docs for the changed terminal behaviour MergeMany no longer swallows errors from per-item observables, and had no OnError entry at all. BufferIf now flushes buffered changesets before completing rather than forwarding completion straight through, and forwards errors from the pause selector. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData/List/ObservableListEx.BufferIf.cs | 4 ++-- src/DynamicData/List/ObservableListEx.MergeMany.cs | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/DynamicData/List/ObservableListEx.BufferIf.cs b/src/DynamicData/List/ObservableListEx.BufferIf.cs index 37224290f..0df446229 100644 --- a/src/DynamicData/List/ObservableListEx.BufferIf.cs +++ b/src/DynamicData/List/ObservableListEx.BufferIf.cs @@ -76,8 +76,8 @@ public static IObservable> BufferIf(this IObservableAny (while active)Passed through immediately. /// Pause selector emits falseAll buffered changesets are flushed downstream as one combined changeset. /// Timeout firesAutomatically resumes and flushes the buffer. - /// OnErrorForwarded immediately (not buffered). - /// OnCompletedForwarded immediately. + /// OnErrorForwarded immediately (not buffered), and buffered data is lost. An error from terminates the output the same way a source error does. + /// OnCompletedAny remaining buffered changesets are flushed before completion is forwarded. /// /// Worth noting: Each pause/resume cycle re-arms the timeout. Rapid toggling can create many small buffer windows. /// diff --git a/src/DynamicData/List/ObservableListEx.MergeMany.cs b/src/DynamicData/List/ObservableListEx.MergeMany.cs index 55dfa4f19..a09b56f77 100644 --- a/src/DynamicData/List/ObservableListEx.MergeMany.cs +++ b/src/DynamicData/List/ObservableListEx.MergeMany.cs @@ -40,6 +40,7 @@ public static partial class ObservableListEx /// ReplaceOld subscription disposed, new subscription created for the replacement item. /// Remove/RemoveRange/ClearSubscription disposed. /// Refresh/MovedNo effect on subscriptions. + /// OnErrorAn error from a per-item observable, or from the source, terminates the merged output. /// OnCompleted (source)Completes only after the source and all active inner observables have completed. /// /// From aa31e20a75ac5a0e0d2643e54242a288aec2da39 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 17:34:05 -0700 Subject: [PATCH 5/7] Add concurrent delivery coverage for the multi-source list operators The list side turns out to be serialized already: Combiner and the operators around it hold their gate across downstream delivery, so two sources cannot both be inside the observer at once. That is worth having a test for rather than an assumption, particularly since the cache equivalent released its lock before delivering and was not serialized at all. Or, Sort, GroupOn, BufferIf and MergeManyChangeSets are each driven from several threads at once and checked for overlapping delivery and change set integrity. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../List/MultiSourceSerializationFixture.cs | 246 ++++++++++++++++++ 1 file changed, 246 insertions(+) create mode 100644 src/DynamicData.Tests/List/MultiSourceSerializationFixture.cs diff --git a/src/DynamicData.Tests/List/MultiSourceSerializationFixture.cs b/src/DynamicData.Tests/List/MultiSourceSerializationFixture.cs new file mode 100644 index 000000000..a5180a053 --- /dev/null +++ b/src/DynamicData.Tests/List/MultiSourceSerializationFixture.cs @@ -0,0 +1,246 @@ +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.List; + +/// +/// Serialization coverage for the list 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 +/// list change sets carry indices and a torn read usually surfaces as a structurally invalid change +/// set rather than as a visible race. +/// +public class MultiSourceSerializationFixture +{ + private const int Seed = 0x1147; + + private static readonly TimeSpan DrainTimeout = TimeSpan.FromMinutes(2); + + [Fact] + public async Task OrDeliversSeriallyWhileBothSourcesAreWritten() + { + using var left = new SourceList(); + using var right = new SourceList(); + + await RunAsync( + left.Connect().Or(right.Connect()), + writers: + [ + randomizer => left.Add(new Person("L" + randomizer.Int(1, 500), randomizer.Int(1, 80))), + randomizer => right.Add(new Person("R" + randomizer.Int(1, 500), randomizer.Int(1, 80))), + randomizer => + { + var items = left.Items.ToArray(); + if (items.Length > 0) + { + left.Remove(items[randomizer.Int(0, items.Length - 1)]); + } + }, + ], + complete: () => + { + left.Dispose(); + right.Dispose(); + }); + } + + [Fact] + public async Task SortDeliversSeriallyWhileTheComparerChanges() + { + using var source = new SourceList(); + 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, resetThreshold: 25, resort: resort), + writers: + [ + randomizer => source.Add(new Person("P" + randomizer.Int(1, 500), 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 SourceList(); + using var regroup = new Subject(); + + await RunAsync( + source.Connect().GroupOn(p => p.Age % 5, regroup).Transform(group => new Person(group.GroupKey.ToString(), group.GroupKey)), + writers: + [ + randomizer => source.Add(new Person("P" + randomizer.Int(1, 500), randomizer.Int(1, 80))), + randomizer => + { + var items = source.Items.ToArray(); + if (items.Length > 0) + { + source.Remove(items[randomizer.Int(0, items.Length - 1)]); + } + }, + _ => regroup.OnNext(Unit.Default), + ], + complete: () => + { + regroup.OnCompleted(); + source.Dispose(); + }); + } + + [Fact] + public async Task BufferIfDeliversSeriallyWhilePausingConcurrently() + { + using var source = new SourceList(); + using var pause = new Subject(); + + await RunAsync( + source.Connect().BufferIf(pause), + writers: + [ + randomizer => source.Add(new Person("P" + randomizer.Int(1, 500), randomizer.Int(1, 80))), + randomizer => + { + var items = source.Items.ToArray(); + if (items.Length > 0) + { + source.Remove(items[randomizer.Int(0, items.Length - 1)]); + } + }, + randomizer => pause.OnNext(randomizer.Bool()), + ], + complete: () => + { + // Leave the buffer open, so anything held back has to be flushed on completion. + pause.OnNext(false); + pause.OnCompleted(); + source.Dispose(); + }); + } + + [Fact] + public async Task MergeManyChangeSetsDeliversSeriallyWhileChildrenAreWritten() + { + using var owners = new SourceList(); + var created = new List(); + + for (var i = 0; i < 5; i++) + { + var owner = new AnimalOwner("Owner" + i); + created.Add(owner); + owners.Add(owner); + } + + var nextName = 0; + + await RunAsync( + owners.Connect().MergeManyChangeSets(o => o.Animals.Connect()).Transform(a => new Person(a.Name, a.Name.Length)), + 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. + 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() + + // 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"); + } +} From 9ebb760050819772930ee5814d3d1b378f77844c Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Tue, 28 Jul 2026 08:09:20 -0700 Subject: [PATCH 6/7] Tidy the using directives added with these tests The tests were added in bulk with a fixed set of imports, so several files picked up ones they do not use. Removed those, moved the rest into the group they belong in rather than after Xunit, and sorted each group. --- src/DynamicData.Tests/AggregationTests/MaxFixture.cs | 6 +----- src/DynamicData.Tests/List/AndFixture.cs | 7 ++----- src/DynamicData.Tests/List/BufferFixture.cs | 7 +++---- src/DynamicData.Tests/List/ExceptFixture.cs | 7 ++----- src/DynamicData.Tests/List/GroupImmutableFixture.cs | 5 +---- src/DynamicData.Tests/List/GroupOnFixture.cs | 8 ++------ src/DynamicData.Tests/List/MergeChangeSetsFixture.cs | 4 +--- .../List/MergeManyChangeSetsFixture.cs | 12 +++++------- src/DynamicData.Tests/List/MergeManyFixture.cs | 6 ++---- src/DynamicData.Tests/List/OrFixture.cs | 7 ++----- .../List/QueryWhenChangedFixture.cs | 8 ++------ src/DynamicData.Tests/List/SortFixture.cs | 4 +--- src/DynamicData.Tests/List/ToCollectionFixture.cs | 6 ++---- src/DynamicData.Tests/List/XOrFixture.cs | 7 ++----- 14 files changed, 28 insertions(+), 66 deletions(-) diff --git a/src/DynamicData.Tests/AggregationTests/MaxFixture.cs b/src/DynamicData.Tests/AggregationTests/MaxFixture.cs index 56301895f..23fe05d6a 100644 --- a/src/DynamicData.Tests/AggregationTests/MaxFixture.cs +++ b/src/DynamicData.Tests/AggregationTests/MaxFixture.cs @@ -1,4 +1,5 @@ using System; +using System.Reactive.Subjects; using DynamicData.Aggregation; using DynamicData.Tests.Domain; @@ -6,11 +7,6 @@ using FluentAssertions; using Xunit; -using System.Collections.Generic; -using System.Reactive.Concurrency; -using System.Reactive.Linq; -using System.Reactive.Subjects; -using DynamicData.Binding; namespace DynamicData.Tests.AggregationTests; diff --git a/src/DynamicData.Tests/List/AndFixture.cs b/src/DynamicData.Tests/List/AndFixture.cs index ed1de11f1..fef62fb8c 100644 --- a/src/DynamicData.Tests/List/AndFixture.cs +++ b/src/DynamicData.Tests/List/AndFixture.cs @@ -1,15 +1,12 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Subjects; +using DynamicData.Tests.Domain; using FluentAssertions; using Xunit; -using System.Reactive.Concurrency; -using System.Reactive.Linq; -using System.Reactive.Subjects; -using DynamicData.Binding; -using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/BufferFixture.cs b/src/DynamicData.Tests/List/BufferFixture.cs index 07aca16bd..ef3dfbad3 100644 --- a/src/DynamicData.Tests/List/BufferFixture.cs +++ b/src/DynamicData.Tests/List/BufferFixture.cs @@ -1,5 +1,8 @@ using System; +using System.Collections.Generic; +using System.Reactive.Concurrency; using System.Reactive.Linq; +using System.Reactive.Subjects; using DynamicData.Tests.Domain; @@ -8,10 +11,6 @@ using Microsoft.Reactive.Testing; using Xunit; -using System.Collections.Generic; -using System.Reactive.Concurrency; -using System.Reactive.Subjects; -using DynamicData.Binding; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/ExceptFixture.cs b/src/DynamicData.Tests/List/ExceptFixture.cs index f62eb507f..eab8cd97a 100644 --- a/src/DynamicData.Tests/List/ExceptFixture.cs +++ b/src/DynamicData.Tests/List/ExceptFixture.cs @@ -1,15 +1,12 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Subjects; +using DynamicData.Tests.Domain; using FluentAssertions; using Xunit; -using System.Reactive.Concurrency; -using System.Reactive.Linq; -using System.Reactive.Subjects; -using DynamicData.Binding; -using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/GroupImmutableFixture.cs b/src/DynamicData.Tests/List/GroupImmutableFixture.cs index f640d7252..75b541691 100644 --- a/src/DynamicData.Tests/List/GroupImmutableFixture.cs +++ b/src/DynamicData.Tests/List/GroupImmutableFixture.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Linq; using System.Reactive; using System.Reactive.Subjects; @@ -9,10 +10,6 @@ using FluentAssertions; using Xunit; -using System.Collections.Generic; -using System.Reactive.Concurrency; -using System.Reactive.Linq; -using DynamicData.Binding; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/GroupOnFixture.cs b/src/DynamicData.Tests/List/GroupOnFixture.cs index 63828ab54..693f883ae 100644 --- a/src/DynamicData.Tests/List/GroupOnFixture.cs +++ b/src/DynamicData.Tests/List/GroupOnFixture.cs @@ -1,17 +1,13 @@ using System; +using System.Collections.Generic; using System.Linq; +using System.Reactive.Subjects; using DynamicData.Tests.Domain; 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.Binding; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs b/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs index 636ecf6b8..ce11ca522 100644 --- a/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs +++ b/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Reactive.Concurrency; @@ -12,9 +13,6 @@ using FluentAssertions; using Microsoft.Reactive.Testing; using Xunit; -using System.Collections.Concurrent; -using System.Reactive.Subjects; -using DynamicData.Binding; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs b/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs index 118ae0111..f8fbed867 100644 --- a/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs +++ b/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs @@ -1,14 +1,12 @@ -using System.Linq; -using FluentAssertions; - -using Xunit; -using System; +using System; using System.Collections.Generic; -using System.Reactive.Concurrency; +using System.Linq; using System.Reactive.Linq; using System.Reactive.Subjects; -using DynamicData.Binding; using DynamicData.Tests.Domain; +using FluentAssertions; + +using Xunit; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/MergeManyFixture.cs b/src/DynamicData.Tests/List/MergeManyFixture.cs index c9a991041..438f2e845 100644 --- a/src/DynamicData.Tests/List/MergeManyFixture.cs +++ b/src/DynamicData.Tests/List/MergeManyFixture.cs @@ -1,14 +1,12 @@ using System; +using System.Collections.Generic; using System.Reactive.Linq; using System.Reactive.Subjects; +using DynamicData.Tests.Domain; using FluentAssertions; using Xunit; -using System.Collections.Generic; -using System.Reactive.Concurrency; -using DynamicData.Binding; -using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/OrFixture.cs b/src/DynamicData.Tests/List/OrFixture.cs index ec382921a..91c5fafe7 100644 --- a/src/DynamicData.Tests/List/OrFixture.cs +++ b/src/DynamicData.Tests/List/OrFixture.cs @@ -1,15 +1,12 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Subjects; +using DynamicData.Tests.Domain; using FluentAssertions; using Xunit; -using System.Reactive.Concurrency; -using System.Reactive.Linq; -using System.Reactive.Subjects; -using DynamicData.Binding; -using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs b/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs index ce93bea5e..569451f64 100644 --- a/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs +++ b/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs @@ -1,16 +1,12 @@ using System; +using System.Collections.Generic; +using System.Reactive.Subjects; using DynamicData.Tests.Domain; 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.Binding; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/SortFixture.cs b/src/DynamicData.Tests/List/SortFixture.cs index 4fc551216..0fe9a3e17 100644 --- a/src/DynamicData.Tests/List/SortFixture.cs +++ b/src/DynamicData.Tests/List/SortFixture.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Linq; using System.Reactive.Subjects; using DynamicData.Binding; using DynamicData.Tests.Domain; @@ -8,9 +9,6 @@ using FluentAssertions; using Xunit; -using System.Reactive; -using System.Reactive.Concurrency; -using System.Reactive.Linq; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/ToCollectionFixture.cs b/src/DynamicData.Tests/List/ToCollectionFixture.cs index 1ab6e277d..05a2dd6a5 100644 --- a/src/DynamicData.Tests/List/ToCollectionFixture.cs +++ b/src/DynamicData.Tests/List/ToCollectionFixture.cs @@ -1,13 +1,11 @@ using System; using System.Collections.Generic; using System.Reactive.Linq; -using FluentAssertions; -using Xunit; -using System.Reactive; -using System.Reactive.Concurrency; using System.Reactive.Subjects; using DynamicData.Binding; using DynamicData.Tests.Domain; +using FluentAssertions; +using Xunit; namespace DynamicData.Tests.List; diff --git a/src/DynamicData.Tests/List/XOrFixture.cs b/src/DynamicData.Tests/List/XOrFixture.cs index 84ad95a8d..f52b6a6c4 100644 --- a/src/DynamicData.Tests/List/XOrFixture.cs +++ b/src/DynamicData.Tests/List/XOrFixture.cs @@ -1,15 +1,12 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Subjects; +using DynamicData.Tests.Domain; using FluentAssertions; using Xunit; -using System.Reactive.Concurrency; -using System.Reactive.Linq; -using System.Reactive.Subjects; -using DynamicData.Binding; -using DynamicData.Tests.Domain; namespace DynamicData.Tests.List; From 8c8a28bf3aa16be10e27c35396833cc9363f191e Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Tue, 28 Jul 2026 11:01:27 -0700 Subject: [PATCH 7/7] Drop the repeated comments on no-op error handlers Each of these sat above a Subscribe passing a discarding onError, saying the same thing about Rx rethrowing without a handler. The argument already says it. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData/List/Internal/DynamicCombiner.cs | 4 ---- src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs | 2 -- src/DynamicData/List/Internal/MergeManyListChangeSets.cs | 2 -- 3 files changed, 8 deletions(-) diff --git a/src/DynamicData/List/Internal/DynamicCombiner.cs b/src/DynamicData/List/Internal/DynamicCombiner.cs index 1f196d3e4..73316d4e6 100644 --- a/src/DynamicData/List/Internal/DynamicCombiner.cs +++ b/src/DynamicData/List/Internal/DynamicCombiner.cs @@ -66,8 +66,6 @@ public IObservable> Run() => Observable.Create>( } } }) - // The merge subscription above reports errors to the observer. Without a handler here Rx - // would rethrow them out of the subscription as well. .Subscribe(static _ => { }, static _ => { }); // When a list is added, update all items that are in that list @@ -90,8 +88,6 @@ public IObservable> Run() => Observable.Create>( } } }) - // The merge subscription above reports errors to the observer. Without a handler here Rx - // would rethrow them out of the subscription as well. .Subscribe(static _ => { }, static _ => { }); return new CompositeDisposable(sourceLists, allChanges, removedItem, sourceChanged); diff --git a/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs b/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs index 1ab5ae85f..238614ff8 100644 --- a/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs +++ b/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs @@ -51,8 +51,6 @@ public IObservable> Run() => Observabl changeTracker.EmitChanges(observer); parentUpdate = false; }) - // The subscription above already reports errors to the observer. Without a handler here Rx - // would rethrow them out of the subscription as well. .Subscribe(static _ => { }, static _ => { }); return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove); diff --git a/src/DynamicData/List/Internal/MergeManyListChangeSets.cs b/src/DynamicData/List/Internal/MergeManyListChangeSets.cs index 1767720dd..5c52f6729 100644 --- a/src/DynamicData/List/Internal/MergeManyListChangeSets.cs +++ b/src/DynamicData/List/Internal/MergeManyListChangeSets.cs @@ -47,8 +47,6 @@ public IObservable> Run() => Observable.Create { }, static _ => { }); return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove);