Add AssumeNeverEmits, RaceWithSignalAndTimer, and FromEventBuffered

src/RxSharp/Extras/AssumeNeverEmits.cs

@@ -0,0 +1,31 @@
+using RxSharp.Operators;
+
+namespace RxSharp.Extras;
+
+/// <summary>Extension methods widening an error-only <see cref="Observable{T}"/> to another element type.</summary>
+public static partial class Extensions
+{
+    /// <summary>
+    /// Widens an <see cref="Observable{Unit}"/> that only ever calls <see cref="IObserver{T}.OnError"/> (such as
+    /// <see cref="Extensions.FromCancellationToken"/> or <see cref="Extensions.Timeout"/>) so it
+    /// type-checks anywhere an <see cref="Observable{T}"/> is expected - most commonly as one of the branches
+    /// passed to <see cref="RaceOperator.RaceWith{T}"/> alongside a source that actually produces
+    /// <typeparamref name="TResult"/> values. Mirrors how rxjs relies on TypeScript structurally accepting
+    /// <c>Observable&lt;never&gt;</c> anywhere an <c>Observable&lt;T&gt;</c> is expected; C# has no bottom type,
+    /// so this exists to fill that gap explicitly.
+    /// </summary>
+    /// <remarks>
+    /// This is only safe for a <paramref name="source"/> that is guaranteed, by contract, to never call
+    /// <see cref="IObserver{T}.OnNext"/>. If it ever does, the returned observable throws
+    /// <see cref="InvalidOperationException"/> 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 <c>never</c>. Do not
+    /// apply this to a source that might legitimately emit.
+    /// </remarks>
+    /// <typeparam name="TResult">The element type to widen to.</typeparam>
+    /// <param name="source">An observable that only ever errors, never emits.</param>
+    /// <returns>An <see cref="Observable{TResult}"/> that mirrors <paramref name="source"/>'s errors and throws if it ever emits.</returns>
+    public static Observable<TResult> AssumeNeverEmits<TResult>(this Observable<Unit> source)
+        => source.Map<Unit, TResult>(_ => 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."));
+}

src/RxSharp/Extras/FilterAsync.cs

@@ -3,7 +3,7 @@
 namespace RxSharp.Extras;
 
 /// <summary>Extension methods providing an async-predicate flavor of <c>Filter</c>.</summary>
-public static class FilterAsyncExtras
+public static partial class Extensions
 {
     /// <summary>
     /// An operator supporting an async predicate, mirroring Puppeteer's own <c>filterAsync</c> helper (implemented,

src/RxSharp/Extras/FromCancellationToken.cs

@@ -1,7 +1,7 @@
 namespace RxSharp.Extras;
 
 /// <summary>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").</summary>
-public static class CancellationExtras
+public static partial class Extensions
 {
     /// <summary>
     /// An observable that never emits and errors as soon as <paramref name="cancellationToken"/> is cancelled.
@@ -18,7 +18,7 @@ public static class CancellationExtras
     public static Observable<Unit> FromCancellationToken(CancellationToken cancellationToken, Func<Exception>? causeFactory = null)
         => new Observable<Unit>(subscriber =>
         {
-            var makeCause = causeFactory ?? DefaultCause;
+            var makeCause = causeFactory ?? DefaultCancellationCause;
 
             if (cancellationToken.IsCancellationRequested)
             {
@@ -30,5 +30,5 @@ public static Observable<Unit> FromCancellationToken(CancellationToken cancellat
             return new Subscription(() => registration.Dispose());
         });
 
-    private static Exception DefaultCause() => new OperationCanceledException();
+    private static Exception DefaultCancellationCause() => new OperationCanceledException();
 }

src/RxSharp/Extras/FromEventBuffered.cs

@@ -0,0 +1,96 @@
+using RxSharp.Subjects;
+
+namespace RxSharp.Extras;
+
+/// <summary>
+/// A handle to an eagerly-attached, buffered .NET event source created by
+/// <see cref="Extensions.FromEventBuffered{TDelegate, TEventArgs}"/>. Disposing detaches the underlying event handler.
+/// </summary>
+/// <typeparam name="TEventArgs">The type of the event's payload.</typeparam>
+public sealed class BufferedEventSource<TEventArgs> : IDisposable
+{
+    private readonly ReplaySubject<TEventArgs> _subject;
+    private readonly Action _detach;
+    private bool _disposed;
+
+    internal BufferedEventSource(ReplaySubject<TEventArgs> subject, Action detach)
+    {
+        _subject = subject;
+        _detach = detach;
+    }
+
+    /// <summary>
+    /// Exposes the buffered event payloads as an <see cref="Observable{T}"/>. Each subscriber first receives
+    /// any payloads buffered since this source was created, then live payloads as they arrive - exactly like
+    /// subscribing to a <see cref="ReplaySubject{T}"/> directly, because that is what this wraps.
+    /// </summary>
+    /// <returns>An observable over this source's buffered and live event payloads.</returns>
+    public Observable<TEventArgs> AsObservable() => _subject.AsObservable();
+
+    /// <summary>Detaches the underlying event handler and disposes the buffer. Safe to call more than once.</summary>
+    public void Dispose()
+    {
+        if (_disposed)
+        {
+            return;
+        }
+
+        _disposed = true;
+        _detach();
+        _subject.Dispose();
+    }
+}
+
+/// <summary>Extension methods providing an eagerly-attached, buffered variant of <see cref="Observable.FromEvent{TEventArgs}"/>.</summary>
+public static partial class Extensions
+{
+    /// <summary>
+    /// Like <see cref="Observable.FromEvent{TDelegate, TEventArgs}"/>, but attaches the underlying event handler
+    /// immediately - when this method is called - rather than lazily at subscribe time, buffering up to
+    /// <paramref name="bufferSize"/> payloads that arrive before any subscriber attaches and replaying them to
+    /// the first subscriber(s).
+    /// </summary>
+    /// <remarks>
+    /// Ordinary <see cref="Observable.FromEvent{TDelegate, TEventArgs}"/> is cold: nothing is attached to the
+    /// underlying event until <c>Subscribe</c> 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 <c>fromEmitterEvent</c> has no equivalent problem, since nothing can fire between two
+    /// synchronous statements. Most code should use the ordinary, cold
+    /// <see cref="Observable.FromEvent{TDelegate, TEventArgs}"/> instead; reach for this only when that gap is a
+    /// real, provable race, not by default.
+    /// </remarks>
+    /// <typeparam name="TDelegate">The delegate type of the event handler.</typeparam>
+    /// <typeparam name="TEventArgs">The type of the event's payload.</typeparam>
+    /// <param name="addHandler">Called immediately, with the handler to add to the event.</param>
+    /// <param name="removeHandler">Called on <see cref="BufferedEventSource{TEventArgs}.Dispose"/>, with the same handler, to remove it from the event.</param>
+    /// <param name="conversion">Converts an <see cref="Action{TEventArgs}"/> callback into the event's actual delegate shape.</param>
+    /// <param name="bufferSize">The maximum number of most-recent payloads kept for replay. Defaults to 1.</param>
+    /// <returns>A handle exposing the buffered payloads as an observable, and detaching the handler on disposal.</returns>
+    public static BufferedEventSource<TEventArgs> FromEventBuffered<TDelegate, TEventArgs>(
+        Action<TDelegate> addHandler,
+        Action<TDelegate> removeHandler,
+        Func<Action<TEventArgs>, TDelegate> conversion,
+        int bufferSize = 1)
+    {
+        var subject = new ReplaySubject<TEventArgs>(bufferSize);
+        var handler = conversion(subject.OnNext);
+        addHandler(handler);
+        return new BufferedEventSource<TEventArgs>(subject, () => removeHandler(handler));
+    }
+
+    /// <summary>The common case of <see cref="FromEventBuffered{TDelegate, TEventArgs}"/> for standard <see cref="EventHandler{TEventArgs}"/>-shaped .NET events.</summary>
+    /// <typeparam name="TEventArgs">The type of the event's payload.</typeparam>
+    /// <param name="addHandler">Called immediately, with the handler to add to the event.</param>
+    /// <param name="removeHandler">Called on <see cref="BufferedEventSource{TEventArgs}.Dispose"/>, with the same handler, to remove it from the event.</param>
+    /// <param name="bufferSize">The maximum number of most-recent payloads kept for replay. Defaults to 1.</param>
+    /// <returns>A handle exposing the buffered payloads as an observable, and detaching the handler on disposal.</returns>
+    public static BufferedEventSource<TEventArgs> FromEventBuffered<TEventArgs>(
+        Action<EventHandler<TEventArgs>> addHandler,
+        Action<EventHandler<TEventArgs>> removeHandler,
+        int bufferSize = 1)
+        => FromEventBuffered<EventHandler<TEventArgs>, TEventArgs>(addHandler, removeHandler, onNext => (_, args) => onNext(args), bufferSize);
+}

src/RxSharp/Extras/RaceWithSignalAndTimer.cs

@@ -0,0 +1,47 @@
+using RxSharp.Operators;
+
+namespace RxSharp.Extras;
+
+/// <summary>Extension methods racing a single subscription against cancellation and a timeout.</summary>
+public static partial class Extensions
+{
+    /// <summary>
+    /// Races <paramref name="source"/> against cancellation and a timeout, whichever fires first. The
+    /// non-retrying half of <see cref="Extensions.RetryAndRaceWithSignalAndTimer{T}(Observable{T}, TimeSpan, Func{Exception}, TimeSpan?, CancellationToken)"/>
+    /// - 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.
+    /// </summary>
+    /// <typeparam name="T">The type of values produced by <paramref name="source"/>.</typeparam>
+    /// <param name="source">The source sequence to race.</param>
+    /// <param name="timeout">The overall duration before giving up with a timeout error. A zero or negative value disables the timeout.</param>
+    /// <param name="causeFactory">
+    /// Produces the exception used for both the cancellation and timeout branches. Defaults to
+    /// <see cref="OperationCanceledException"/> for cancellation and <see cref="TimeoutException"/> for the timeout,
+    /// via the defaults of <see cref="Extensions.FromCancellationToken"/> and <see cref="Extensions.Timeout"/> respectively.
+    /// Since one factory covers both branches, a caller needing to tell the two apart by exception type should
+    /// pass <see langword="null"/> here (so each branch keeps its own distinct default type) and catch/rethrow
+    /// as needed at the call site.
+    /// </param>
+    /// <param name="cancellationToken">A token that, when cancelled, aborts the wait immediately.</param>
+    /// <returns>An observable that mirrors <paramref name="source"/> unless the timeout or cancellation fires first.</returns>
+    public static Observable<T> RaceWithSignalAndTimer<T>(
+        this Observable<T> source,
+        TimeSpan timeout,
+        Func<Exception>? causeFactory,
+        CancellationToken cancellationToken)
+        => source.RaceWith(
+            Extensions.FromCancellationToken(cancellationToken, causeFactory).AssumeNeverEmits<T>(),
+            Extensions.Timeout(timeout, causeFactory).AssumeNeverEmits<T>());
+
+    /// <summary>
+    /// Overload of <see cref="RaceWithSignalAndTimer{T}(Observable{T}, TimeSpan, Func{Exception}, CancellationToken)"/>
+    /// using the default cause factory (<see cref="OperationCanceledException"/>/<see cref="TimeoutException"/>).
+    /// </summary>
+    /// <typeparam name="T">The type of values produced by <paramref name="source"/>.</typeparam>
+    /// <param name="source">The source sequence to race.</param>
+    /// <param name="timeout">The overall duration before giving up with a timeout error.</param>
+    /// <param name="cancellationToken">A token that, when cancelled, aborts the wait immediately.</param>
+    /// <returns>An observable that mirrors <paramref name="source"/> unless the timeout or cancellation fires first.</returns>
+    public static Observable<T> RaceWithSignalAndTimer<T>(this Observable<T> source, TimeSpan timeout, CancellationToken cancellationToken)
+        => source.RaceWithSignalAndTimer(timeout, causeFactory: null, cancellationToken);
+}

src/RxSharp/Extras/RetryAndRaceWithSignalAndTimer.cs

@@ -3,7 +3,7 @@
 namespace RxSharp.Extras;
 
 /// <summary>Extension methods composing retry, cancellation, and timeout into the single combinator Puppeteer's <c>Locator</c> actions rely on.</summary>
-public static class RetryAndRaceWithSignalAndTimerExtras
+public static partial class Extensions
 {
     /// <summary>
     /// The combinator behind Puppeteer's <c>Locator</c> 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 <paramref name="timeout"/> — whichever happens first. Since the cancellation
     /// and timeout branches are <see cref="Observable{T}"/> sources that only ever error (see
-    /// <see cref="CancellationExtras.FromCancellationToken"/> and <see cref="TimeoutExtras.Timeout"/>), the only way
+    /// <see cref="Extensions.FromCancellationToken"/> and <see cref="Extensions.Timeout"/>), the only way
     /// this combinator produces a value is if the retried <paramref name="source"/> itself emits one before either
     /// branch fires.
     /// </summary>
@@ -23,7 +23,7 @@ public static class RetryAndRaceWithSignalAndTimerExtras
     /// <param name="causeFactory">
     /// Produces the exception used for both the cancellation and timeout branches. Defaults to
     /// <see cref="OperationCanceledException"/> for cancellation and <see cref="TimeoutException"/> for the timeout,
-    /// via the defaults of <see cref="CancellationExtras.FromCancellationToken"/> and <see cref="TimeoutExtras.Timeout"/> respectively.
+    /// via the defaults of <see cref="Extensions.FromCancellationToken"/> and <see cref="Extensions.Timeout"/> respectively.
     /// </param>
     /// <param name="retryDelay">The delay between retry attempts. Defaults to 50 milliseconds.</param>
     /// <param name="cancellationToken">A token that, when cancelled, aborts the whole operation immediately.</param>
@@ -36,9 +36,7 @@ public static Observable<T> RetryAndRaceWithSignalAndTimer<T>(
         CancellationToken cancellationToken)
         => source
             .Retry(delay: retryDelay ?? TimeSpan.FromMilliseconds(50))
-            .RaceWith(
-                CancellationExtras.FromCancellationToken(cancellationToken, causeFactory).Map<Unit, T>(NeverReached<T>),
-                TimeoutExtras.Timeout(timeout, causeFactory).Map<Unit, T>(NeverReached<T>));
+            .RaceWithSignalAndTimer(timeout, causeFactory, cancellationToken);
 
     /// <summary>
     /// Overload of <see cref="RetryAndRaceWithSignalAndTimer{T}(Observable{T}, TimeSpan, Func{Exception}, TimeSpan?, CancellationToken)"/>
@@ -51,9 +49,4 @@ public static Observable<T> RetryAndRaceWithSignalAndTimer<T>(
     /// <returns>An observable that retries <paramref name="source"/> until it succeeds, times out, or is cancelled.</returns>
     public static Observable<T> RetryAndRaceWithSignalAndTimer<T>(this Observable<T> 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<Unit> notifiers type-check inside RaceWith(Observable<T>), matching how rxjs relies
-    // on TypeScript accepting Observable<never> anywhere an Observable<T> is expected. C# has no bottom type.
-    private static TResult NeverReached<TResult>(Unit value) => throw new InvalidOperationException("unreachable: this observable only ever errors");
 }

src/RxSharp/Extras/Timeout.cs

@@ -3,7 +3,7 @@
 namespace RxSharp.Extras;
 
 /// <summary>Extension methods providing a standalone, <see cref="TimeoutException"/>-throwing timer observable.</summary>
-public static class TimeoutExtras
+public static partial class Extensions
 {
     /// <summary>
     /// An observable that never emits and errors after <paramref name="delay"/> elapses, or never errors at all if
@@ -25,9 +25,9 @@ public static Observable<Unit> Timeout(TimeSpan delay, Func<Exception>? causeFac
             return Observable.Never<Unit>();
         }
 
-        var makeCause = causeFactory ?? DefaultCause;
+        var makeCause = causeFactory ?? DefaultTimeoutCause;
         return Observable.Timer(delay, scheduler).Map<long, Unit>(_ => throw makeCause());
     }
 
-    private static Exception DefaultCause() => new TimeoutException();
+    private static Exception DefaultTimeoutCause() => new TimeoutException();
 }

src/RxSharp/RxSharp.csproj

@@ -16,9 +16,9 @@
     <PackageProjectUrl>https://github.com/hardkoded/ReactiveExtensions-Sharp</PackageProjectUrl>
     <RepositoryUrl>https://github.com/hardkoded/ReactiveExtensions-Sharp</RepositoryUrl>
     <PackageReadmeFile>README.md</PackageReadmeFile>
-    <PackageVersion>0.1.1</PackageVersion>
-    <AssemblyVersion>0.1.1.0</AssemblyVersion>
-    <FileVersion>0.1.1.0</FileVersion>
+    <PackageVersion>0.1.2</PackageVersion>
+    <AssemblyVersion>0.1.2.0</AssemblyVersion>
+    <FileVersion>0.1.2.0</FileVersion>
   </PropertyGroup>
 
   <PropertyGroup>

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<Unit>(() => error)
+            .AssumeNeverEmits<string>()
+            .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<string>();
+
+        Observable.Empty<Unit>()
+            .AssumeNeverEmits<string>()
+            .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<string>()
+            .Subscribe(onError: err =>
+            {
+                received = err;
+                signal.Set();
+            });
+
+        Assert.That(signal.Wait(TimeSpan.FromSeconds(2)), Is.True);
+        Assert.That(received, Is.InstanceOf<InvalidOperationException>());
+    }
+}