Make the cache operators honour the Rx observable contract

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<IChangeSet<Person, string>> CreateObservable();
+
+    [Fact]
+    public void CompletesOnlyWhenEverySourceCompletes()
+    {
+        var completed = false;
+
+        using var first = new Subject<IChangeSet<Person, string>>();
+        using var second = new Subject<IChangeSet<Person, string>>();
+        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<IChangeSet<Person, string>>();
+        using var second = new Subject<IChangeSet<Person, string>>();
+        using var subscription = ObservableCacheEx.And(first, second).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        second.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }

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<IChangeSet<Person, string>>();
+        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<IChangeSet<Person, string>>();
+        using var pause = new BehaviorSubject<bool>(true);
+        using var subscription = source.BatchIf(pause, Scheduler.Immediate).Subscribe(_ => received++, () => completed = true);
+
+        source.OnNext(new ChangeSet<Person, string> { 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<IChangeSet<Person, string>>();
+        using var pause = new Subject<bool>();
+        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()
     {

src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs

@@ -1,11 +1,15 @@
-using System;
+using System;
 using System.Linq;
 
 using DynamicData.Tests.Domain;
 
 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<IChangeSet<Person, string>>();
+        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<IChangeSet<Person, string>>();
+        using var subscription = source.SkipInitial().Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue("SkipInitial is built on DeferUntilLoaded");
+    }
 }

src/DynamicData.Tests/Cache/ExceptFixture.cs

@@ -1,11 +1,13 @@
-using System;
+using System;
 using System.Collections.Generic;
 
 using DynamicData.Tests.Domain;
 
 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<IChangeSet<Person, string>> CreateObservable();
+
+    [Fact]
+    public void CompletesOnlyWhenEverySourceCompletes()
+    {
+        var completed = false;
+
+        using var first = new Subject<IChangeSet<Person, string>>();
+        using var second = new Subject<IChangeSet<Person, string>>();
+        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, string>>();
+        using var second = new Subject<IChangeSet<Person, string>>();
+        using var subscription = first.Except(second).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        second.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }

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<GroupEntryViewModel> Entries => _entries;
     }
+
+    [Fact]
+    public void CompletesWhenNoRegrouperIsSupplied()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person, string>>();
+        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<IChangeSet<Person, string>>();
+        using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        source.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
+
+    [Fact]
+    public void DeliversTheErrorWhenAResultGroupSourceIsSupplied()
+    {
+        Exception? error = null;
+
+        using var source = new Subject<IChangeSet<Person, string>>();
+        using var subscription = source
+            .Group(p => p.Age, Observable.Never<IDistinctChangeSet<int>>())
+            .Subscribe(_ => { }, ex => error = ex, () => { });
+
+        source.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
+
+    [Fact]
+    public void DeliversTerminalEventsFromTheResultGroupSource()
+    {
+        var completed = false;
+
+        using (var source = new Subject<IChangeSet<Person, string>>())
+        using (var groups = new Subject<IDistinctChangeSet<int>>())
+        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<IChangeSet<Person, string>>())
+        using (var groups = new Subject<IDistinctChangeSet<int>>())
+        using (source.Group(p => p.Age, groups).Subscribe(_ => { }, ex => error = ex, () => { }))
+        {
+            groups.OnError(new InvalidOperationException("boom"));
+        }
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }

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<IChangeSet<Person, string>>();
+        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/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<IChangeSet<Person, string>>();
+        using var subscription = left
+            .InnerJoinMany(Observable.Empty<IChangeSet<Person, string>>(), 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");
+    }
 }