Enable bundle reader for Node.

packages/firestore/src/platform/platform.ts

@@ -50,6 +50,11 @@ export interface Platform {
    */
   randomBytes(nBytes: number): Uint8Array;
 
+  /**
+   * Builds a `ByteStreamReader` from a data source.
+   */
+  toByteStreamReader(source: unknown): ByteStreamReader;
+
   /** The Platform's 'window' implementation or null if not available. */
   readonly window: Window | null;
 
@@ -60,6 +65,54 @@ export interface Platform {
   readonly base64Available: boolean;
 }
 
+/**
+ * An interface compatible with Web's ReadableStream.getReader() return type.
+ *
+ * This can be used as an abstraction to mimic `ReadableStream` where it is not
+ * available.
+ */
+export interface ByteStreamReader {
+  read(): Promise<ByteStreamReadResult>;
+  cancel(reason?: string): Promise<void>;
+}
+
+/**
+ * An interface compatible with ReadableStreamReadResult<UInt8Array>.
+ */
+export interface ByteStreamReadResult {
+  done: boolean;
+  value?: Uint8Array;
+}
+
+/**
+ * Builds a `ByteStreamReader` from a UInt8Array.
+ * @param source The data source to use.
+ * @param bytesPerRead How many bytes each `read()` from the returned reader
+ *        will read.
+ */
+export function toByteStreamReader(
+  source: Uint8Array,
+  bytesPerRead = 10240
+): ByteStreamReader {
+  let readFrom = 0;
+  return new (class implements ByteStreamReader {
+    async read(): Promise<ByteStreamReadResult> {
+      if (readFrom < source.byteLength) {
+        const result = {
+          value: source.slice(readFrom, readFrom + bytesPerRead),
+          done: false
+        };
+        readFrom += bytesPerRead;
+        return result;
+      }
+
+      return { value: undefined, done: true };
+    }
+
+    async cancel(reason?: string): Promise<void> {}
+  })();
+}
+
 /**
  * Provides singleton helpers where setup code can inject a platform at runtime.
  * setPlatform needs to be set before Firestore is used and must be set exactly
@@ -74,6 +127,13 @@ export class PlatformSupport {
     PlatformSupport.platform = platform;
   }
 
+  /**
+   * Forcing to set the platform instance, testing only!
+   */
+  private static _forceSetPlatform(platform: Platform): void {
+    PlatformSupport.platform = platform;
+  }
+
   static getPlatform(): Platform {
     if (!PlatformSupport.platform) {
       fail('Platform not set');

packages/firestore/src/platform_browser/browser_platform.ts

@@ -16,7 +16,12 @@
  */
 
 import { DatabaseId, DatabaseInfo } from '../core/database_info';
-import { Platform } from '../platform/platform';
+import {
+  ByteStreamReader,
+  ByteStreamReadResult,
+  Platform,
+  toByteStreamReader
+} from '../platform/platform';
 import { Connection } from '../remote/connection';
 import { JsonProtoSerializer } from '../remote/serializer';
 import { ConnectivityMonitor } from './../remote/connectivity_monitor';
@@ -93,4 +98,33 @@ export class BrowserPlatform implements Platform {
     }
     return bytes;
   }
+
+  /**
+   * On web, a `ReadableStream` is wrapped around by a `ByteStreamReader`.
+   */
+  toByteStreamReader(
+    source: Uint8Array | ArrayBuffer | ReadableStream<Uint8Array>
+  ): ByteStreamReader {
+    if (source instanceof Uint8Array) {
+      return toByteStreamReader(source);
+    }
+    if (source instanceof ArrayBuffer) {
+      return toByteStreamReader(new Uint8Array(source));
+    }
+    if (source instanceof ReadableStream) {
+      const reader = source.getReader();
+      return new (class implements ByteStreamReader {
+        read(): Promise<ByteStreamReadResult> {
+          return reader.read();
+        }
+
+        cancel(reason?: string): Promise<void> {
+          return reader.cancel(reason);
+        }
+      })();
+    }
+    throw new Error(
+      'Source of `toByteStreamReader` has to be Uint8Array, ArrayBuffer or ReadableStream'
+    );
+  }
 }

packages/firestore/src/platform_node/node_platform.ts

@@ -19,7 +19,11 @@ import { randomBytes } from 'crypto';
 import { inspect } from 'util';
 
 import { DatabaseId, DatabaseInfo } from '../core/database_info';
-import { Platform } from '../platform/platform';
+import {
+  ByteStreamReader,
+  Platform,
+  toByteStreamReader
+} from '../platform/platform';
 import { Connection } from '../remote/connection';
 import { JsonProtoSerializer } from '../remote/serializer';
 import { Code, FirestoreError } from '../util/error';
@@ -83,4 +87,14 @@ export class NodePlatform implements Platform {
 
     return randomBytes(nBytes);
   }
+
+  /**
+   * On Node, only supported data source is a `Uint8Array` for now.
+   */
+  toByteStreamReader(source: unknown): ByteStreamReader {
+    if (source instanceof Uint8Array) {
+      return toByteStreamReader(source);
+    }
+    throw new Error('Source of `toByteStreamReader` has to be Uint8Array');
+  }
 }

