Skip to content

Add AssumeNeverEmits, RaceWithSignalAndTimer, and FromEventBuffered - #1

Merged
kblok merged 3 commits into
mainfrom
add-rx-primitives
Jul 27, 2026
Merged

Add AssumeNeverEmits, RaceWithSignalAndTimer, and FromEventBuffered#1
kblok merged 3 commits into
mainfrom
add-rx-primitives

Conversation

@kblok

@kblok kblok commented Jul 27, 2026

Copy link
Copy Markdown
Member

Motivated by puppeteer-sharp's WaitForTargetAsync/WaitForFrameAsync — all three promote patterns that already existed as private, duplicated workarounds in consumer code.

AssumeNeverEmits — promotes the NeverReached helper RetryAndRaceWithSignalAndTimer already had privately: widens an error-only Observable<Unit> to any T, since C# has no bottom type for RaceWith to accept the way TypeScript's Observable<never> does. RetryAndRaceWithSignalAndTimer now uses the public version instead of its own copy.

RaceWithSignalAndTimer — the non-retrying half of RetryAndRaceWithSignalAndTimer, for a single wait (e.g. "wait for the next matching event") that doesn't need retrying. RetryAndRaceWithSignalAndTimer is now built on top of it (Retry().RaceWithSignalAndTimer()).

FromEventBuffered — an eagerly-attaching, buffered variant of FromEvent. Ordinary FromEvent is cold — nothing attaches to the underlying event until Subscribe is called. That's fine in upstream rxjs, where single-threaded run-to-completion semantics mean nothing can fire between attaching a handler and checking some existing state for an already-matching item. .NET has no such guarantee (event delivery can run on another thread), so that gap is a real, provable race — FromEventBuffered attaches immediately and buffers into a ReplaySubject so nothing fired in that gap is lost. (This closes a real bug found while porting puppeteer-sharp's WaitForTargetAsync: a plain Subject bridge dropped a matching TargetCreated event that fired in exactly that gap, hanging every page creation until timeout.)

Also: consolidated every Extras class (the three above plus the pre-existing CancellationExtras/TimeoutExtras/FilterAsyncExtras/RetryAndRaceWithSignalAndTimerExtras) into a single public static partial class Extensions (in the existing RxSharp.Extras namespace), split across the same per-concern files via partial. Extension methods (FilterAsync, RaceWithSignalAndTimer, ...) never needed the class name typed out — using RxSharp.Extras; already surfaces them via dot-completion — but the three factory-style methods (no natural this Observable<T> receiver) gave callers no way to guess which class held which method. One name fixes that for zero cost to the extension methods.

All 839 tests pass (825 existing + 14 new).

🤖 Generated with Claude Code

https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

kblok and others added 3 commits July 27, 2026 15:30
Three additions motivated by puppeteer-sharp's WaitForTargetAsync/WaitForFrameAsync,
promoting patterns that already existed as private, duplicated workarounds:

- AssumeNeverEmits: promotes the NeverReached helper RetryAndRaceWithSignalAndTimer
  already had privately (widening an error-only Observable<Unit> to any T, since C#
  has no bottom type for RaceWith to accept the way TypeScript's Observable<never>
  does). RetryAndRaceWithSignalAndTimer now uses the public version instead of its own
  copy.

- RaceWithSignalAndTimer: the non-retrying half of RetryAndRaceWithSignalAndTimer, for
  a single wait (e.g. "wait for the next matching event") that doesn't need retrying.
  RetryAndRaceWithSignalAndTimer is now built on top of it (Retry().RaceWithSignalAndTimer()).

- FromEventBuffered: an eagerly-attaching, buffered variant of FromEvent. Ordinary
  FromEvent is cold - nothing attaches to the underlying event until Subscribe is
  called. That's fine in upstream rxjs, where single-threaded run-to-completion
  semantics mean nothing can fire between attaching a handler and checking some
  existing state for an already-matching item. .NET has no such guarantee (event
  delivery can run on another thread), so that gap is a real, provable race -
  FromEventBuffered attaches immediately and buffers into a ReplaySubject so nothing
  fired in that gap is lost.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s
Every Extras method previously lived in its own single-method static class
(CancellationExtras, TimeoutExtras, FilterAsyncExtras, ...), each named after its one
method - fine for the extension methods (FilterAsync, RaceWithSignalAndTimer, etc.),
which already show up via dot-completion on Observable<T> with just `using
RxSharp.Extras;`, but a real discoverability tax on the three factory-style methods
(FromCancellationToken, Timeout, FromEventBuffered) that have no natural `this
Observable<T>` receiver to hang off of: a caller had no way to guess which class held
which method.

