Implement BundleCache for IDB and memory.

packages/firestore/src/platform/platform.ts

@@ -20,6 +20,8 @@ import { Connection } from '../remote/connection';
 import { JsonProtoSerializer } from '../remote/serializer';
 import { fail } from '../util/assert';
 import { ConnectivityMonitor } from './../remote/connectivity_monitor';
+import { BundleSource } from '../util/bundle_reader';
+import { validatePositiveNumber } from '../util/input_validation';
 
 /**
  * Provides a common interface to load anything platform dependent, e.g.
@@ -52,8 +54,15 @@ export interface Platform {
 
   /**
    * Builds a `ByteStreamReader` from a data source.
+   * @param source The data source to use.
+   * @param bytesPerRead How many bytes each `read()` from the returned reader
+   *        will read. It is ignored if the passed in source does not provide
+   *        such control(example: ReadableStream).
    */
-  toByteStreamReader(source: unknown): ByteStreamReader;
+  toByteStreamReader(
+    source: BundleSource,
+    bytesPerRead: number
+  ): ByteStreamReader;
 
   /** The Platform's 'window' implementation or null if not available. */
   readonly window: Window | null;
@@ -92,10 +101,11 @@ export interface ByteStreamReadResult {
  */
 export function toByteStreamReader(
   source: Uint8Array,
-  bytesPerRead = 10240
+  bytesPerRead: number
 ): ByteStreamReader {
+  validatePositiveNumber('toByteStreamReader', 2, bytesPerRead);
   let readFrom = 0;
-  return new (class implements ByteStreamReader {
+  const reader: ByteStreamReader = {
     async read(): Promise<ByteStreamReadResult> {
       if (readFrom < source.byteLength) {
         const result = {
@@ -107,10 +117,12 @@ export function toByteStreamReader(
       }
 
       return { value: undefined, done: true };
-    }
+    },
 
     async cancel(reason?: string): Promise<void> {}
-  })();
+  };
+
+  return reader;
 }
 
 /**

packages/firestore/src/platform_browser/browser_platform.ts

@@ -18,7 +18,6 @@
 import { DatabaseId, DatabaseInfo } from '../core/database_info';
 import {
   ByteStreamReader,
-  ByteStreamReadResult,
   Platform,
   toByteStreamReader
 } from '../platform/platform';
@@ -30,6 +29,7 @@ import { NoopConnectivityMonitor } from '../remote/connectivity_monitor_noop';
 import { BrowserConnectivityMonitor } from './browser_connectivity_monitor';
 import { WebChannelConnection } from './webchannel_connection';
 import { debugAssert } from '../util/assert';
+import { BundleSource } from '../util/bundle_reader';
 
 // Implements the Platform API for browsers and some browser-like environments
 // (including ReactNative).
@@ -103,28 +103,20 @@ export class BrowserPlatform implements Platform {
    * On web, a `ReadableStream` is wrapped around by a `ByteStreamReader`.
    */
   toByteStreamReader(
-    source: Uint8Array | ArrayBuffer | ReadableStream<Uint8Array>
+    source: BundleSource,
+    bytesPerRead: number
   ): ByteStreamReader {
     if (source instanceof Uint8Array) {
-      return toByteStreamReader(source);
+      return toByteStreamReader(source, bytesPerRead);
     }
     if (source instanceof ArrayBuffer) {
-      return toByteStreamReader(new Uint8Array(source));
+      return toByteStreamReader(new Uint8Array(source), bytesPerRead);
     }
     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);
-        }
-      })();
+      return source.getReader();
     }
     throw new Error(
-      'Source of `toByteStreamReader` has to be Uint8Array, ArrayBuffer or ReadableStream'
+      'Source of `toByteStreamReader` has to be a ArrayBuffer or ReadableStream'
     );
   }
 }

packages/firestore/src/platform_node/node_platform.ts

@@ -33,6 +33,7 @@ import { NoopConnectivityMonitor } from './../remote/connectivity_monitor_noop';
 import { GrpcConnection } from './grpc_connection';
 import { loadProtos } from './load_protos';
 import { debugAssert } from '../util/assert';
