refactor!: use ReadableStreams (#11805)

docs/api/puppeteer.page.createpdfstream.md

@@ -10,7 +10,9 @@ Generates a PDF of the page with the `print` CSS media type.
 
 ```typescript
 class Page {
-  abstract createPDFStream(options?: PDFOptions): Promise<Readable>;
+  abstract createPDFStream(
+    options?: PDFOptions
+  ): Promise<ReadableStream<Uint8Array>>;
 }
 ```
 
@@ -22,7 +24,7 @@ class Page {
 
 **Returns:**
 
-Promise&lt;Readable&gt;
+Promise&lt;ReadableStream&lt;Uint8Array&gt;&gt;
 
 ## Remarks
 

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

@@ -4,8 +4,6 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import type {Readable} from 'stream';
-
 import type {Protocol} from 'devtools-protocol';
 
 import {
@@ -2574,7 +2572,9 @@ export abstract class Page extends EventEmitter<PageEvents> {
    * {@link https://developer.mozilla.org/en-US/docs/Web/CSS/-webkit-print-color-adjust | `-webkit-print-color-adjust`}
    * property to force rendering of exact colors.
    */
-  abstract createPDFStream(options?: PDFOptions): Promise<Readable>;
+  abstract createPDFStream(
+    options?: PDFOptions
+  ): Promise<ReadableStream<Uint8Array>>;
 
   /**
    * {@inheritDoc Page.createPDFStream}

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

@@ -4,8 +4,6 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import type {Readable} from 'stream';
-
 import * as Bidi from 'chromium-bidi/lib/cjs/protocol/protocol.js';
 import type Protocol from 'devtools-protocol';
 
@@ -638,19 +636,15 @@ export class BidiPage extends Page {
 
   override async createPDFStream(
     options?: PDFOptions | undefined
-  ): Promise<Readable> {
+  ): Promise<ReadableStream<Uint8Array>> {
     const buffer = await this.pdf(options);
-    try {
-      const {Readable} = await import('stream');
-      return Readable.from(buffer);
-    } catch (error) {
-      if (error instanceof TypeError) {
-        throw new Error(
-          'Can only pass a file path in a Node-like environment.'
-        );
-      }
-      throw error;
-    }
+
+    return new ReadableStream({
+      start(controller) {
+        controller.enqueue(buffer);
+        controller.close();
+      },
+    });
   }
 
   override async _screenshot(

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

@@ -4,8 +4,6 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import type {Readable} from 'stream';
-
 import type {Protocol} from 'devtools-protocol';
 
 import {firstValueFrom, from, raceWith} from '../../third_party/rxjs/rxjs.js';
@@ -1105,7 +1103,9 @@ export class CdpPage extends Page {
     return data;
   }
 
-  override async createPDFStream(options: PDFOptions = {}): Promise<Readable> {
+  override async createPDFStream(
+    options: PDFOptions = {}
+  ): Promise<ReadableStream<Uint8Array>> {
     const {timeout: ms = this._timeoutSettings.timeout()} = options;
     const {
       landscape,

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

@@ -5,13 +5,10 @@
  */
 
 import type FS from 'fs/promises';
-import type {Readable} from 'stream';
 
 import {map, NEVER, Observable, timer} from '../../third_party/rxjs/rxjs.js';
 import type {CDPSession} from '../api/CDPSession.js';
-import {isNode} from '../environment.js';
 import {assert} from '../util/assert.js';
-import {isErrorLike} from '../util/ErrorLike.js';
 
 import {debug} from './Debug.js';
 import {TimeoutError} from './Errors.js';
@@ -209,69 +206,74 @@ export async function importFSPromises(): Promise<typeof FS> {
  * @internal
  */
 export async function getReadableAsBuffer(
-  readable: Readable,
+  readable: ReadableStream<Uint8Array>,
   path?: string
 ): Promise<Buffer | null> {
   const buffers = [];
+  const reader = readable.getReader();
   if (path) {
     const fs = await importFSPromises();
     const fileHandle = await fs.open(path, 'w+');
     try {
-      for await (const chunk of readable) {
-        buffers.push(chunk);
-        await fileHandle.writeFile(chunk);
+      while (true) {
+        const {done, value} = await reader.read();
+        if (done) {
+          break;
+        }
+        buffers.push(value);
+        await fileHandle.writeFile(value);
       }
     } finally {
       await fileHandle.close();
     }
   } else {
-    for await (const chunk of readable) {
-      buffers.push(chunk);
+    while (true) {
+      const {done, value} = await reader.read();
+      if (done) {
+        break;
+      }
+      buffers.push(value);
     }
   }
   try {
     return Buffer.concat(buffers);
   } catch (error) {
+    debugError(error);
     return null;
   }
 }
 
+/**
+ * @internal
+ */
+
 /**
  * @internal
  */
 export async function getReadableFromProtocolStream(
   client: CDPSession,
   handle: string
-): Promise<Readable> {
-  // TODO: Once Node 18 becomes the lowest supported version, we can migrate to
-  // ReadableStream.
-  if (!isNode) {
-    throw new Error('Cannot create a stream outside of Node.js environment.');
-  }
-
-  const {Readable} = await import('stream');
-
-  let eof = false;
-  return new Readable({
-    async read(size: number) {
-      if (eof) {
-        return;
+): Promise<ReadableStream<Uint8Array>> {
+  return new ReadableStream({
+    async pull(controller) {
+      function getUnit8Array(data: string, isBase64: boolean): Uint8Array {
+        if (isBase64) {
+          return Uint8Array.from(atob(data), m => {
+            return m.codePointAt(0)!;
+          });
+        }
+        const encoder = new TextEncoder();
+        return encoder.encode(data);
       }
 
-      try {
-        const response = await client.send('IO.read', {handle, size});
-        this.push(response.data, response.base64Encoded ? 'base64' : undefined);
-        if (response.eof) {
-          eof = true;
-          await client.send('IO.close', {handle});
-          this.push(null);
-        }
-      } catch (error) {
-        if (isErrorLike(error)) {
-          this.destroy(error);
-          return;
-        }
-        throw error;
+      const {data, base64Encoded, eof} = await client.send('IO.read', {
+        handle,
+      });
+
+      controller.enqueue(getUnit8Array(data, base64Encoded ?? false));
+      if (eof) {
+        await client.send('IO.close', {handle});
+        controller.close();
       }
     },
   });

test/src/page.spec.ts

@@ -1964,9 +1964,15 @@ describe('Page', function () {
 
       const stream = await page.createPDFStream();
       let size = 0;
-      for await (const chunk of stream) {
-        size += chunk.length;
+      const reader = stream.getReader();
+      while (true) {
+        const {done, value} = await reader.read();
+        if (done) {
+          break;
+        }
+        size += value.length;
       }
+
       expect(size).toBeGreaterThan(0);
     });