refactor: implement fromEmitterEvent (#11658)

packages/puppeteer-core/src/api/Browser.ts

@@ -9,18 +9,15 @@ import type {ChildProcess} from 'child_process';
 import type {Protocol} from 'devtools-protocol';
 
 import {
+  filterAsync,
   firstValueFrom,
   from,
   merge,
   raceWith,
-  filterAsync,
-  fromEvent,
-  type Observable,
 } from '../../third_party/rxjs/rxjs.js';
 import type {ProtocolType} from '../common/ConnectOptions.js';
 import {EventEmitter, type EventType} from '../common/EventEmitter.js';
-import {debugError} from '../common/util.js';
+import {debugError, fromEmitterEvent, timeout} from '../common/util.js';
-import {timeout} from '../common/util.js';
 import {asyncDisposeSymbol, disposeSymbol} from '../util/disposable.js';
 
 import type {BrowserContext} from './BrowserContext.js';
@@ -343,8 +340,8 @@ export abstract class Browser extends EventEmitter<BrowserEvents> {
     const {timeout: ms = 30000} = options;
     return await firstValueFrom(
       merge(
-        fromEvent(this, BrowserEvent.TargetCreated) as Observable<Target>,
+        fromEmitterEvent(this, BrowserEvent.TargetCreated),
-        fromEvent(this, BrowserEvent.TargetChanged) as Observable<Target>,
+        fromEmitterEvent(this, BrowserEvent.TargetChanged),
         from(this.targets())
       ).pipe(filterAsync(predicate), raceWith(timeout(ms)))
     );

packages/puppeteer-core/src/api/Page.ts

@@ -15,7 +15,6 @@ import {
   first,
   firstValueFrom,
   from,
-  fromEvent,
   map,
   merge,
   of,
@@ -58,6 +57,7 @@ import type {
 } from '../common/types.js';
 import {
   debugError,
+  fromEmitterEvent,
   importFSPromises,
   isString,
   timeout,
@@ -1677,25 +1677,16 @@ export abstract class Page extends EventEmitter<PageEvents> {
     requestsInFlight = 0
   ): Observable<void> {
     return merge(
-      fromEvent(
+      fromEmitterEvent(networkManager, NetworkManagerEvent.Request),
-        networkManager,
+      fromEmitterEvent(networkManager, NetworkManagerEvent.Response),
-        NetworkManagerEvent.Request as unknown as string
+      fromEmitterEvent(networkManager, NetworkManagerEvent.RequestFailed)
-      ) as Observable<void>,
-      fromEvent(
-        networkManager,
-        NetworkManagerEvent.Response as unknown as string
-      ) as Observable<void>,
-      fromEvent(
-        networkManager,
-        NetworkManagerEvent.RequestFailed as unknown as string
-      ) as Observable<void>
     ).pipe(
       startWith(undefined),
       filter(() => {
         return networkManager.inFlightRequestsCount() <= requestsInFlight;
       }),
-      switchMap(v => {
+      switchMap(() => {
-        return of(v).pipe(delay(idleTime));
+        return of(undefined).pipe(delay(idleTime));
       })
     );
   }
@@ -1725,15 +1716,15 @@ export abstract class Page extends EventEmitter<PageEvents> {
 
     return await firstValueFrom(
       merge(
-        fromEvent(this, PageEvent.FrameAttached) as Observable<Frame>,
+        fromEmitterEvent(this, PageEvent.FrameAttached),
-        fromEvent(this, PageEvent.FrameNavigated) as Observable<Frame>,
+        fromEmitterEvent(this, PageEvent.FrameNavigated),
         from(this.frames())
       ).pipe(
         filterAsync(urlOrPredicate),
         first(),
         raceWith(
           timeout(ms),
-          fromEvent(this, PageEvent.Close).pipe(
+          fromEmitterEvent(this, PageEvent.Close).pipe(
             map(() => {
               throw new TargetCloseError('Page closed.');
             })

packages/puppeteer-core/src/api/locators/locators.ts

@@ -8,27 +8,27 @@ import type {
   OperatorFunction,
 } from '../../../third_party/rxjs/rxjs.js';
 import {
-  mergeMap,
-  from,
   EMPTY,
+  catchError,
+  defaultIfEmpty,
   defer,
   filter,
   first,
+  firstValueFrom,
+  from,
+  fromEvent,
   identity,
   ignoreElements,
-  retry,
-  throwIfEmpty,
-  race,
-  catchError,
-  defaultIfEmpty,
-  firstValueFrom,
-  fromEvent,
   map,
   merge,
+  mergeMap,
   noop,
   pipe,
+  race,
   raceWith,
+  retry,
   tap,
+  throwIfEmpty,
 } from '../../../third_party/rxjs/rxjs.js';
 import type {EventType} from '../../common/EventEmitter.js';
 import {EventEmitter} from '../../common/EventEmitter.js';

packages/puppeteer-core/src/bidi/Frame.ts

@@ -7,29 +7,27 @@
 import * as Bidi from 'chromium-bidi/lib/cjs/protocol/protocol.js';
 
 import {
-  type Observable,
-  from,
-  fromEvent,
-  merge,
-  map,
-  forkJoin,
   first,
   firstValueFrom,
+  forkJoin,
+  from,
+  map,
+  merge,
   raceWith,
 } from '../../third_party/rxjs/rxjs.js';
 import type {CDPSession} from '../api/CDPSession.js';
 import type {ElementHandle} from '../api/ElementHandle.js';
 import {
   Frame,
+  throwIfDetached,
   type GoToOptions,
   type WaitForOptions,
-  throwIfDetached,
 } from '../api/Frame.js';
 import type {WaitForSelectorOptions} from '../api/Page.js';
 import {UnsupportedOperation} from '../common/Errors.js';
 import type {TimeoutSettings} from '../common/TimeoutSettings.js';
 import type {Awaitable, NodeFor} from '../common/types.js';
-import {UTILITY_WORLD_NAME, timeout} from '../common/util.js';
+import {fromEmitterEvent, timeout, UTILITY_WORLD_NAME} from '../common/util.js';
 import {Deferred} from '../util/Deferred.js';
 import {disposeSymbol} from '../util/disposable.js';
 
@@ -163,7 +161,7 @@ export class BidiFrame extends Frame {
       this.#page
         ._waitWithNetworkIdle(
           forkJoin([
-            fromEvent(this.#context, waitEvent).pipe(first()),
+            fromEmitterEvent(this.#context, waitEvent).pipe(first()),
             from(this.setFrameContent(html)),
           ]).pipe(
             map(() => {
@@ -194,18 +192,16 @@ export class BidiFrame extends Frame {
 
     const navigatedObservable = merge(
       forkJoin([
-        fromEvent(
+        fromEmitterEvent(
           this.#context,
           Bidi.ChromiumBidi.BrowsingContext.EventNames.NavigationStarted
         ).pipe(first()),
-        fromEvent(this.#context, waitUntilEvent).pipe(
+        fromEmitterEvent(this.#context, waitUntilEvent).pipe(first()),
-          first()
-        ) as Observable<Bidi.BrowsingContext.NavigationInfo>,
       ]),
-      fromEvent(
+      fromEmitterEvent(
         this.#context,
         Bidi.ChromiumBidi.BrowsingContext.EventNames.FragmentNavigated
-      ) as Observable<Bidi.BrowsingContext.NavigationInfo>
+      )
     ).pipe(
       map(result => {
         if (Array.isArray(result)) {

packages/puppeteer-core/src/common/util.ts

@@ -9,16 +9,15 @@ import type {Readable} from 'stream';
 
 import type {Protocol} from 'devtools-protocol';
 
-import type {Observable} from '../../third_party/rxjs/rxjs.js';
 import {
+  filterAsync,
+  firstValueFrom,
+  from,
   map,
   NEVER,
-  timer,
+  Observable,
-  firstValueFrom,
-  fromEvent,
-  filterAsync,
-  from,
   raceWith,
+  timer,
 } from '../../third_party/rxjs/rxjs.js';
 import type {CDPSession} from '../api/CDPSession.js';
 import {isNode} from '../environment.js';
@@ -594,9 +593,7 @@ export async function waitForHTTP<T extends {url(): string}>(
   cancelation: Deferred<never>
 ): Promise<T> {
   return await firstValueFrom(
-    (
+    (fromEmitterEvent(networkManager, eventName) as Observable<T>).pipe(
-      fromEvent(networkManager, eventName as unknown as string) as Observable<T>
-    ).pipe(
       filterAsync(async http => {
         if (isString(urlOrPredicate)) {
           return urlOrPredicate === http.url();
@@ -711,3 +708,21 @@ function convertPrintParameterToInches(
   }
   return pixels / unitToPixels[lengthUnit];
 }
+
+/**
+ * @internal
+ */
+export function fromEmitterEvent<
+  Events extends Record<EventType, unknown>,
+  Event extends keyof Events,
+>(emitter: EventEmitter<Events>, eventName: Event): Observable<Events[Event]> {
+  return new Observable(subscriber => {
+    const listener = (event: Events[Event]) => {
+      subscriber.next(event);
+    };
+    emitter.on(eventName, listener);
+    return () => {
+      emitter.off(eventName, listener);
+    };
+  });
+}

packages/puppeteer-core/src/node/ScreenRecorder.ts

@@ -9,12 +9,8 @@ import {spawn, spawnSync} from 'child_process';
 import {PassThrough} from 'stream';
 
 import debug from 'debug';
-import type Protocol from 'devtools-protocol';
 
-import type {
+import type {OperatorFunction} from '../../third_party/rxjs/rxjs.js';
-  Observable,
-  OperatorFunction,
-} from '../../third_party/rxjs/rxjs.js';
 import {
   bufferCount,
   concatMap,
@@ -29,7 +25,7 @@ import {
 import {CDPSessionEvent} from '../api/CDPSession.js';
 import type {BoundingBox} from '../api/ElementHandle.js';
 import type {Page} from '../api/Page.js';
-import {debugError} from '../common/util.js';
+import {debugError, fromEmitterEvent} from '../common/util.js';
 import {guarded} from '../util/decorators.js';
 import {asyncDisposeSymbol} from '../util/disposable.js';
 
@@ -137,12 +133,7 @@ export class ScreenRecorder extends PassThrough {
     });
 
     this.#lastFrame = lastValueFrom(
-      (
+      fromEmitterEvent(client, 'Page.screencastFrame').pipe(
-        fromEvent(
-          client,
-          'Page.screencastFrame'
-        ) as Observable<Protocol.Page.ScreencastFrameEvent>
-      ).pipe(
         tap(event => {
           void client.send('Page.screencastFrameAck', {
             sessionId: event.sessionId,

packages/puppeteer-core/third_party/rxjs/rxjs.ts

@@ -25,6 +25,7 @@ export {
   mergeMap,
   NEVER,
   noop,
+  Observable,
   of,
   pipe,
   race,