From 9c9b70e5473234c81f5cd80d6fb150206db9358a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dar=C3=ADo=20Kondratiuk?= Date: Fri, 31 Jul 2026 10:36:11 -0300 Subject: [PATCH] Race timeout/cancellation/signals at the Task level, drop AssumeNeverEmits RaceWithSignalAndTimer used to force cancellation and timeout into fake Observable branches (via AssumeNeverEmits) just so they could sit in an Observable-level RaceWith next to the real data stream - a workaround for C# having no bottom type. Racing at the Task level instead sidesteps the problem entirely: Task.WhenAny doesn't care that the branches share a type. Also adds a Task-signal overload for racing against an already-existing "only ever faults" task (e.g. a session-closed signal) instead of a CancellationToken. Bump to 0.3.0 - RaceWithSignalAndTimer/RetryAndRaceWithSignalAndTimer now return Task instead of Observable, and AssumeNeverEmits is gone. --- README.md | 2 +- .../Extras/AssumeNeverEmits.cs | 31 ------ .../Extras/RaceWithSignalAndTimer.cs | 100 ++++++++++++++--- .../Extras/RetryAndRaceWithSignalAndTimer.cs | 21 ++-- .../ReactiveExtensionsSharp.csproj | 6 +- .../Extras/AssumeNeverEmitsExtrasTests.cs | 64 ----------- .../RaceWithSignalAndTimerTests.cs | 101 +++++++----------- .../RetryAndRaceWithSignalAndTimerTests.cs | 62 +++-------- .../Samples/RetryUntilTimeoutSample.cs | 2 +- 9 files changed, 154 insertions(+), 235 deletions(-) delete mode 100644 src/ReactiveExtensionsSharp/Extras/AssumeNeverEmits.cs delete mode 100644 test/ReactiveExtensionsSharp.Tests/Extras/AssumeNeverEmitsExtrasTests.cs diff --git a/README.md b/README.md index e33af8a..2a18f09 100644 --- a/README.md +++ b/README.md @@ -90,7 +90,7 @@ ReactiveExtensionsSharp ports that exact combinator as `RetryAndRaceWithSignalAn public static async Task FindElementOnceItRendersAsync(Func> tryFindElement, CancellationToken cancellationToken) => await Observable.Defer(() => Observable.From(tryFindElement())) .RetryAndRaceWithSignalAndTimer(TimeSpan.FromSeconds(5), cancellationToken) - .FirstValueFrom().ConfigureAwait(false); + .ConfigureAwait(false); ``` snippet source | anchor diff --git a/src/ReactiveExtensionsSharp/Extras/AssumeNeverEmits.cs b/src/ReactiveExtensionsSharp/Extras/AssumeNeverEmits.cs deleted file mode 100644 index e3e6add..0000000 --- a/src/ReactiveExtensionsSharp/Extras/AssumeNeverEmits.cs +++ /dev/null @@ -1,31 +0,0 @@ -using ReactiveExtensionsSharp.Operators; - -namespace ReactiveExtensionsSharp.Extras; - -/// Extension methods widening an error-only to another element type. -public static partial class RxExtensions -{ - /// - /// Widens an that only ever calls (such as - /// or ) so it - /// type-checks anywhere an is expected - most commonly as one of the branches - /// passed to alongside a source that actually produces - /// values. Mirrors how rxjs relies on TypeScript structurally accepting - /// Observable<never> anywhere an Observable<T> is expected; C# has no bottom type, - /// so this exists to fill that gap explicitly. - /// - /// - /// This is only safe for a that is guaranteed, by contract, to never call - /// . If it ever does, the returned observable throws - /// from that point on, converting a real value into a crash rather - /// than propagating it - there is no compile-time check for this, unlike TypeScript's never. Do not - /// apply this to a source that might legitimately emit. - /// - /// The element type to widen to. - /// An observable that only ever errors, never emits. - /// An that mirrors 's errors and throws if it ever emits. - public static Observable AssumeNeverEmits(this Observable source) - => source.Map(_ => throw new InvalidOperationException( - "AssumeNeverEmits: the source observable emitted a value, but was assumed to only ever error. " + - "This is a contract violation in the caller, not in AssumeNeverEmits itself.")); -} diff --git a/src/ReactiveExtensionsSharp/Extras/RaceWithSignalAndTimer.cs b/src/ReactiveExtensionsSharp/Extras/RaceWithSignalAndTimer.cs index 19420a2..90e8cc2 100644 --- a/src/ReactiveExtensionsSharp/Extras/RaceWithSignalAndTimer.cs +++ b/src/ReactiveExtensionsSharp/Extras/RaceWithSignalAndTimer.cs @@ -1,37 +1,42 @@ -using ReactiveExtensionsSharp.Operators; - namespace ReactiveExtensionsSharp.Extras; -/// Extension methods racing a single subscription against cancellation and a timeout. +/// Extension methods racing a single subscription against cancellation/an external signal and a timeout. public static partial class RxExtensions { /// - /// Races against cancellation and a timeout, whichever fires first. The - /// non-retrying half of + /// Races 's first value against cancellation and a timeout, whichever fires first. + /// The non-retrying half of /// - use this directly for a single wait (e.g. "wait for the next matching event") that doesn't need - /// retrying, and reach for the retrying combinator when it does. + /// retrying, and reach for the retrying combinator when it does. Races at the level + /// rather than the level: cancellation and the timeout never produce a value + /// of type , only ever fault or never complete, and C# has no bottom type to make + /// an error-only Observable<Unit> type-check as Observable<T> the way TypeScript's + /// never lets rxjs do it - racing plain s sidesteps that entirely. /// /// The type of values produced by . /// The source sequence to race. /// The overall duration before giving up with a timeout error. A zero or negative value disables the timeout. /// /// Produces the exception used for both the cancellation and timeout branches. Defaults to - /// for cancellation and for the timeout, - /// via the defaults of and respectively. + /// for cancellation and for the timeout. /// Since one factory covers both branches, a caller needing to tell the two apart by exception type should /// pass here (so each branch keeps its own distinct default type) and catch/rethrow /// as needed at the call site. /// /// A token that, when cancelled, aborts the wait immediately. - /// An observable that mirrors unless the timeout or cancellation fires first. - public static Observable RaceWithSignalAndTimer( + /// A task that resolves to 's first value unless the timeout or cancellation fires first. + public static async Task RaceWithSignalAndTimer( this Observable source, TimeSpan timeout, Func? causeFactory, CancellationToken cancellationToken) - => source.RaceWith( - RxExtensions.FromCancellationToken(cancellationToken, causeFactory).AssumeNeverEmits(), - RxExtensions.Timeout(timeout, causeFactory).AssumeNeverEmits()); + { + var makeCancelCause = causeFactory ?? DefaultCancellationCause; + var cancelTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var registration = cancellationToken.Register(() => cancelTcs.TrySetException(makeCancelCause())); + + return await source.RaceWithSignalAndTimer(timeout, causeFactory, cancelTcs.Task).ConfigureAwait(false); + } /// /// Overload of @@ -41,7 +46,72 @@ public static Observable RaceWithSignalAndTimer( /// The source sequence to race. /// The overall duration before giving up with a timeout error. /// A token that, when cancelled, aborts the wait immediately. - /// An observable that mirrors unless the timeout or cancellation fires first. - public static Observable RaceWithSignalAndTimer(this Observable source, TimeSpan timeout, CancellationToken cancellationToken) + /// A task that resolves to 's first value unless the timeout or cancellation fires first. + public static Task RaceWithSignalAndTimer(this Observable source, TimeSpan timeout, CancellationToken cancellationToken) => source.RaceWithSignalAndTimer(timeout, causeFactory: null, cancellationToken); + + /// + /// Races 's first value against an already-existing task + /// and a timeout, whichever fires first. Use this instead of the overload + /// when the "give up" condition already exists as a task elsewhere (e.g. a task that faults when a session + /// closes), rather than one this combinator needs to build. + /// + /// + /// is expected, by contract, to only ever fault or never complete. If it completes + /// without faulting, that is a contract violation in the caller and surfaces as an + /// rather than silently returning a value. + /// + /// The type of values produced by . + /// The source sequence to race. + /// The overall duration before giving up with a timeout error. A zero or negative value disables the timeout. + /// Produces the exception thrown once elapses. Defaults to a new . + /// An already-existing task that, by contract, only ever faults or never completes. + /// A task that resolves to 's first value unless the timeout or fires first. + public static async Task RaceWithSignalAndTimer( + this Observable source, + TimeSpan timeout, + Func? causeFactory, + Task signal) + { + var makeTimeoutCause = causeFactory ?? DefaultTimeoutCause; + + var sourceTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = source.Subscribe( + onNext: value => sourceTcs.TrySetResult(value), + onError: err => sourceTcs.TrySetException(err)); + + using var timeoutCts = new CancellationTokenSource(); + var timeoutTask = Task.Delay( + timeout > TimeSpan.Zero ? timeout : System.Threading.Timeout.InfiniteTimeSpan, + timeoutCts.Token); + + try + { + var winner = await Task.WhenAny(sourceTcs.Task, signal, timeoutTask).ConfigureAwait(false); + if (winner == sourceTcs.Task) + { + return await sourceTcs.Task.ConfigureAwait(false); + } + + if (winner == signal) + { + if (signal.IsFaulted) + { + await signal.ConfigureAwait(false); + } + + throw new InvalidOperationException( + "RaceWithSignalAndTimer: the signal task completed without faulting, but was assumed to only ever fault or never complete."); + } + + throw makeTimeoutCause(); + } + finally + { + // Stops a losing timer/delay promptly instead of leaving it scheduled until it would have + // naturally elapsed; nothing observes timeoutTask after this, so its resulting cancellation + // (or, if it already won, its own exception) is discarded safely. + timeoutCts.Cancel(); + } + } } diff --git a/src/ReactiveExtensionsSharp/Extras/RetryAndRaceWithSignalAndTimer.cs b/src/ReactiveExtensionsSharp/Extras/RetryAndRaceWithSignalAndTimer.cs index 5b4aa49..42c2c24 100644 --- a/src/ReactiveExtensionsSharp/Extras/RetryAndRaceWithSignalAndTimer.cs +++ b/src/ReactiveExtensionsSharp/Extras/RetryAndRaceWithSignalAndTimer.cs @@ -11,24 +11,23 @@ public static partial class RxExtensions /// retryAndRaceWithSignalAndTimer: pipe(retry({delay}), raceWith(fromAbortSignal(...), timeout(...))). /// This is what lets a Locator action keep re-attempting a flaky operation (e.g. "find and click an element /// that may not have rendered yet") while still giving up promptly, either because the caller cancelled it - /// or because it took longer than — whichever happens first. Since the cancellation - /// and timeout branches are sources that only ever error (see - /// and ), the only way - /// this combinator produces a value is if the retried itself emits one before either - /// branch fires. + /// or because it took longer than - whichever happens first. The cancellation and + /// timeout branches only ever fault or never complete (see + /// ), + /// so the only way this combinator produces a value is if the retried itself + /// produces one before either branch fires. /// /// The type of values produced by . /// The source sequence to retry (e.g. a single Locator attempt that may throw). /// The overall duration before the action is abandoned with a timeout error. A zero or negative value disables the timeout. /// /// Produces the exception used for both the cancellation and timeout branches. Defaults to - /// for cancellation and for the timeout, - /// via the defaults of and respectively. + /// for cancellation and for the timeout. /// /// The delay between retry attempts. Defaults to 50 milliseconds. /// A token that, when cancelled, aborts the whole operation immediately. - /// An observable that retries until it succeeds, times out, or is cancelled. - public static Observable RetryAndRaceWithSignalAndTimer( + /// A task that retries until it succeeds, times out, or is cancelled. + public static Task RetryAndRaceWithSignalAndTimer( this Observable source, TimeSpan timeout, Func? causeFactory, @@ -46,7 +45,7 @@ public static Observable RetryAndRaceWithSignalAndTimer( /// The source sequence to retry. /// The overall duration before the action is abandoned with a timeout error. /// A token that, when cancelled, aborts the whole operation immediately. - /// An observable that retries until it succeeds, times out, or is cancelled. - public static Observable RetryAndRaceWithSignalAndTimer(this Observable source, TimeSpan timeout, CancellationToken cancellationToken) + /// A task that retries until it succeeds, times out, or is cancelled. + public static Task RetryAndRaceWithSignalAndTimer(this Observable source, TimeSpan timeout, CancellationToken cancellationToken) => source.RetryAndRaceWithSignalAndTimer(timeout, causeFactory: null, retryDelay: null, cancellationToken); } diff --git a/src/ReactiveExtensionsSharp/ReactiveExtensionsSharp.csproj b/src/ReactiveExtensionsSharp/ReactiveExtensionsSharp.csproj index 8f9689a..4ef4246 100644 --- a/src/ReactiveExtensionsSharp/ReactiveExtensionsSharp.csproj +++ b/src/ReactiveExtensionsSharp/ReactiveExtensionsSharp.csproj @@ -16,9 +16,9 @@ https://github.com/hardkoded/ReactiveExtensions-Sharp https://github.com/hardkoded/ReactiveExtensions-Sharp README.md - 0.2.0 - 0.2.0.0 - 0.2.0.0 + 0.3.0 + 0.3.0.0 + 0.3.0.0 diff --git a/test/ReactiveExtensionsSharp.Tests/Extras/AssumeNeverEmitsExtrasTests.cs b/test/ReactiveExtensionsSharp.Tests/Extras/AssumeNeverEmitsExtrasTests.cs deleted file mode 100644 index 5cc3bd7..0000000 --- a/test/ReactiveExtensionsSharp.Tests/Extras/AssumeNeverEmitsExtrasTests.cs +++ /dev/null @@ -1,64 +0,0 @@ -using ReactiveExtensionsSharp.Extras; - -namespace ReactiveExtensionsSharp.Tests.Extras; - -[TestFixture] -public class AssumeNeverEmitsExtrasTests -{ - [Test] - public void ShouldPropagateTheErrorFromAnErrorOnlySource() - { - var error = new InvalidOperationException("boom"); - using var signal = new ManualResetEventSlim(); - Exception? received = null; - - Observable.ThrowError(() => error) - .AssumeNeverEmits() - .Subscribe(onError: err => - { - received = err; - signal.Set(); - }); - - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(received, Is.SameAs(error)); - } - - [Test] - public void ShouldCompleteWithoutEmittingIfTheSourceCompletesWithoutErroring() - { - using var signal = new ManualResetEventSlim(); - var completed = false; - var received = new List(); - - Observable.Empty() - .AssumeNeverEmits() - .Subscribe(received.Add, onComplete: () => - { - completed = true; - signal.Set(); - }); - - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(completed, Is.True); - Assert.That(received, Is.Empty); - } - - [Test] - public void ShouldThrowIfTheSourceViolatesTheNeverEmitsContractByEmittingAValue() - { - using var signal = new ManualResetEventSlim(); - Exception? received = null; - - Observable.Of(Unit.Default) - .AssumeNeverEmits() - .Subscribe(onError: err => - { - received = err; - signal.Set(); - }); - - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(received, Is.InstanceOf()); - } -} diff --git a/test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/RaceWithSignalAndTimerTests.cs b/test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/RaceWithSignalAndTimerTests.cs index 0bba473..05407be 100644 --- a/test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/RaceWithSignalAndTimerTests.cs +++ b/test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/RaceWithSignalAndTimerTests.cs @@ -12,100 +12,77 @@ namespace ReactiveExtensionsSharp.Tests.PuppeteerScenarios; public class RaceWithSignalAndTimerTests { [Test] - public void ShouldEmitTheSourceValueIfItArrivesBeforeTimeoutOrCancellation() + public async Task ShouldResolveWithTheSourceValueIfItArrivesBeforeTimeoutOrCancellation() { - using var signal = new ManualResetEventSlim(); - var results = new List(); - var completed = false; - - Observable.Of("target-created") + var result = await Observable.Of("target-created") .RaceWithSignalAndTimer(TimeSpan.FromSeconds(5), CancellationToken.None) - .Subscribe(results.Add, onComplete: () => - { - completed = true; - signal.Set(); - }); - - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(results, Is.EqualTo(new[] { "target-created" })); - Assert.That(completed, Is.True); + .ConfigureAwait(false); + + Assert.That(result, Is.EqualTo("target-created")); } [Test] public void ShouldFailWithATimeoutErrorIfTheSourceNeverEmits() { - using var signal = new ManualResetEventSlim(); - Exception? received = null; - - Observable.Never() - .RaceWithSignalAndTimer(TimeSpan.FromMilliseconds(30), CancellationToken.None) - .Subscribe(onError: err => - { - received = err; - signal.Set(); - }); - - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(received, Is.InstanceOf()); + Assert.ThrowsAsync(() => + Observable.Never().RaceWithSignalAndTimer(TimeSpan.FromMilliseconds(30), CancellationToken.None)); } [Test] public void ShouldFailWithCancellationIfTheCallerAborts() { using var cts = new CancellationTokenSource(); - using var signal = new ManualResetEventSlim(); - Exception? received = null; - - Observable.Never() - .RaceWithSignalAndTimer(TimeSpan.FromSeconds(5), cts.Token) - .Subscribe(onError: err => - { - received = err; - signal.Set(); - }); + var task = Observable.Never().RaceWithSignalAndTimer(TimeSpan.FromSeconds(5), cts.Token); cts.Cancel(); - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(received, Is.InstanceOf()); + Assert.ThrowsAsync(() => task); } [Test] public void ShouldUseASharedCauseForBothCancellationAndTimeout() { - using var signal = new ManualResetEventSlim(); - Exception? received = null; var cause = new TimeoutException("waiting for target timed out"); - Observable.Never() - .RaceWithSignalAndTimer(TimeSpan.FromMilliseconds(30), () => cause, CancellationToken.None) - .Subscribe(onError: err => - { - received = err; - signal.Set(); - }); + var ex = Assert.ThrowsAsync(() => + Observable.Never().RaceWithSignalAndTimer(TimeSpan.FromMilliseconds(30), () => cause, CancellationToken.None)); - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(received, Is.SameAs(cause)); + Assert.That(ex, Is.SameAs(cause)); } [Test] - public void ShouldDisableTheTimeoutForAZeroOrNegativeValue() + public async Task ShouldDisableTheTimeoutForAZeroOrNegativeValue() { - using var signal = new ManualResetEventSlim(); - var results = new List(); - var subject = new ReactiveExtensionsSharp.Subjects.Subject(); - subject.AsObservable() - .RaceWithSignalAndTimer(TimeSpan.Zero, CancellationToken.None) - .Subscribe(results.Add, onComplete: signal.Set); + var task = subject.AsObservable().RaceWithSignalAndTimer(TimeSpan.Zero, CancellationToken.None); // Prove the timeout branch is truly disabled, not just long, by outliving what would otherwise fire. - Thread.Sleep(50); + await Task.Delay(50).ConfigureAwait(false); subject.OnNext("late-target"); - subject.OnCompleted(); - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(results, Is.EqualTo(new[] { "late-target" })); + var result = await task.ConfigureAwait(false); + Assert.That(result, Is.EqualTo("late-target")); + } + + [Test] + public void ShouldFailWithASignalTasksExceptionIfItFiresBeforeTimeout() + { + var signalTcs = new TaskCompletionSource(); + var cause = new InvalidOperationException("session closed"); + signalTcs.SetException(cause); + + var ex = Assert.ThrowsAsync(() => + Observable.Never().RaceWithSignalAndTimer(TimeSpan.FromSeconds(5), causeFactory: null, signalTcs.Task)); + + Assert.That(ex, Is.SameAs(cause)); + } + + [Test] + public void ShouldFailWithATimeoutErrorIfItFiresBeforeAPendingSignalTask() + { + var signalTcs = new TaskCompletionSource(); + + Assert.ThrowsAsync(() => + Observable.Never().RaceWithSignalAndTimer(TimeSpan.FromMilliseconds(30), causeFactory: null, signalTcs.Task)); } } diff --git a/test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/RetryAndRaceWithSignalAndTimerTests.cs b/test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/RetryAndRaceWithSignalAndTimerTests.cs index c7430c2..212d5f4 100644 --- a/test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/RetryAndRaceWithSignalAndTimerTests.cs +++ b/test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/RetryAndRaceWithSignalAndTimerTests.cs @@ -5,7 +5,7 @@ namespace ReactiveExtensionsSharp.Tests.PuppeteerScenarios; /// /// Modeled directly on how Puppeteer's Locator actions (click/fill/hover/wait) compose /// retry + cancellation + timeout: pipe(retry({delay}), raceWith(fromAbortSignal(...), timeout(...))). -/// No upstream rxjs spec equivalent exists for this — it's Puppeteer's own combinator, not rxjs's — so this +/// No upstream rxjs spec equivalent exists for this - it's Puppeteer's own combinator, not rxjs's - so this /// suite is hand-written per the project's "Puppeteer usage must show up in tests" rule. This is also M2's /// stated exit criteria: prove retry+race+timeout work together before touching the real puppeteer-sharp repo. /// @@ -26,44 +26,27 @@ private static Observable FakeClick(int failuresBeforeSuccess, List } [Test] - public void ClickShouldSucceedAfterRetryingUntilTheElementIsReady() + public async Task ClickShouldSucceedAfterRetryingUntilTheElementIsReady() { var attempts = new List(); - var results = new List(); - var completed = false; - using var signal = new ManualResetEventSlim(); - FakeClick(3, attempts) + var result = await FakeClick(3, attempts) .RetryAndRaceWithSignalAndTimer(TimeSpan.FromSeconds(5), CancellationToken.None) - .Subscribe(results.Add, onComplete: () => - { - completed = true; - signal.Set(); - }); + .ConfigureAwait(false); - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); Assert.That(attempts, Is.EqualTo(new[] { 1, 2, 3, 4 })); - Assert.That(results, Is.EqualTo(new[] { "clicked" })); - Assert.That(completed, Is.True); + Assert.That(result, Is.EqualTo("clicked")); } [Test] public void ClickShouldFailWithATimeoutErrorIfTheElementNeverBecomesReady() { var attempts = new List(); - using var signal = new ManualResetEventSlim(); - Exception? received = null; - FakeClick(int.MaxValue, attempts) - .RetryAndRaceWithSignalAndTimer(TimeSpan.FromMilliseconds(60), causeFactory: null, retryDelay: TimeSpan.FromMilliseconds(10), CancellationToken.None) - .Subscribe(onError: err => - { - received = err; - signal.Set(); - }); + Assert.ThrowsAsync(() => + FakeClick(int.MaxValue, attempts) + .RetryAndRaceWithSignalAndTimer(TimeSpan.FromMilliseconds(60), causeFactory: null, retryDelay: TimeSpan.FromMilliseconds(10), CancellationToken.None)); - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(received, Is.InstanceOf()); Assert.That(attempts.Count, Is.GreaterThan(1), "should have retried at least once before timing out"); } @@ -72,41 +55,26 @@ public void ClickShouldFailWithCancellationIfTheCallerAborts() { var attempts = new List(); using var cts = new CancellationTokenSource(); - using var signal = new ManualResetEventSlim(); - Exception? received = null; - FakeClick(int.MaxValue, attempts) - .RetryAndRaceWithSignalAndTimer(TimeSpan.FromSeconds(5), causeFactory: null, retryDelay: TimeSpan.FromMilliseconds(10), cts.Token) - .Subscribe(onError: err => - { - received = err; - signal.Set(); - }); + var task = FakeClick(int.MaxValue, attempts) + .RetryAndRaceWithSignalAndTimer(TimeSpan.FromSeconds(5), causeFactory: null, retryDelay: TimeSpan.FromMilliseconds(10), cts.Token); Thread.Sleep(30); cts.Cancel(); - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(received, Is.InstanceOf()); + Assert.ThrowsAsync(() => task); } [Test] public void ClickShouldUseASharedCauseForBothCancellationAndTimeout() { var attempts = new List(); - using var signal = new ManualResetEventSlim(); - Exception? received = null; var cause = new TimeoutException("waiting for selector timed out"); - FakeClick(int.MaxValue, attempts) - .RetryAndRaceWithSignalAndTimer(TimeSpan.FromMilliseconds(30), () => cause, TimeSpan.FromMilliseconds(10), CancellationToken.None) - .Subscribe(onError: err => - { - received = err; - signal.Set(); - }); + var ex = Assert.ThrowsAsync(() => + FakeClick(int.MaxValue, attempts) + .RetryAndRaceWithSignalAndTimer(TimeSpan.FromMilliseconds(30), () => cause, TimeSpan.FromMilliseconds(10), CancellationToken.None)); - Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); - Assert.That(received, Is.SameAs(cause)); + Assert.That(ex, Is.SameAs(cause)); } } diff --git a/test/ReactiveExtensionsSharp.Tests/Samples/RetryUntilTimeoutSample.cs b/test/ReactiveExtensionsSharp.Tests/Samples/RetryUntilTimeoutSample.cs index 707e434..7bf3786 100644 --- a/test/ReactiveExtensionsSharp.Tests/Samples/RetryUntilTimeoutSample.cs +++ b/test/ReactiveExtensionsSharp.Tests/Samples/RetryUntilTimeoutSample.cs @@ -17,6 +17,6 @@ public static class RetryUntilTimeoutSample public static async Task FindElementOnceItRendersAsync(Func> tryFindElement, CancellationToken cancellationToken) => await Observable.Defer(() => Observable.From(tryFindElement())) .RetryAndRaceWithSignalAndTimer(TimeSpan.FromSeconds(5), cancellationToken) - .FirstValueFrom().ConfigureAwait(false); + .ConfigureAwait(false); // end-snippet }