Bundles load from secondary tabs

packages/firestore/src/core/firestore_client.ts

@@ -53,6 +53,7 @@ import { LoadBundleTask } from '../api/bundle';
 import { newConnection } from '../platform/connection';
 import { newSerializer } from '../platform/serializer';
 import { toByteStreamReader } from '../platform/byte_stream_reader';
+import { newTextEncoder } from '../platform/dom';
 
 const LOG_TAG = 'FirestoreClient';
 const MAX_CONCURRENT_LIMBO_RESOLUTIONS = 100;
@@ -529,7 +530,7 @@ export class FirestoreClient {
 
     let content: ReadableStream<Uint8Array> | ArrayBuffer;
     if (typeof data === 'string') {
-      content = new TextEncoder().encode(data);
+      content = newTextEncoder().encode(data);
     } else {
       content = data;
     }

packages/firestore/src/core/sync_engine.ts

@@ -293,7 +293,7 @@ class SyncEngineImpl implements SyncEngine {
     protected remoteStore: RemoteStore,
     protected datastore: Datastore,
     // PORTING NOTE: Manages state synchronization in multi-tab environments.
-    protected sharedClientState: SharedClientState,
+    public sharedClientState: SharedClientState,
     private currentUser: User,
     private maxConcurrentLimboResolutions: number
   ) {}
@@ -1100,6 +1100,12 @@ class MultiTabSyncEngineImpl extends SyncEngineImpl {
     }
   }
 
