diff --git a/src/RxSharp/Extras/AssumeNeverEmits.cs b/src/RxSharp/Extras/AssumeNeverEmits.cs new file mode 100644 index 0000000..6727380 --- /dev/null +++ b/src/RxSharp/Extras/AssumeNeverEmits.cs @@ -0,0 +1,31 @@ +using RxSharp.Operators; + +namespace RxSharp.Extras; + +/// Extension methods widening an error-only to another element type. +public static partial class Extensions +{ + /// + /// 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/RxSharp/Extras/FilterAsync.cs b/src/RxSharp/Extras/FilterAsync.cs index 0516192..e789517 100644 --- a/src/RxSharp/Extras/FilterAsync.cs +++ b/src/RxSharp/Extras/FilterAsync.cs @@ -3,7 +3,7 @@ namespace RxSharp.Extras; /// Extension methods providing an async-predicate flavor of Filter. -public static class FilterAsyncExtras +public static partial class Extensions { /// /// An operator supporting an async predicate, mirroring Puppeteer's own filterAsync helper (implemented, diff --git a/src/RxSharp/Extras/FromCancellationToken.cs b/src/RxSharp/Extras/FromCancellationToken.cs index b424ad6..7c3a587 100644 --- a/src/RxSharp/Extras/FromCancellationToken.cs +++ b/src/RxSharp/Extras/FromCancellationToken.cs @@ -1,7 +1,7 @@ namespace RxSharp.Extras; /// Puppeteer-flavored combinators built on top of the core primitives — the C# analogues of the helpers Puppeteer itself layers on top of rxjs (see CLAUDE.md's "Puppeteer-essential surface"). -public static class CancellationExtras +public static partial class Extensions { /// /// An observable that never emits and errors as soon as is cancelled. @@ -18,7 +18,7 @@ public static class CancellationExtras public static Observable FromCancellationToken(CancellationToken cancellationToken, Func? causeFactory = null) => new Observable(subscriber => { - var makeCause = causeFactory ?? DefaultCause; + var makeCause = causeFactory ?? DefaultCancellationCause; if (cancellationToken.IsCancellationRequested) { @@ -30,5 +30,5 @@ public static Observable FromCancellationToken(CancellationToken cancellat return new Subscription(() => registration.Dispose()); }); - private static Exception DefaultCause() => new OperationCanceledException(); + private static Exception DefaultCancellationCause() => new OperationCanceledException(); } diff --git a/src/RxSharp/Extras/FromEventBuffered.cs b/src/RxSharp/Extras/FromEventBuffered.cs new file mode 100644 index 0000000..7a7dad4 --- /dev/null +++ b/src/RxSharp/Extras/FromEventBuffered.cs @@ -0,0 +1,96 @@ +using RxSharp.Subjects; + +namespace RxSharp.Extras; + +/// +/// A handle to an eagerly-attached, buffered .NET event source created by +/// . Disposing detaches the underlying event handler. +/// +/// The type of the event's payload. +public sealed class BufferedEventSource : IDisposable +{ + private readonly ReplaySubject _subject; + private readonly Action _detach; + private bool _disposed; + + internal BufferedEventSource(ReplaySubject subject, Action detach) + { + _subject = subject; + _detach = detach; + } + + /// + /// Exposes the buffered event payloads as an . Each subscriber first receives + /// any payloads buffered since this source was created, then live payloads as they arrive - exactly like + /// subscribing to a directly, because that is what this wraps. + /// + /// An observable over this source's buffered and live event payloads. + public Observable AsObservable() => _subject.AsObservable(); + + /// Detaches the underlying event handler and disposes the buffer. Safe to call more than once. + public void Dispose() + { + if (_disposed) + { + return; + } + + _disposed = true; + _detach(); + _subject.Dispose(); + } +} + +/// Extension methods providing an eagerly-attached, buffered variant of . +public static partial class Extensions +{ + /// + /// Like , but attaches the underlying event handler + /// immediately - when this method is called - rather than lazily at subscribe time, buffering up to + /// payloads that arrive before any subscriber attaches and replaying them to + /// the first subscriber(s). + /// + /// + /// Ordinary is cold: nothing is attached to the + /// underlying event until Subscribe is called, matching every other RxSharp source. This is + /// deliberately not that. It exists for the narrow case where a handler must be attached before + /// synchronously checking some existing state that might already satisfy what the caller is waiting for + /// (e.g. checking a collection for an already-matching item) without losing an event that fires in the real + /// gap between attaching the raw handler and actually subscribing to the returned observable. That gap is + /// provably real on .NET, where event delivery can run on another thread - unlike single-threaded JS, where + /// rxjs's own fromEmitterEvent has no equivalent problem, since nothing can fire between two + /// synchronous statements. Most code should use the ordinary, cold + /// instead; reach for this only when that gap is a + /// real, provable race, not by default. + /// + /// The delegate type of the event handler. + /// The type of the event's payload. + /// Called immediately, with the handler to add to the event. + /// Called on , with the same handler, to remove it from the event. + /// Converts an callback into the event's actual delegate shape. + /// The maximum number of most-recent payloads kept for replay. Defaults to 1. + /// A handle exposing the buffered payloads as an observable, and detaching the handler on disposal. + public static BufferedEventSource FromEventBuffered( + Action addHandler, + Action removeHandler, + Func, TDelegate> conversion, + int bufferSize = 1) + { + var subject = new ReplaySubject(bufferSize); + var handler = conversion(subject.OnNext); + addHandler(handler); + return new BufferedEventSource(subject, () => removeHandler(handler)); + } + + /// The common case of for standard -shaped .NET events. + /// The type of the event's payload. + /// Called immediately, with the handler to add to the event. + /// Called on , with the same handler, to remove it from the event. + /// The maximum number of most-recent payloads kept for replay. Defaults to 1. + /// A handle exposing the buffered payloads as an observable, and detaching the handler on disposal. + public static BufferedEventSource FromEventBuffered( + Action> addHandler, + Action> removeHandler, + int bufferSize = 1) + => FromEventBuffered, TEventArgs>(addHandler, removeHandler, onNext => (_, args) => onNext(args), bufferSize); +} diff --git a/src/RxSharp/Extras/RaceWithSignalAndTimer.cs b/src/RxSharp/Extras/RaceWithSignalAndTimer.cs new file mode 100644 index 0000000..3478f77 --- /dev/null +++ b/src/RxSharp/Extras/RaceWithSignalAndTimer.cs @@ -0,0 +1,47 @@ +using RxSharp.Operators; + +namespace RxSharp.Extras; + +/// Extension methods racing a single subscription against cancellation and a timeout. +public static partial class Extensions +{ + /// + /// Races 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. + /// + /// 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. + /// 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( + this Observable source, + TimeSpan timeout, + Func? causeFactory, + CancellationToken cancellationToken) + => source.RaceWith( + Extensions.FromCancellationToken(cancellationToken, causeFactory).AssumeNeverEmits(), + Extensions.Timeout(timeout, causeFactory).AssumeNeverEmits()); + + /// + /// Overload of + /// using the default cause factory (/). + /// + /// The type of values produced by . + /// 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) + => source.RaceWithSignalAndTimer(timeout, causeFactory: null, cancellationToken); +} diff --git a/src/RxSharp/Extras/RetryAndRaceWithSignalAndTimer.cs b/src/RxSharp/Extras/RetryAndRaceWithSignalAndTimer.cs index f90648b..79e95d1 100644 --- a/src/RxSharp/Extras/RetryAndRaceWithSignalAndTimer.cs +++ b/src/RxSharp/Extras/RetryAndRaceWithSignalAndTimer.cs @@ -3,7 +3,7 @@ namespace RxSharp.Extras; /// Extension methods composing retry, cancellation, and timeout into the single combinator Puppeteer's Locator actions rely on. -public static class RetryAndRaceWithSignalAndTimerExtras +public static partial class Extensions { /// /// The combinator behind Puppeteer's Locator actions (click/fill/hover/wait): retry the source @@ -13,7 +13,7 @@ public static class RetryAndRaceWithSignalAndTimerExtras /// 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 + /// and ), the only way /// this combinator produces a value is if the retried itself emits one before either /// branch fires. /// @@ -23,7 +23,7 @@ public static class RetryAndRaceWithSignalAndTimerExtras /// /// 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. + /// via the defaults of and respectively. /// /// The delay between retry attempts. Defaults to 50 milliseconds. /// A token that, when cancelled, aborts the whole operation immediately. @@ -36,9 +36,7 @@ public static Observable RetryAndRaceWithSignalAndTimer( CancellationToken cancellationToken) => source .Retry(delay: retryDelay ?? TimeSpan.FromMilliseconds(50)) - .RaceWith( - CancellationExtras.FromCancellationToken(cancellationToken, causeFactory).Map(NeverReached), - TimeoutExtras.Timeout(timeout, causeFactory).Map(NeverReached)); + .RaceWithSignalAndTimer(timeout, causeFactory, cancellationToken); /// /// Overload of @@ -51,9 +49,4 @@ public static Observable RetryAndRaceWithSignalAndTimer( /// An observable that retries until it succeeds, times out, or is cancelled. public static Observable RetryAndRaceWithSignalAndTimer(this Observable source, TimeSpan timeout, CancellationToken cancellationToken) => source.RetryAndRaceWithSignalAndTimer(timeout, causeFactory: null, retryDelay: null, cancellationToken); - - // FromCancellationToken/Timeout only ever call OnError, never OnNext — this projection exists purely so - // the two Observable notifiers type-check inside RaceWith(Observable), matching how rxjs relies - // on TypeScript accepting Observable anywhere an Observable is expected. C# has no bottom type. - private static TResult NeverReached(Unit value) => throw new InvalidOperationException("unreachable: this observable only ever errors"); } diff --git a/src/RxSharp/Extras/Timeout.cs b/src/RxSharp/Extras/Timeout.cs index a315881..d8dce67 100644 --- a/src/RxSharp/Extras/Timeout.cs +++ b/src/RxSharp/Extras/Timeout.cs @@ -3,7 +3,7 @@ namespace RxSharp.Extras; /// Extension methods providing a standalone, -throwing timer observable. -public static class TimeoutExtras +public static partial class Extensions { /// /// An observable that never emits and errors after elapses, or never errors at all if @@ -25,9 +25,9 @@ public static Observable Timeout(TimeSpan delay, Func? causeFac return Observable.Never(); } - var makeCause = causeFactory ?? DefaultCause; + var makeCause = causeFactory ?? DefaultTimeoutCause; return Observable.Timer(delay, scheduler).Map(_ => throw makeCause()); } - private static Exception DefaultCause() => new TimeoutException(); + private static Exception DefaultTimeoutCause() => new TimeoutException(); } diff --git a/src/RxSharp/RxSharp.csproj b/src/RxSharp/RxSharp.csproj index 0aad034..21e79bc 100644 --- a/src/RxSharp/RxSharp.csproj +++ b/src/RxSharp/RxSharp.csproj @@ -16,9 +16,9 @@ https://github.com/hardkoded/ReactiveExtensions-Sharp https://github.com/hardkoded/ReactiveExtensions-Sharp README.md - 0.1.1 - 0.1.1.0 - 0.1.1.0 + 0.1.2 + 0.1.2.0 + 0.1.2.0 diff --git a/test/RxSharp.Tests/Extras/AssumeNeverEmitsExtrasTests.cs b/test/RxSharp.Tests/Extras/AssumeNeverEmitsExtrasTests.cs new file mode 100644 index 0000000..d84c309 --- /dev/null +++ b/test/RxSharp.Tests/Extras/AssumeNeverEmitsExtrasTests.cs @@ -0,0 +1,64 @@ +using RxSharp.Extras; + +namespace RxSharp.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/RxSharp.Tests/Extras/CancellationExtrasTests.cs b/test/RxSharp.Tests/Extras/CancellationExtrasTests.cs index 219ddd5..2611d4f 100644 --- a/test/RxSharp.Tests/Extras/CancellationExtrasTests.cs +++ b/test/RxSharp.Tests/Extras/CancellationExtrasTests.cs @@ -12,7 +12,7 @@ public void ShouldErrorImmediatelyIfTokenIsAlreadyCancelled() cts.Cancel(); Exception? received = null; - CancellationExtras.FromCancellationToken(cts.Token).Subscribe(onError: err => received = err); + Extensions.FromCancellationToken(cts.Token).Subscribe(onError: err => received = err); Assert.That(received, Is.InstanceOf()); } @@ -22,7 +22,7 @@ public void ShouldErrorAsSoonAsTheTokenIsCancelled() { using var cts = new CancellationTokenSource(); Exception? received = null; - CancellationExtras.FromCancellationToken(cts.Token).Subscribe(onError: err => received = err); + Extensions.FromCancellationToken(cts.Token).Subscribe(onError: err => received = err); Assert.That(received, Is.Null); @@ -36,7 +36,7 @@ public void ShouldNeverEmitAValue() { using var cts = new CancellationTokenSource(); var nextCalled = false; - var subscription = CancellationExtras.FromCancellationToken(cts.Token).Subscribe(_ => nextCalled = true); + var subscription = Extensions.FromCancellationToken(cts.Token).Subscribe(_ => nextCalled = true); subscription.Dispose(); cts.Cancel(); @@ -52,7 +52,7 @@ public void ShouldUseTheCustomCauseFactory() var cause = new InvalidOperationException("custom cause"); Exception? received = null; - CancellationExtras.FromCancellationToken(cts.Token, () => cause).Subscribe(onError: err => received = err); + Extensions.FromCancellationToken(cts.Token, () => cause).Subscribe(onError: err => received = err); Assert.That(received, Is.SameAs(cause)); } diff --git a/test/RxSharp.Tests/Extras/FromEventBufferedExtrasTests.cs b/test/RxSharp.Tests/Extras/FromEventBufferedExtrasTests.cs new file mode 100644 index 0000000..0a68745 --- /dev/null +++ b/test/RxSharp.Tests/Extras/FromEventBufferedExtrasTests.cs @@ -0,0 +1,116 @@ +using RxSharp.Extras; + +namespace RxSharp.Tests.Extras; + +[TestFixture] +public class FromEventBufferedExtrasTests +{ + private sealed class FakeTargetManager + { + public event EventHandler? TargetCreated; + + public void RaiseTargetCreated(string targetId) => TargetCreated?.Invoke(this, targetId); + } + + [Test] + public void ShouldAttachTheHandlerImmediatelyRatherThanAtSubscribeTime() + { + var manager = new FakeTargetManager(); + + using var source = Extensions.FromEventBuffered( + h => manager.TargetCreated += h, + h => manager.TargetCreated -= h); + + // Raised before AsObservable() is ever subscribed to - this is the exact gap FromEventBuffered exists + // to close (a plain, cold FromEvent would lose this). + manager.RaiseTargetCreated("early-target"); + + var received = new List(); + source.AsObservable().Subscribe(received.Add); + + Assert.That(received, Is.EqualTo(new[] { "early-target" })); + } + + [Test] + public void ShouldReplayBufferedValuesToALateSubscriberThenDeliverLiveValues() + { + var manager = new FakeTargetManager(); + using var source = Extensions.FromEventBuffered( + h => manager.TargetCreated += h, + h => manager.TargetCreated -= h); + + manager.RaiseTargetCreated("buffered-target"); + + var received = new List(); + source.AsObservable().Subscribe(received.Add); + manager.RaiseTargetCreated("live-target"); + + Assert.That(received, Is.EqualTo(new[] { "buffered-target", "live-target" })); + } + + [Test] + public void ShouldDeliverToEveryIndependentSubscriber() + { + var manager = new FakeTargetManager(); + using var source = Extensions.FromEventBuffered( + h => manager.TargetCreated += h, + h => manager.TargetCreated -= h); + + var first = new List(); + var second = new List(); + source.AsObservable().Subscribe(first.Add); + source.AsObservable().Subscribe(second.Add); + + manager.RaiseTargetCreated("shared-target"); + + Assert.That(first, Is.EqualTo(new[] { "shared-target" })); + Assert.That(second, Is.EqualTo(new[] { "shared-target" })); + } + + [Test] + public void ShouldOnlyReplayUpToTheRequestedBufferSize() + { + var manager = new FakeTargetManager(); + using var source = Extensions.FromEventBuffered( + h => manager.TargetCreated += h, + h => manager.TargetCreated -= h, + bufferSize: 2); + + manager.RaiseTargetCreated("one"); + manager.RaiseTargetCreated("two"); + manager.RaiseTargetCreated("three"); + + var received = new List(); + source.AsObservable().Subscribe(received.Add); + + Assert.That(received, Is.EqualTo(new[] { "two", "three" })); + } + + [Test] + public void ShouldDetachTheHandlerOnDispose() + { + var manager = new FakeTargetManager(); + var source = Extensions.FromEventBuffered( + h => manager.TargetCreated += h, + h => manager.TargetCreated -= h); + + source.Dispose(); + manager.RaiseTargetCreated("after-dispose"); + + // If the handler weren't detached, this would throw ObjectDisposedException from the underlying + // subject instead of silently doing nothing. + Assert.DoesNotThrow(() => manager.RaiseTargetCreated("another-after-dispose")); + } + + [Test] + public void ShouldBeSafeToDisposeMoreThanOnce() + { + var manager = new FakeTargetManager(); + var source = Extensions.FromEventBuffered( + h => manager.TargetCreated += h, + h => manager.TargetCreated -= h); + + source.Dispose(); + Assert.DoesNotThrow(() => source.Dispose()); + } +} diff --git a/test/RxSharp.Tests/Extras/TimeoutExtrasTests.cs b/test/RxSharp.Tests/Extras/TimeoutExtrasTests.cs index a773b13..625f46a 100644 --- a/test/RxSharp.Tests/Extras/TimeoutExtrasTests.cs +++ b/test/RxSharp.Tests/Extras/TimeoutExtrasTests.cs @@ -11,7 +11,7 @@ public void ShouldErrorAfterTheDelay() using var signal = new ManualResetEventSlim(); Exception? received = null; - TimeoutExtras.Timeout(TimeSpan.FromMilliseconds(20)).Subscribe(onError: err => + Extensions.Timeout(TimeSpan.FromMilliseconds(20)).Subscribe(onError: err => { received = err; signal.Set(); @@ -25,7 +25,7 @@ public void ShouldErrorAfterTheDelay() public void ShouldNeverErrorWhenDelayIsZero() { var errored = false; - TimeoutExtras.Timeout(TimeSpan.Zero).Subscribe(onError: _ => errored = true); + Extensions.Timeout(TimeSpan.Zero).Subscribe(onError: _ => errored = true); Thread.Sleep(50); @@ -39,7 +39,7 @@ public void ShouldUseTheCustomCauseFactory() var cause = new InvalidOperationException("custom timeout cause"); Exception? received = null; - TimeoutExtras.Timeout(TimeSpan.FromMilliseconds(10), () => cause).Subscribe(onError: err => + Extensions.Timeout(TimeSpan.FromMilliseconds(10), () => cause).Subscribe(onError: err => { received = err; signal.Set(); diff --git a/test/RxSharp.Tests/PuppeteerScenarios/RaceWithSignalAndTimerTests.cs b/test/RxSharp.Tests/PuppeteerScenarios/RaceWithSignalAndTimerTests.cs new file mode 100644 index 0000000..1f0e42f --- /dev/null +++ b/test/RxSharp.Tests/PuppeteerScenarios/RaceWithSignalAndTimerTests.cs @@ -0,0 +1,111 @@ +using RxSharp.Extras; + +namespace RxSharp.Tests.PuppeteerScenarios; + +/// +/// Modeled on how Puppeteer's Browser.waitForTarget/Page.waitForRequest-shaped methods race a +/// single wait against cancellation and a timeout, without retrying: pipe(filterAsync(predicate), +/// raceWith(fromAbortSignal(...), timeout(...))). Like RetryAndRaceWithSignalAndTimer, no upstream +/// rxjs spec equivalent exists for this combinator. +/// +[TestFixture] +public class RaceWithSignalAndTimerTests +{ + [Test] + public void ShouldEmitTheSourceValueIfItArrivesBeforeTimeoutOrCancellation() + { + using var signal = new ManualResetEventSlim(); + var results = new List(); + var completed = false; + + 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); + } + + [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()); + } + + [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(); + }); + + cts.Cancel(); + + Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); + Assert.That(received, Is.InstanceOf()); + } + + [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(); + }); + + Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); + Assert.That(received, Is.SameAs(cause)); + } + + [Test] + public void ShouldDisableTheTimeoutForAZeroOrNegativeValue() + { + using var signal = new ManualResetEventSlim(); + var results = new List(); + + var subject = new RxSharp.Subjects.Subject(); + subject.AsObservable() + .RaceWithSignalAndTimer(TimeSpan.Zero, CancellationToken.None) + .Subscribe(results.Add, onComplete: signal.Set); + + // Prove the timeout branch is truly disabled, not just long, by outliving what would otherwise fire. + Thread.Sleep(50); + subject.OnNext("late-target"); + subject.OnCompleted(); + + Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True); + Assert.That(results, Is.EqualTo(new[] { "late-target" })); + } +}