Migrate from the Node to the Web ReadableStream

dlarocque · dlarocque · commit 7231defd196a · 2024-08-06T17:16:00.000-04:00
diff --git a/.changeset/itchy-boxes-try.md b/.changeset/itchy-boxes-try.md
@@ -0,0 +1,5 @@
+---
+'@firebase/storage': minor
+---
+
+Migrate from the Node to Web ReadableStream interface
diff --git a/common/api-review/storage.api.md b/common/api-review/storage.api.md
@@ -13,6 +13,7 @@ import { FirebaseError } from '@firebase/util';
 import { _FirebaseService } from '@firebase/app';
 import { NextFn } from '@firebase/util';
 import { Provider } from '@firebase/component';
+import { ReadableStream as ReadableStream_2 } from 'stream/web';
 import { Subscribe } from '@firebase/util';
 import { Unsubscribe } from '@firebase/util';
 
@@ -132,7 +133,7 @@ export function getMetadata(ref: StorageReference): Promise<FullMetadata>;
 export function getStorage(app?: FirebaseApp, bucketUrl?: string): FirebaseStorage;
 
 // @public
-export function getStream(ref: StorageReference, maxDownloadSizeBytes?: number): NodeJS.ReadableStream;
+export function getStream(ref: StorageReference, maxDownloadSizeBytes?: number): ReadableStream_2;
 
 // @internal (undocumented)
 export function _invalidArgument(message: string): StorageError;
diff --git a/docs-devsite/storage.md b/docs-devsite/storage.md
@@ -279,7 +279,7 @@ This API is only available in Node.
 <b>Signature:</b>
 
 ```typescript
-export declare function getStream(ref: StorageReference, maxDownloadSizeBytes?: number): NodeJS.ReadableStream;
+export declare function getStream(ref: StorageReference, maxDownloadSizeBytes?: number): ReadableStream;
 ```
 
 #### Parameters
