diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs index ddc9d6c07..521fbebc3 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs @@ -1,7 +1,8 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; using System.Reactive.Linq; +using System.Reactive.Subjects; using System.Threading.Tasks; using FluentAssertions; @@ -453,6 +454,64 @@ public void ReSuspendThenResumeDeliversAllInSingleBatch() results.IsCompleted.Should().BeFalse(); } + [Fact] + public void OnErrorFiresIfCacheFailsWhileSuspended() + { + // A connection made while suspended is deferred, and must still be told when the source + // fails. Reporting the failure as a successful completion leaves the subscriber's error + // handling unrun and its data looking complete. + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + using var suspend = cache.SuspendNotifications(); + using var results = cache.Connect().AsAggregator(); + + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + results.Error.Should().Be(expectedError, "a connection deferred by a suspension should see the source fail"); + results.IsCompleted.Should().BeFalse("the source failed, it did not complete"); + } + + [Fact] + public void OnErrorFiresIfCacheFailsWhileWatchIsSuspended() + { + // Watch defers the same way Connect does, and has the same obligation. + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + using var suspend = cache.SuspendNotifications(); + Exception? actualError = null; + var isCompleted = false; + using var subscription = cache.Watch(1).Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true); + + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + actualError.Should().Be(expectedError, "a watch deferred by a suspension should see the source fail"); + isCompleted.Should().BeFalse("the source failed, it did not complete"); + } + + [Fact] + public void OnErrorFiresIfCacheFailsAfterResumingWhileConnectionWasSuspended() + { + // The deferred connection has activated by the time the failure arrives, so this covers + // the path through the connection itself rather than through the suspension gate. + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + var suspend = cache.SuspendNotifications(); + using var results = cache.Connect().AsAggregator(); + source.OnNext(new ChangeSet { new(ChangeReason.Add, 1, 1) }); + + suspend.Dispose(); + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + results.Error.Should().Be(expectedError, "an activated connection should still see the source fail"); + results.Data.Count.Should().Be(1, "the data written before the failure should have arrived"); + } + public void Dispose() { _source.Dispose(); diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index afce06cc6..60642ee29 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -120,9 +120,14 @@ public IObservable> Connect(Func? predi // Create the Connection Observable ? CreateConnectObservable(predicate, suppressEmptyChangeSets) - // Defer until notifications are no longer suspended - : _suspensionTracker.Value.NotificationsSuspendedObservable.Do(static _ => { }, observer.OnCompleted) - .Where(static b => !b).Take(1).Select(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)).Switch(); + // Defer until notifications are no longer suspended. Take(1) means there is only + // ever one inner sequence, so SelectMany carries the terminal event of the gate + // through on its own: the connection ends when the cache does, and fails when it + // fails, rather than reporting a failure as a successful completion. + : _suspensionTracker.Value.NotificationsSuspendedObservable + .Where(static areNotificationsSuspended => !areNotificationsSuspended) + .Take(1) + .SelectMany(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)); return observable.SubscribeSafe(observer); } @@ -144,9 +149,12 @@ public IObservable> Watch(TKey key) => // Create the Watch Observable ? CreateWatchObservable(key) - // Defer until notifications are no longer suspended - : _suspensionTracker.Value.NotificationsSuspendedObservable.Do(static _ => { }, observer.OnCompleted) - .Where(static b => !b).Take(1).Select(_ => CreateWatchObservable(key)).Switch(); + // Defer until notifications are no longer suspended. See Connect() for why + // SelectMany is used here. + : _suspensionTracker.Value.NotificationsSuspendedObservable + .Where(static areNotificationsSuspended => !areNotificationsSuspended) + .Take(1) + .SelectMany(_ => CreateWatchObservable(key)); return observable.SubscribeSafe(observer); } @@ -394,7 +402,7 @@ public void OnError(Exception error) if (cache._suspensionTracker.IsValueCreated) { - cache._suspensionTracker.Value.Dispose(); + cache._suspensionTracker.Value.Fault(error); } } @@ -517,6 +525,12 @@ public void EmitResumeNotification() } } + public void Fault(Exception error) + { + _areNotificationsSuspended.OnError(error); + _areNotificationsSuspended.Dispose(); + } + public void Dispose() { _areNotificationsSuspended.OnCompleted();