packages/firestore/src/util/bundle_reader.ts

@@ -20,6 +20,8 @@ import {
   BundleMetadata
 } from '../protos/firestore_bundle_proto';
 import { Deferred } from './promise';
+import { ByteStreamReader, PlatformSupport } from '../platform/platform';
+import { debugAssert } from './assert';
 
 /**
  * A complete element in the bundle stream, together with the byte length it
@@ -37,30 +39,6 @@ export class SizedBundleElement {
   }
 }
 
-/**
- * Create a `ReadableStream` from a underlying buffer.
- *
- * @param data: Underlying buffer.
- * @param bytesPerRead: How many bytes to read from the underlying buffer from
- * each read through the stream.
- */
-export function toReadableStream(
-  data: Uint8Array | ArrayBuffer,
-  bytesPerRead = 10240
-): ReadableStream<Uint8Array | ArrayBuffer> {
-  let readFrom = 0;
-  return new ReadableStream({
-    start(controller) {},
-    async pull(controller): Promise<void> {
-      controller.enqueue(data.slice(readFrom, readFrom + bytesPerRead));
-      readFrom += bytesPerRead;
-      if (readFrom >= data.byteLength) {
-        controller.close();
-      }
-    }
-  });
-}
-
 /**
  * A class representing a bundle.
  *
@@ -70,8 +48,8 @@ export function toReadableStream(
 export class BundleReader {
   /** Cached bundle metadata. */
   private metadata: Deferred<BundleMetadata> = new Deferred<BundleMetadata>();
-  /** The reader instance of the given ReadableStream. */
-  private reader: ReadableStreamDefaultReader;
+  /** The reader to read from underlying binary bundle data source. */
+  private reader: ByteStreamReader;
   /**
    * Internal buffer to hold bundle content, accumulating incomplete element
    * content.
@@ -86,13 +64,9 @@ export class BundleReader {
       | Uint8Array
       | ArrayBuffer
   ) {
-    if (
-      bundleStream instanceof Uint8Array ||
-      bundleStream instanceof ArrayBuffer
-    ) {
-      this.bundleStream = toReadableStream(bundleStream);
-    }
-    this.reader = (this.bundleStream as ReadableStream).getReader();
+    this.reader = PlatformSupport.getPlatform().toByteStreamReader(
+      bundleStream
+    );
 
     // Read the metadata (which is the first element).
     this.nextElementImpl().then(
@@ -231,11 +205,12 @@ export class BundleReader {
   private async pullMoreDataToBuffer(): Promise<boolean> {
     const result = await this.reader.read();
     if (!result.done) {
+      debugAssert(!!result.value, 'Read undefined when "done" is false.');
       const newBuffer = new Uint8Array(
-        this.buffer.length + result.value.length
+        this.buffer.length + result.value!.length
       );
       newBuffer.set(this.buffer);
-      newBuffer.set(result.value, this.buffer.length);
+      newBuffer.set(result.value!, this.buffer.length);
       this.buffer = newBuffer;
     }
     return result.done;

packages/firestore/test/unit/platform/platform.test.ts

@@ -16,10 +16,34 @@
  */
 
 import { expect } from 'chai';
-import { PlatformSupport } from '../../../src/platform/platform';
+import {
+  PlatformSupport,
+  toByteStreamReader
+} from '../../../src/platform/platform';
 
 describe('Platform', () => {
   it('can load the platform at runtime', () => {
     expect(PlatformSupport.getPlatform()).to.exist;
   });
+
+  it('toByteStreamReader() steps underlying data', async () => {
+    const encoder = new TextEncoder();
+    const r = toByteStreamReader(encoder.encode('0123456789'), 4);
+
+    let result = await r.read();
+    expect(result.value).to.deep.equal(encoder.encode('0123'));
+    expect(result.done).to.be.false;
+
+    result = await r.read();
+    expect(result.value).to.deep.equal(encoder.encode('4567'));
+    expect(result.done).to.be.false;
+
+    result = await r.read();
+    expect(result.value).to.deep.equal(encoder.encode('89'));
+    expect(result.done).to.be.false;
+
+    result = await r.read();
+    expect(result.value).to.be.undefined;
+    expect(result.done).to.be.true;
+  });
 });