@@ -291,7 +291,7 @@ export declare function getStream(ref: StorageReference, maxDownloadSizeBytes?:
 
 <b>Returns:</b>
 
-NodeJS.ReadableStream
+ReadableStream
 
 A stream with the object's data as bytes
 
diff --git a/packages/storage/src/api.browser.ts b/packages/storage/src/api.browser.ts
@@ -18,6 +18,7 @@
 import { StorageReference } from './public-types';
 import { Reference, getBlobInternal } from './reference';
 import { getModularInstance } from '@firebase/util';
+import { ReadableStream } from 'stream/web';
 
 /**
  * Downloads the data at the object's location. Returns an error if the object
@@ -58,6 +59,6 @@ export function getBlob(
 export function getStream(
   ref: StorageReference,
   maxDownloadSizeBytes?: number
-): NodeJS.ReadableStream {
+): ReadableStream {
   throw new Error('getStream() is only supported by NodeJS builds');
 }
diff --git a/packages/storage/src/api.node.ts b/packages/storage/src/api.node.ts
@@ -18,6 +18,7 @@
 import { StorageReference } from './public-types';
 import { Reference, getStreamInternal } from './reference';
 import { getModularInstance } from '@firebase/util';
+import { ReadableStream } from 'stream/web';
 
 /**
  * Downloads the data at the object's location. Returns an error if the object
@@ -58,7 +59,7 @@ export function getBlob(
 export function getStream(
   ref: StorageReference,
   maxDownloadSizeBytes?: number
-): NodeJS.ReadableStream {
+): ReadableStream {
   ref = getModularInstance(ref);
   return getStreamInternal(ref as Reference, maxDownloadSizeBytes);
 }
diff --git a/packages/storage/src/implementation/connection.ts b/packages/storage/src/implementation/connection.ts
@@ -14,16 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+import { ReadableStream } from 'stream/web';
 
 /** Network headers */
 export type Headers = Record<string, string>;
 
 /** Response type exposed by the networking APIs. */
-export type ConnectionType =
-  | string
-  | ArrayBuffer
-  | Blob
-  | NodeJS.ReadableStream;
+export type ConnectionType = string | ArrayBuffer | Blob | ReadableStream;
 
 /**
  * A lightweight wrapper around XMLHttpRequest with a
diff --git a/packages/storage/src/platform/browser/connection.ts b/packages/storage/src/platform/browser/connection.ts
@@ -22,6 +22,7 @@ import {
   Headers
 } from '../../implementation/connection';
 import { internalError } from '../../implementation/error';
+import { ReadableStream } from 'stream/web';
 
 /** An override for the text-based Connection. Used in tests. */
 let textFactoryOverride: (() => Connection<string>) | null = null;
@@ -171,7 +172,7 @@ export function newBlobConnection(): Connection<Blob> {
   return new XhrBlobConnection();
 }
 
-export function newStreamConnection(): Connection<NodeJS.ReadableStream> {
+export function newStreamConnection(): Connection<ReadableStream> {
   throw new Error('Streams are only supported on Node');
 }
 
diff --git a/packages/storage/src/platform/connection.ts b/packages/storage/src/platform/connection.ts
@@ -22,6 +22,7 @@ import {
   newStreamConnection as nodeNewStreamConnection,
   injectTestConnection as nodeInjectTestConnection
 } from './node/connection';
+import { ReadableStream } from 'stream/web';
 
 export function injectTestConnection(
   factory: (() => Connection<string>) | null
@@ -45,7 +46,7 @@ export function newBlobConnection(): Connection<Blob> {
   return nodeNewBlobConnection();
 }
 
-export function newStreamConnection(): Connection<NodeJS.ReadableStream> {
+export function newStreamConnection(): Connection<ReadableStream> {
   // This file is only used in Node.js tests using ts-node.
   return nodeNewStreamConnection();
 }
diff --git a/packages/storage/src/platform/node/connection.ts b/packages/storage/src/platform/node/connection.ts
@@ -22,6 +22,7 @@ import {
 } from '../../implementation/connection';
 import { internalError } from '../../implementation/error';
 import { fetch as undiciFetch, Headers as undiciHeaders } from 'undici';
+import { ReadableStream } from 'stream/web';
 
 /** An override for the text-based Connection. Used in tests. */
 let textFactoryOverride: (() => Connection<string>) | null = null;
@@ -50,7 +51,7 @@ abstract class FetchConnection<T extends ConnectionType>
   async send(
     url: string,
     method: string,
-    body?: ArrayBufferView | Blob | string,
+    body?: NodeJS.ArrayBufferView | Blob | string,
     headers?: Record<string, string>
   ): Promise<void> {
     if (this.sent_) {
@@ -62,7 +63,7 @@ abstract class FetchConnection<T extends ConnectionType>
       const response = await this.fetch_(url, {
         method,
         headers: headers || {},
-        body: body as ArrayBufferView | string
+        body: body as NodeJS.ArrayBufferView | string
       });
       this.headers_ = response.headers;
       this.statusCode_ = response.status;
@@ -146,13 +147,13 @@ export function newBytesConnection(): Connection<ArrayBuffer> {
   return new FetchBytesConnection();
 }
 
-export class FetchStreamConnection extends FetchConnection<NodeJS.ReadableStream> {
-  private stream_: NodeJS.ReadableStream | null = null;
+export class FetchStreamConnection extends FetchConnection<ReadableStream> {
+  private stream_: ReadableStream | null = null;
 
   async send(
     url: string,
     method: string,
-    body?: ArrayBufferView | Blob | string,
+    body?: NodeJS.ArrayBufferView | Blob | string,
     headers?: Record<string, string>
   ): Promise<void> {
     if (this.sent_) {
@@ -164,7 +165,7 @@ export class FetchStreamConnection extends FetchConnection<NodeJS.ReadableStream
       const response = await this.fetch_(url, {
         method,
         headers: headers || {},
-        body: body as ArrayBufferView | string
+        body: body as NodeJS.ArrayBufferView | string
       });
       this.headers_ = response.headers;
       this.statusCode_ = response.status;
@@ -178,15 +179,15 @@ export class FetchStreamConnection extends FetchConnection<NodeJS.ReadableStream
     }
   }
 
-  getResponse(): NodeJS.ReadableStream {
+  getResponse(): ReadableStream {
     if (!this.stream_) {
       throw internalError('cannot .getResponse() before sending');
     }
     return this.stream_;
   }
 }
 
-export function newStreamConnection(): Connection<NodeJS.ReadableStream> {
+export function newStreamConnection(): Connection<ReadableStream> {
   return new FetchStreamConnection();
 }
 
diff --git a/packages/storage/src/reference.ts b/packages/storage/src/reference.ts
@@ -19,7 +19,7 @@
  * @fileoverview Defines the Firebase StorageReference class.
  */
 
-import { PassThrough, Transform, TransformOptions } from 'stream';
+import { ReadableStream, TransformStream, Transformer } from 'stream/web';
 
 import { FbsBlob } from './implementation/blob';
 import { Location } from './implementation/location';
@@ -48,6 +48,7 @@ import {
   newStreamConnection,
   newTextConnection
 } from './platform/connection';
+import { RequestInfo } from './implementation/requestinfo';
 
 /**
  * Provides methods to interact with a bucket in the Firebase Storage service.
@@ -203,42 +204,42 @@ export function getBlobInternal(
 export function getStreamInternal(
   ref: Reference,
   maxDownloadSizeBytes?: number
-): NodeJS.ReadableStream {
+): ReadableStream {
   ref._throwIfRoot('getStream');
-  const requestInfo = getBytes(
+  const requestInfo: RequestInfo<ReadableStream, ReadableStream> = getBytes(
     ref.storage,
     ref._location,
     maxDownloadSizeBytes
   );
 
-  /** A transformer that passes through the first n bytes. */
-  const newMaxSizeTransform: (n: number) => TransformOptions = n => {
+  // Transforms the stream so that only `maxDownloadSizeBytes` bytes are piped to the result
+  const newMaxSizeTransform = (n: number): Transformer => {
     let missingBytes = n;
     return {
-      transform(chunk, encoding, callback) {
+      transform(chunk, controller: TransformStreamDefaultController) {
         // GCS may not honor the Range header for small files
         if (chunk.length < missingBytes) {
-          this.push(chunk);
+          controller.enqueue(chunk);
           missingBytes -= chunk.length;
         } else {
-          this.push(chunk.slice(0, missingBytes));
-          this.emit('end');
+          controller.enqueue(chunk.slice(0, missingBytes));
+          controller.terminate();
         }
-        callback();
       }
-    } as TransformOptions;
+    };
   };
 
   const result =
     maxDownloadSizeBytes !== undefined
-      ? new Transform(newMaxSizeTransform(maxDownloadSizeBytes))
-      : new PassThrough();
+      ? new TransformStream(newMaxSizeTransform(maxDownloadSizeBytes))
+      : new TransformStream(); // The default transformer forwards all chunks to its readable side
 
   ref.storage
     .makeRequestWithTokens(requestInfo, newStreamConnection)
-    .then(stream => (stream as NodeJS.ReadableStream).pipe(result))
-    .catch(e => result.destroy(e));
-  return result;
+    .then(readableStream => readableStream.pipeThrough(result))
+    .catch(err => result.writable.getWriter().abort(err));
+
+  return result.readable;
 }
 
 /**
diff --git a/packages/storage/test/node/stream.test.ts b/packages/storage/test/node/stream.test.ts
@@ -20,13 +20,29 @@ import { createApp, createStorage } from '../integration/integration.test';
 import { FirebaseApp, deleteApp } from '@firebase/app';
 import { getStream, ref, uploadBytes } from '../../src/index.node';
 import * as types from '../../src/public-types';
+import { ReadableStream } from 'stream/web';
 
-async function readData(reader: NodeJS.ReadableStream): Promise<number[]> {
-  return new Promise<number[]>((resolve, reject) => {
-    const data: number[] = [];
-    reader.on('error', e => reject(e));
-    reader.on('data', chunk => data.push(...Array.from(chunk as Buffer)));
-    reader.on('end', () => resolve(data));
+// See: https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream/getReader
+async function readData(readableStream: ReadableStream): Promise<Uint8Array> {
+  return new Promise<Uint8Array>((resolve, reject) => {
+    const reader: ReadableStreamDefaultReader = readableStream.getReader();
+    const result: any[] = [];
+    reader
+      .read()
+      .then(function processBytes({ done, value }): any {
+        if (done) {
+          resolve(new Uint8Array(result));
+          return;
+        }
+
+        result.push(...value);
+        return reader.read().then(processBytes);
+      })
+      .catch(err => {
+        console.error(err);
+        reject(err);
+        return;
+      });
   });
 }
 
@@ -46,19 +62,17 @@ describe('Firebase Storage > getStream', () => {
   it('can get stream', async () => {
     const reference = ref(storage, 'public/exp-bytes');
     await uploadBytes(reference, new Uint8Array([0, 1, 3, 128, 255]));
-    const stream = await getStream(reference);
+    const stream = getStream(reference);
     const data = await readData(stream);
-    expect(new Uint8Array(data)).to.deep.equal(
-      new Uint8Array([0, 1, 3, 128, 255])
-    );
+    expect(data).to.deep.equal(new Uint8Array([0, 1, 3, 128, 255]));
   });
 
   it('can get first n bytes of stream', async () => {
     const reference = ref(storage, 'public/exp-bytes');
     await uploadBytes(reference, new Uint8Array([0, 1, 3]));
-    const stream = await getStream(reference, 2);
+    const stream = getStream(reference, 2);
     const data = await readData(stream);
-    expect(new Uint8Array(data)).to.deep.equal(new Uint8Array([0, 1]));
+    expect(data).to.deep.equal(new Uint8Array([0, 1]));
   });
 
   it('getStream() throws for missing file', async () => {

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +---
 +'@firebase/storage': minor
 +---
++
 +Migrate from the Node to Web ReadableStream interface
Original file line number	Diff line number	Diff line change
`@@ -22,6 +22,7 @@ import {`
`22`	`22`	`Headers`
`23`	`23`	`} from '../../implementation/connection';`
`24`	`24`	`import { internalError } from '../../implementation/error';`
	`25`	`+import { ReadableStream } from 'stream/web';`
`25`	`26`
`26`	`27`	`/** An override for the text-based Connection. Used in tests. */`
`27`	`28`	`let textFactoryOverride: (() => Connection<string>) \| null = null;`
`@@ -171,7 +172,7 @@ export function newBlobConnection(): Connection<Blob> {`
`171`	`172`	`return new XhrBlobConnection();`
`172`	`173`	`}`
`173`	`174`
`174`		`-export function newStreamConnection(): Connection<NodeJS.ReadableStream> {`
	`175`	`+export function newStreamConnection(): Connection<ReadableStream> {`
`175`	`176`	`throw new Error('Streams are only supported on Node');`
`176`	`177`	`}`
`177`	`178`