+import { invalidClassError } from '../util/input_validation';
 
 export class NodePlatform implements Platform {
   readonly base64Available = true;
@@ -91,10 +92,18 @@ export class NodePlatform implements Platform {
   /**
    * On Node, only supported data source is a `Uint8Array` for now.
    */
-  toByteStreamReader(source: unknown): ByteStreamReader {
-    if (source instanceof Uint8Array) {
-      return toByteStreamReader(source);
+  toByteStreamReader(
+    source: Uint8Array,
+    bytesPerRead: number
+  ): ByteStreamReader {
+    if (!(source instanceof Uint8Array)) {
+      throw invalidClassError(
+        'NodePlatform.toByteStreamReader',
+        'Uint8Array',
+        1,
+        source
+      );
     }
-    throw new Error('Source of `toByteStreamReader` has to be Uint8Array');
+    return toByteStreamReader(source, bytesPerRead);
   }
 }

packages/firestore/src/util/bundle_reader.ts

@@ -39,6 +39,20 @@ export class SizedBundleElement {
   }
 }
 
+export type BundleSource =
+  | ReadableStream<Uint8Array>
+  | ArrayBuffer
+  | Uint8Array;
+
+/**
+ * When applicable, how many bytets to read from the underlying data source
+ * each time.
+ *
+ * It is not application when we don't really have control, for example, when
+ * source is a ReadableStream.
+ */
+const BYTES_PER_READ = 10240;
+
 /**
  * A class representing a bundle.
  *
@@ -48,8 +62,6 @@ export class SizedBundleElement {
 export class BundleReader {
   /** Cached bundle metadata. */
   private metadata: Deferred<BundleMetadata> = new Deferred<BundleMetadata>();
-  /** The reader to read from underlying binary bundle data source. */
-  private reader: ByteStreamReader;
   /**
    * Internal buffer to hold bundle content, accumulating incomplete element
    * content.
@@ -58,16 +70,16 @@ export class BundleReader {
   /** The decoder used to parse binary data into strings. */
   private textDecoder = new TextDecoder('utf-8');
 
-  constructor(
-    private bundleStream:
-      | ReadableStream<Uint8Array | ArrayBuffer>
-      | Uint8Array
-      | ArrayBuffer
-  ) {
-    this.reader = PlatformSupport.getPlatform().toByteStreamReader(
-      bundleStream
+  static fromBundleSource(source: BundleSource): BundleReader {
+    return new BundleReader(
+      PlatformSupport.getPlatform().toByteStreamReader(source, BYTES_PER_READ)
     );
+  }
 
+  constructor(
+    /** The reader to read from underlying binary bundle data source. */
+    private reader: ByteStreamReader
+  ) {
     // Read the metadata (which is the first element).
     this.nextElementImpl().then(
       element => {

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

@@ -28,7 +28,10 @@ describe('Platform', () => {
 
   it('toByteStreamReader() steps underlying data', async () => {
     const encoder = new TextEncoder();
-    const r = toByteStreamReader(encoder.encode('0123456789'), 4);
+    const r = toByteStreamReader(
+      encoder.encode('0123456789'),
+      /* bytesPerRead */ 4
+    );
 
     let result = await r.read();
     expect(result.value).to.deep.equal(encoder.encode('0123'));

packages/firestore/test/unit/util/bundle.test.ts

@@ -20,10 +20,9 @@ import {
   SizedBundleElement
 } from '../../../src/util/bundle_reader';
 import { BundleElement } from '../../../src/protos/firestore_bundle_proto';
-import { isNode } from '../../util/test_platform';
 import {
-  PlatformSupport,
-  toByteStreamReader
+  ByteStreamReader,
+  PlatformSupport
 } from '../../../src/platform/platform';
 
 /**
@@ -33,22 +32,12 @@ import {
  * @param bytesPerRead: How many bytes to read from the underlying buffer from
  * each read through the stream.
  */
-export function readableStreamFromString(
+export function byteStreamReaderFromString(
   content: string,
-  bytesPerRead = 10240
-): ReadableStream<Uint8Array | ArrayBuffer> {
-  let readFrom = 0;
+  bytesPerRead: number
+): ByteStreamReader {
   const data = new TextEncoder().encode(content);
-  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();
-      }
-    }
-  });
+  return PlatformSupport.getPlatform().toByteStreamReader(data, bytesPerRead);
 }
 
 function lengthPrefixedString(o: {}): string {
@@ -58,13 +47,10 @@ function lengthPrefixedString(o: {}): string {
 }
 
 // Testing readableStreamFromString() is working as expected.
-// eslint-disable-next-line no-restricted-properties
-(isNode() ? describe.skip : describe)('readableStreamFromString()', () => {
-  it('returns stepping readable stream', async () => {
+describe('byteStreamReaderFromString()', () => {
+  it('returns a reader stepping readable stream', async () => {
     const encoder = new TextEncoder();
-    const r = PlatformSupport.getPlatform().toByteStreamReader(
-      readableStreamFromString('0123456789', 4)
-    );
+    const r = byteStreamReaderFromString('0123456789', 4);
 
     let result = await r.read();
     expect(result.value).to.deep.equal(encoder.encode('0123'));
@@ -92,21 +78,8 @@ describe('Bundle ', () => {
 });
 
 function genericBundleReadingTests(bytesPerRead: number): void {
-  // On Node, we need to override `bytesPerRead` from it's platform's `toByteStreamReader` call.
-  if (isNode()) {
-    const platform = PlatformSupport.getPlatform();
-    platform.toByteStreamReader = source =>
-      toByteStreamReader(source as Uint8Array, bytesPerRead);
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    (PlatformSupport as any)._forceSetPlatform(platform);
-  }
-
   function bundleFromString(s: string): BundleReader {
-    if (isNode()) {
-      return new BundleReader(encoder.encode(s));
-    } else {
-      return new BundleReader(readableStreamFromString(s, bytesPerRead));
-    }
+    return new BundleReader(byteStreamReaderFromString(s, bytesPerRead));
   }
 
   const encoder = new TextEncoder();

packages/firestore/test/util/test_platform.ts

@@ -22,6 +22,7 @@ import { JsonProtoSerializer } from '../../src/remote/serializer';
 import { debugAssert, fail } from '../../src/util/assert';
 import { ConnectivityMonitor } from './../../src/remote/connectivity_monitor';
 import { NoopConnectivityMonitor } from './../../src/remote/connectivity_monitor_noop';
+import { BundleSource } from '../../src/util/bundle_reader';
 
 /* eslint-disable no-restricted-globals */
 
@@ -271,8 +272,11 @@ export class TestPlatform implements Platform {
     return this.basePlatform.randomBytes(nBytes);
   }
 
-  toByteStreamReader(source: unknown): ByteStreamReader {
-    return this.basePlatform.toByteStreamReader(source);
+  toByteStreamReader(
+    source: BundleSource,
+    bytesPerRead: number
+  ): ByteStreamReader {
+    return this.basePlatform.toByteStreamReader(source, bytesPerRead);
   }
 }