+  synchronizeWithChangedDocuments(): Promise<void> {
+    return this.localStore
+      .getNewDocumentChanges()
+      .then(changes => this.emitNewSnapsAndNotifyLocalStore(changes));
+  }
+
   async applyBatchState(
     batchId: BatchId,
     batchState: MutationBatchState,
@@ -1412,7 +1418,9 @@ export function loadBundle(
   syncEngineImpl.assertSubscribed('loadBundle()');
 
   // eslint-disable-next-line @typescript-eslint/no-floating-promises
-  loadBundleImpl(syncEngineImpl, bundleReader, task);
+  loadBundleImpl(syncEngineImpl, bundleReader, task).then(() => {
+    syncEngineImpl.sharedClientState.notifyBundleChangedRemoteDocuments();
+  });
 }
 
 async function loadBundleImpl(

packages/firestore/src/local/shared_client_state.ts

@@ -40,6 +40,7 @@ import {
 import {
   CLIENT_STATE_KEY_PREFIX,
   ClientStateSchema,
+  createRemoteDocumentsLoadFromBundleKey,
   createWebStorageClientStateKey,
   createWebStorageMutationBatchKey,
   createWebStorageOnlineStateKey,
@@ -173,6 +174,12 @@ export interface SharedClientState {
   setOnlineState(onlineState: OnlineState): void;
 
   writeSequenceNumber(sequenceNumber: ListenSequenceNumber): void;
+
+  /**
+   * Notifies other clients when remote documents have changed due to loading
+   * a bundle.
+   */
+  notifyBundleChangedRemoteDocuments(): void;
 }
 
 /**
@@ -477,6 +484,7 @@ export class WebStorageSharedClientState implements SharedClientState {
   private readonly sequenceNumberKey: string;
   private readonly storageListener = this.handleWebStorageEvent.bind(this);
   private readonly onlineStateKey: string;
+  private readonly bundleChangedRemoteDocumentsKey: string;
   private readonly clientStateKeyRe: RegExp;
   private readonly mutationBatchKeyRe: RegExp;
   private readonly queryTargetKeyRe: RegExp;
@@ -532,6 +540,10 @@ export class WebStorageSharedClientState implements SharedClientState {
 
     this.onlineStateKey = createWebStorageOnlineStateKey(this.persistenceKey);
 
+    this.bundleChangedRemoteDocumentsKey = createRemoteDocumentsLoadFromBundleKey(
+      this.persistenceKey
+    );
+
     // Rather than adding the storage observer during start(), we add the
     // storage observer during initialization. This ensures that we collect
     // events before other components populate their initial state (during their
@@ -711,6 +723,10 @@ export class WebStorageSharedClientState implements SharedClientState {
     this.persistOnlineState(onlineState);
   }
 
+  notifyBundleChangedRemoteDocuments(): void {
+    this.persistBundleChangedRemoteDocumentsState();
+  }
+
   shutdown(): void {
     if (this.started) {
       this.window.removeEventListener('storage', this.storageListener);
@@ -818,6 +834,8 @@ export class WebStorageSharedClientState implements SharedClientState {
           if (sequenceNumber !== ListenSequence.INVALID) {
             this.sequenceNumberHandler!(sequenceNumber);
           }
+        } else if (storageEvent.key === this.bundleChangedRemoteDocumentsKey) {
+          return this.syncEngine!.synchronizeWithChangedDocuments();
         }
       });
     }
@@ -883,6 +901,10 @@ export class WebStorageSharedClientState implements SharedClientState {
     this.setItem(targetKey, targetMetadata.toWebStorageJSON());
   }
 
+  private persistBundleChangedRemoteDocumentsState(): void {
+    this.setItem(this.bundleChangedRemoteDocumentsKey, 'value-not-used');
+  }
+
   /**
    * Parses a client state key in WebStorage. Returns null if the key does not
    * match the expected key format.
@@ -1131,4 +1153,8 @@ export class MemorySharedClientState implements SharedClientState {
   shutdown(): void {}
 
   writeSequenceNumber(sequenceNumber: ListenSequenceNumber): void {}
+
+  notifyBundleChangedRemoteDocuments(): void {
+    // No op.
+  }
 }

packages/firestore/src/local/shared_client_state_schema.ts

@@ -115,6 +115,18 @@ export function createWebStorageOnlineStateKey(persistenceKey: string): string {
   return `${ONLINE_STATE_KEY_PREFIX}_${persistenceKey}`;
 }
 
+// The WebStorage prefix that plays as a event to indicate the remote documents
+// might have changed due to some secondary tabs loading a bundle.
+// format of the key is:
+//     firestore_remote_documents_changed_<persistenceKey>
+export const REMOTE_DOCUMENTS_LOAD_FROM_BUNDLE_KEY_PREFIX =
+  'firestore_remote_documents_changed';
+export function createRemoteDocumentsLoadFromBundleKey(
+  persistenceKey: string
+): string {
+  return `${REMOTE_DOCUMENTS_LOAD_FROM_BUNDLE_KEY_PREFIX}_${persistenceKey}`;
+}
+
 /**
  * The JSON representation of the system's online state, as written by the
  * primary client.

packages/firestore/src/local/shared_client_state_syncer.ts

@@ -49,4 +49,10 @@ export interface SharedClientStateSyncer {
 
   /** Returns the IDs of the clients that are currently active. */
   getActiveClients(): Promise<ClientId[]>;
+
+  /**
+   * Retrieves newly changed documents from remote document cache and raises
+   * snapshots if needed.
+   */
+  synchronizeWithChangedDocuments(): Promise<void>;
 }

packages/firestore/src/platform/browser/dom.ts

@@ -28,3 +28,17 @@ export function getDocument(): Document | null {
   // eslint-disable-next-line no-restricted-globals
   return typeof document !== 'undefined' ? document : null;
 }
+
+/**
+ * An instance of the Platform's 'TextEncoder' implementation.
+ */
+export function newTextEncoder(): TextEncoder {
+  return new TextEncoder();
+}
+
+/**
+ * An instance of the Platform's 'TextDecoder' implementation.
+ */
+export function newTextDecoder(): TextDecoder {
+  return new TextDecoder('utf-8');
+}

packages/firestore/src/platform/dom.ts

@@ -41,3 +41,29 @@ export function getDocument(): Document | null {
     return browser.getDocument();
   }
 }
+
+/**
+ * An instance of the Platform's 'TextEncoder' implementation.
+ */
+export function newTextEncoder(): TextEncoder {
+  if (isNode()) {
+    return node.newTextEncoder();
+  } else if (isReactNative()) {
+    return rn.newTextEncoder();
+  } else {
+    return browser.newTextEncoder();
+  }
+}
+
+/**
+ * An instance of the Platform's 'TextDecoder' implementation.
+ */
+export function newTextDecoder(): TextDecoder {
+  if (isNode()) {
+    return node.newTextDecoder() as TextDecoder;
+  } else if (isReactNative()) {
+    return rn.newTextDecoder();
+  } else {
+    return browser.newTextDecoder();
+  }
+}

packages/firestore/src/platform/node/dom.ts

@@ -15,6 +15,8 @@
  * limitations under the License.
  */
 
+import { TextEncoder, TextDecoder } from 'util';
+
 /** The Platform's 'window' implementation or null if not available. */
 export function getWindow(): Window | null {
   if (process.env.USE_MOCK_PERSISTENCE === 'YES') {
@@ -29,3 +31,17 @@ export function getWindow(): Window | null {
 export function getDocument(): Document | null {
   return null;
 }
+
+/**
+ * An instance of the Platform's 'TextEncoder' implementation.
+ */
+export function newTextEncoder(): TextEncoder {
+  return new TextEncoder();
+}
+
+/**
+ * An instance of the Platform's 'TextDecoder' implementation.
+ */
+export function newTextDecoder(): TextDecoder {
+  return new TextDecoder('utf-8');
+}

packages/firestore/src/platform/rn/dom.ts

@@ -15,4 +15,9 @@
  * limitations under the License.
  */
 
-export { getWindow, getDocument } from '../browser/dom';
+export {
+  getWindow,
+  getDocument,
+  newTextEncoder,
+  newTextDecoder
+} from '../browser/dom';

packages/firestore/src/util/bundle_reader.ts

@@ -22,6 +22,7 @@ import {
 import { Deferred } from './promise';
 import { debugAssert } from './assert';
 import { toByteStreamReader } from '../platform/byte_stream_reader';
+import { newTextDecoder } from '../platform/dom';
 
 /**
  * A complete element in the bundle stream, together with the byte length it
@@ -67,7 +68,7 @@ export class BundleReader {
    */
   private buffer: Uint8Array = new Uint8Array();
   /** The decoder used to parse binary data into strings. */
-  private textDecoder = new TextDecoder('utf-8');
+  private textDecoder: TextDecoder | null;
 
   static fromBundleSource(source: BundleSource): BundleReader {
     return new BundleReader(toByteStreamReader(source, BYTES_PER_READ));
@@ -77,6 +78,8 @@ export class BundleReader {
     /** The reader to read from underlying binary bundle data source. */
     private reader: ReadableStreamReader<Uint8Array>
   ) {
+    this.textDecoder = newTextDecoder();
+    debugAssert(!!this.textDecoder, 'Cannot create a valid text decoder.');
     // Read the metadata (which is the first element).
     this.nextElementImpl().then(
       element => {
@@ -131,7 +134,7 @@ export class BundleReader {
       return null;
     }
 
-    const lengthString = this.textDecoder.decode(lengthBuffer);
+    const lengthString = this.textDecoder!.decode(lengthBuffer);
     const length = Number(lengthString);
     if (isNaN(length)) {
       this.raiseError(`length string (${lengthString}) is not valid number`);
@@ -199,7 +202,7 @@ export class BundleReader {
       }
     }
 
-    const result = this.textDecoder.decode(this.buffer.slice(0, length));
+    const result = this.textDecoder!.decode(this.buffer.slice(0, length));
     // Update the internal buffer to drop the read json string.
     this.buffer = this.buffer.slice(length);
     return result;

packages/firestore/test/integration/api_internal/bundle.test.ts

@@ -27,6 +27,9 @@ import { DatabaseId } from '../../../src/core/database_info';
 import { key } from '../../util/helpers';
 import { EventsAccumulator } from '../util/events_accumulator';
 import { TestBundleBuilder } from '../../unit/util/bundle_data';
+import { newTextEncoder } from '../../../src/platform/dom';
+
+export const encoder = newTextEncoder();
 
 function verifySuccessProgress(p: firestore.LoadBundleTaskProgress): void {
   expect(p.taskState).to.equal('Success');
@@ -45,7 +48,6 @@ function verifyInProgress(
 }
 
 apiDescribe('Bundles', (persistence: boolean) => {
-  const encoder = new TextEncoder();
   const testDocs: { [key: string]: firestore.DocumentData } = {
     a: { k: { stringValue: 'a' }, bar: { integerValue: 1 } },
     b: { k: { stringValue: 'b' }, bar: { integerValue: 2 } }

packages/firestore/test/unit/local/persistence_test_helpers.ts

@@ -170,6 +170,8 @@ class NoOpSharedClientStateSyncer implements SharedClientStateSyncer {
     removed: TargetId[]
   ): Promise<void> {}
   applyOnlineStateChange(onlineState: OnlineState): void {}
+
+  async synchronizeWithChangedDocuments(): Promise<void> {}
 }
 /**
  * Populates Web Storage with instance data from a pre-existing client.

packages/firestore/test/unit/local/web_storage_shared_client_state.test.ts

@@ -150,6 +150,8 @@ class TestSharedClientSyncer implements SharedClientStateSyncer {
   applyOnlineStateChange(onlineState: OnlineState): void {
     this.onlineState = onlineState;
   }
+
+  async synchronizeWithChangedDocuments(): Promise<void> {}
 }
 
 describe('WebStorageSharedClientState', () => {

packages/firestore/test/unit/specs/bundle_spec.test.ts

@@ -293,12 +293,9 @@ describeSpec('Bundles:', ['no-ios', 'no-android'], () => {
         .client(1)
         .loadBundle(bundleString1)
         .client(0)
-        .becomeVisible();
-      // TODO(wuandy): Loading from secondary client does not notify other
-      // clients for now. We need to fix it and uncomment below.
-      // .expectEvents(query, {
-      //   modified: [doc('collection/a', 500, { value: 'b' })],
-      // })
+        .expectEvents(query, {
+          modified: [doc('collection/a', 500, { value: 'b' })]
+        });
     }
   );
 

packages/firestore/test/unit/specs/spec_test_runner.ts

@@ -126,6 +126,7 @@ import {
 } from '../../util/test_platform';
 import { toByteStreamReader } from '../../../src/platform/byte_stream_reader';
 import { logWarn } from '../../../src/util/log';
+import { newTextEncoder } from '../../../src/platform/dom';
 
 const ARBITRARY_SEQUENCE_NUMBER = 2;
 
@@ -456,7 +457,7 @@ abstract class TestRunner {
 
   private async doLoadBundle(bundle: string): Promise<void> {
     const reader = new BundleReader(
-      toByteStreamReader(new TextEncoder().encode(bundle))
+      toByteStreamReader(newTextEncoder().encode(bundle))
     );
     const task = new LoadBundleTask();
     return this.queue.enqueue(async () => {

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

@@ -39,9 +39,12 @@ import {
   doc1,
   doc2
 } from './bundle_data';
+import { newTextEncoder } from '../../../src/platform/dom';
 
 use(chaiAsPromised);
 
+const encoder = newTextEncoder();
+
 /**
  * Create a `ReadableStream` from a string.
  *
@@ -53,14 +56,13 @@ export function byteStreamReaderFromString(
   content: string,
   bytesPerRead: number
 ): ReadableStreamReader<Uint8Array> {
-  const data = new TextEncoder().encode(content);
+  const data = encoder.encode(content);
   return toByteStreamReader(data, bytesPerRead);
 }
 
 // Testing readableStreamFromString() is working as expected.
 describe('byteStreamReaderFromString()', () => {
   it('returns a reader stepping readable stream', async () => {
-    const encoder = new TextEncoder();
     const r = byteStreamReaderFromString('0123456789', 4);
 
     let result = await r.read();
@@ -93,8 +95,6 @@ function genericBundleReadingTests(bytesPerRead: number): void {
     return new BundleReader(byteStreamReaderFromString(s, bytesPerRead));
   }
 
-  const encoder = new TextEncoder();
-
   async function getAllElements(
     bundle: BundleReader
   ): Promise<SizedBundleElement[]> {