All seven are now one `public static partial class PuppeteerExtras`, split across the
same per-concern files as before via `partial`. (Named PuppeteerExtras rather than
plain Extras: CA1724 flags a type named the same as its containing namespace, and
every file's own doc comments already frame these as puppeteer-sharp's real
motivation.) Two colliding private `DefaultCause` helpers (one per branch's default
exception factory) got distinct names now that they're members of the same type.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s
PuppeteerExtras tied the name too closely to one consumer for what's meant to be a
general-purpose combinator surface. Extensions in the existing RxSharp.Extras
namespace reads the same way as e.g. Observable in the RxSharp namespace.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s
@kblok
kblok merged commit 3d81516 into main Jul 27, 2026
1 check passed
kblok added a commit to hardkoded/puppeteer-sharp that referenced this pull request Jul 27, 2026
Bumps to RxSharp 0.1.2 (pending publish - see hardkoded/ReactiveExtensions-Sharp#1),
which adds three primitives that let these two methods read much closer to upstream's
merge(...).pipe(filterAsync(predicate), raceWith(...)) shape instead of the
ReplaySubject + manual predicate-in-the-handler version from the previous commit:

- FromEventBuffered eagerly attaches the raw event handler (instead of a plain
  Subject fed by a raw handler) and exposes it as a proper Observable, so predicate
  filtering can happen declaratively downstream via .Filter(predicate) instead of
  inside the handler.
- RaceWithSignalAndTimer replaces the hand-built cancellation/timeout race branches
  in WaitForTargetAsync.
- The private NeverReached helper duplicated in both Browser.cs and CdpPage.cs is
  gone, replaced by RxSharp's own AssumeNeverEmits.

Same external behavior as before - same exception types/messages. Verified with 8
repeated reruns of the two tests that first exposed the ReplaySubject race, confirming
FromEventBuffered doesn't reintroduce it.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s
kblok added a commit to hardkoded/puppeteer-sharp that referenced this pull request Jul 27, 2026
RxSharp's FromEventBufferedExtras/TimeoutExtras/CancellationExtras/etc are now one
public static partial class PuppeteerExtras - see hardkoded/ReactiveExtensions-Sharp#1
for why. Mechanical follow-up: this repo's two call sites.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s
kblok added a commit to hardkoded/puppeteer-sharp that referenced this pull request Jul 27, 2026
PuppeteerExtras -> Extensions, see hardkoded/ReactiveExtensions-Sharp#1.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s
kblok added a commit to hardkoded/puppeteer-sharp that referenced this pull request Aug 10, 2026
* Replace hand-rolled retry/timeout plumbing in Locator and WaitForNetworkIdle with RxSharp

Upstream Puppeteer builds Locator actions and waitForNetworkIdle on rxjs; we've been
reimplementing the same retry/timeout/cancellation and debounce logic by hand in C#.
Locator.RunWithRetryAsync was a manual retry loop with a linked CancellationTokenSource
and several catch clauses just to tell timeout apart from cancellation apart from "retry
again." WaitForNetworkIdleAsync hand-rolled the same debounce-on-events pattern with a
System.Timers.Timer.

Both now go through RxSharp (github.com/hardkoded/ReactiveExtensions-Sharp), a faithful
RxJS port built for exactly this kind of swap. Locator uses its
RetryAndRaceWithSignalAndTimer combinator; WaitForNetworkIdleAsync uses a BehaviorSubject
driving DistinctUntilChanged + SwitchMap to express the same "wait for idleTime after the
last change" debounce declaratively. Same external behavior and exception
messages/types as before - all 39 Locator tests and 8 WaitForNetworkIdle tests pass
unchanged, plus the full suite.

RxSharp needed a strong-name-signed release (0.1.1) since PuppeteerSharp signs its own
assembly and referencing an unsigned dependency fails the build under
TreatWarningsAsErrors.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Replace hand-rolled retry/timeout plumbing in WaitForRequest/Response/Frame/Target with RxSharp

Same motivation as the Locator/WaitForNetworkIdle swap: upstream Puppeteer solves this
exact "wait for an event, race it against a timeout and session-close" problem with
firstValueFrom + raceWith on fromEmitterEvent streams, while we were hand-rolling it
per method with TaskCompletionSource + manual event handler removal + WithTimeout.

CdpPage.WaitForRequestAsync/WaitForResponseAsync/WaitForFrameAsync and
Browser.WaitForTargetAsync now go through the same RxSharp combinators. Extracted two
small shared helpers (TimeoutSignal, SessionClosedSignal) in CdpPage since three of the
four methods needed the identical timeout/session-closed race branches - WaitForNetworkIdleAsync
now reuses them too instead of duplicating the same construction.

Found and fixed a real bug along the way: WaitForTargetAsync and WaitForFrameAsync
originally used a plain Subject to bridge the raw event handlers into the Rx pipeline.
A plain Subject has no buffer, so if the matching event fired between attaching the
handler and this method actually subscribing via FirstValueFrom() (a real window, since
CDP events arrive on their own thread), the emission was silently dropped and the call
would hang until timeout - reproduced this against every page creation
(Browser.WaitForTargetAsync is used internally by CreatePageInContextAsync). Fixed by
using a 1-buffered ReplaySubject instead, which replays the value to the late subscriber.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Rewrite WaitForTargetAsync/WaitForFrameAsync closer to upstream's shape

Bumps to RxSharp 0.1.2 (pending publish - see hardkoded/ReactiveExtensions-Sharp#1),
which adds three primitives that let these two methods read much closer to upstream's
merge(...).pipe(filterAsync(predicate), raceWith(...)) shape instead of the
ReplaySubject + manual predicate-in-the-handler version from the previous commit:

- FromEventBuffered eagerly attaches the raw event handler (instead of a plain
  Subject fed by a raw handler) and exposes it as a proper Observable, so predicate
  filtering can happen declaratively downstream via .Filter(predicate) instead of
  inside the handler.
- RaceWithSignalAndTimer replaces the hand-built cancellation/timeout race branches
  in WaitForTargetAsync.
- The private NeverReached helper duplicated in both Browser.cs and CdpPage.cs is
  gone, replaced by RxSharp's own AssumeNeverEmits.

Same external behavior as before - same exception types/messages. Verified with 8
repeated reruns of the two tests that first exposed the ReplaySubject race, confirming
FromEventBuffered doesn't reintroduce it.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Update call sites for RxSharp's consolidated PuppeteerExtras class

RxSharp's FromEventBufferedExtras/TimeoutExtras/CancellationExtras/etc are now one
public static partial class PuppeteerExtras - see hardkoded/ReactiveExtensions-Sharp#1
for why. Mechanical follow-up: this repo's two call sites.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Update call sites for RxSharp's Extensions class rename

PuppeteerExtras -> Extensions, see hardkoded/ReactiveExtensions-Sharp#1.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Fix a real event-loss race in WaitForTargetAsync/WaitForFrameAsync

Code review caught it: FromEventBuffered's default bufferSize of 1 means that if a
matching event and a later non-matching event both land in the narrow gap between
attaching the handler and the Rx pipeline actually subscribing, the size-1 buffer
keeps only the non-matching one - silently dropping the match. The old
TrySetResult-based implementation didn't have this risk (idempotent, first match
always wins regardless of how many events fire before anyone awaits it).

Fixed by passing an explicit EventBufferSize (16) instead of relying on the default -
enough headroom to safely absorb a realistic burst in that gap without buffering
unboundedly for the whole wait (once subscribed, live delivery is unaffected by
buffer size regardless).

Also bumps to RxSharp 0.1.3 (pending publish - see
hardkoded/ReactiveExtensions-Sharp#2), which renames the consolidated Extras class to
RxExtensions and documents this exact bufferSize risk for future callers.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Apply suggestion from @kblok

* Drop EventBufferSize now that RxSharp defaults to unbounded

RxSharp 0.1.4 (pending publish - see hardkoded/ReactiveExtensions-Sharp#3) makes
FromEventBuffered's bufferSize nullable, defaulting to ReplaySubject's own unbounded
default instead of an opinionated 1. That's the same guarantee EventBufferSize = 16
was working around locally, so it's redundant now - the safe behavior is just the
library default.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Update to RxSharp 0.2.0's ReactiveExtensionsSharp namespace

The library's namespace now matches what we've been installing all along
(ReactiveExtensionsSharp - see hardkoded/ReactiveExtensions-Sharp#4 for why). Purely
mechanical: using RxSharp* -> using ReactiveExtensionsSharp* in the three files that
reference it, plus the package reference bump.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Add a RxSharp using alias for ReactiveExtensionsSharp

Purely a naming aid in the using block itself - `Observable<T>`, `Unit`, and every
Map/Filter/RaceWithSignalAndTimer call stay unqualified exactly as before, since the
plain `using ReactiveExtensionsSharp;` import is still there doing that work. The
alias adds no new qualified references anywhere in the file bodies.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92eapPm8e7mX7puT4cF4s

* Drop AssumeNeverEmits, unify Wait* methods into the shared Page class

RaceWithSignalAndTimer used to fake timeout/cancellation into Observable<T>
branches via AssumeNeverEmits just to sit in an Observable-level RaceWith next
to real data. Racing at the Task level instead (ReactiveExtensionsSharp 0.3.0)
sidesteps the problem entirely, and drops a couple of .FirstValueFrom() calls
that are no longer needed now that the combinator returns Task<T> directly.

Also moved WaitForRequestAsync/WaitForResponseAsync/WaitForFrameAsync/
WaitForNetworkIdleAsync from being duplicated per-protocol in CdpPage/BidiPage
into a single shared implementation on the abstract Page class, matching how
upstream's Page.ts does it. Comparing our port against upstream surfaced a
few real gaps this fixes as a side effect: WaitForOptions.CancellationToken
was silently ignored everywhere; we raced against an internal session-closed
task instead of the public Close event; CDP's WaitForNetworkIdleAsync ignored
the Concurrency option entirely; WaitForFrameAsync under CDP never checked
for an already-matching frame before waiting. In-flight request tracking now
lives for the page's whole lifetime (wired once in the constructor, mirroring
upstream's own #inflight$) instead of being rebuilt fresh on every
WaitForNetworkIdleAsync call, so it correctly reflects requests already in
flight before the call.

* Bump version to 25.5.1

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant