Make the list operators honour the Rx observable contract

src/DynamicData.Tests/AggregationTests/MaxFixture.cs

@@ -1,4 +1,5 @@
-using System;
+using System;
+using System.Reactive.Subjects;
 
 using DynamicData.Aggregation;
 using DynamicData.Tests.Domain;
@@ -70,4 +71,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<IChangeSet<Person>>();
+        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<IChangeSet<Person>>();
+        using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        source.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }

src/DynamicData.Tests/List/AndFixture.cs

@@ -1,7 +1,9 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Subjects;
 
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -97,4 +99,34 @@ public void StartingWithNonEmptySourceProducesNoResult()
     }
 
     protected abstract IObservable<IChangeSet<int>> CreateObservable();
+
+    [Fact]
+    public void CompletesOnlyWhenEverySourceCompletes()
+    {
+        var completed = false;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        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<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = ObservableListEx.And(first, second).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        second.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }

src/DynamicData.Tests/List/BufferFixture.cs

@@ -1,5 +1,8 @@
-using System;
+using System;
+using System.Collections.Generic;
+using System.Reactive.Concurrency;
 using System.Reactive.Linq;
+using System.Reactive.Subjects;
 
 using DynamicData.Tests.Domain;
 
@@ -48,4 +51,50 @@ 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<IChangeSet<Person>>();
+        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<IChangeSet<Person>>();
+        using var subscription = source.BufferIf(Observable.Return(true), Scheduler.Immediate).Subscribe(_ => received++, () => completed = true);
+
+        source.OnNext(new ChangeSet<Person> { new Change<Person>(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();
+    }
+
+    [Fact]
+    public void BufferIfFailsWhenThePauseSelectorFails()
+    {
+        var expectedError = new Exception("Test Exception");
+        var actualError = default(Exception);
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var pause = new Subject<bool>();
+        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");
+    }
 }

src/DynamicData.Tests/List/ExceptFixture.cs

@@ -1,7 +1,9 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Subjects;
 
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -106,4 +108,34 @@ public void NothingFromOther()
     }
 
     protected abstract IObservable<IChangeSet<int>> CreateObservable();
+
+    [Fact]
+    public void CompletesOnlyWhenEverySourceCompletes()
+    {
+        var completed = false;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        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<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = first.Except(second).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        second.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }

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;
@@ -181,4 +182,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<IChangeSet<Person>>();
+        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");
+    }
 }

src/DynamicData.Tests/List/GroupOnFixture.cs

@@ -1,5 +1,7 @@
-using System;
+using System;
+using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Subjects;
 
 using DynamicData.Tests.Domain;
 
@@ -74,4 +76,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<IChangeSet<Person>>();
+        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");
+    }
 }

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,7 +13,6 @@
 using FluentAssertions;
 using Microsoft.Reactive.Testing;
 using Xunit;
-using System.Collections.Concurrent;
 
 namespace DynamicData.Tests.List;
 
@@ -439,4 +439,21 @@ private Animal UseInsert(AnimalOwner owner)
 
     private IEnumerable<IObservable<IChangeSet<Animal>>> GetEnumerableObservable() => _animalOwners.Select(owner => owner.Animals.Connect());
     private IObservable<IObservable<IChangeSet<Animal>>> GetObservableObservable() => GetEnumerableObservable().ToObservable();
+
+    [Fact]
+    public void WithAComparerDoesNotRecurse()
+    {
+        var completed = false;
+
+        using var sources = new SourceList<IObservable<IChangeSet<int, int>>>();
+
+        // This overload used to bind to itself and exhaust the stack before returning.
+        using var subscription = sources.Connect()
+            .MergeChangeSets(Comparer<int>.Default)
+            .Subscribe(_ => { }, () => completed = true);
+
+        sources.Dispose();
+
+        completed.Should().BeTrue();
+    }
 }

src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs

@@ -1,4 +1,9 @@
+using System;
+using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -52,4 +57,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<IChangeSet<Person>>();
+        using var subscription = source
+            .MergeManyChangeSets(_ => Observable.Empty<IChangeSet<Person>>(), EqualityComparer<Person>.Default)
+            .Subscribe(_ => { }, ex => error = ex, () => { });
+
+        source.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }

src/DynamicData.Tests/List/MergeManyFixture.cs

@@ -1,7 +1,9 @@
-using System;
+using System;
+using System.Collections.Generic;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -114,7 +116,7 @@ public void MergedStreamCompletesWhenSourceAndItemsComplete()
     /// Stream completes even if one of the children fails.
     /// </summary>
     [Fact]
-    public void MergedStreamCompletesIfLastItemFails()
+    public void MergedStreamFailsIfLastItemFails()
     {
         var receivedError = default(Exception);
         var streamCompleted = false;
@@ -129,9 +131,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");
     }
 
     /// <summary>
@@ -174,4 +176,19 @@ public void InvokeObservable(bool value)
             _changed.OnNext(value);
         }
     }
+
+    [Fact]
+    public void DeliversAnErrorRaisedByAChild()
+    {
+        Exception? error = null;
+
+        using var source = new SourceList<Person>();
+        using var child = new Subject<int>();
+        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<InvalidOperationException>("Merge propagates a failure from any inner stream");
+    }
 }