Move Datastore to ComponentProvider (#3417)

Author: schmidt-sebastian

packages/firestore/lite/src/api/database.ts

@@ -92,9 +92,11 @@ export class Firestore
     if (!this._datastorePromise) {
       const settings = this._getSettings();
       const databaseInfo = this._makeDatabaseInfo(settings.host, settings.ssl);
+      const serializer = newSerializer(databaseInfo.databaseId);
+      const datastore = newDatastore(this._credentials, serializer);
       this._datastorePromise = newConnection(databaseInfo).then(connection => {
-        const serializer = newSerializer(databaseInfo.databaseId);
-        return newDatastore(connection, this._credentials, serializer);
+        datastore.start(connection);
+        return datastore;
       });
     }
 

packages/firestore/src/core/component_provider.ts

@@ -37,7 +37,7 @@ import { RemoteStore } from '../remote/remote_store';
 import { EventManager } from './event_manager';
 import { AsyncQueue } from '../util/async_queue';
 import { DatabaseId, DatabaseInfo } from './database_info';
-import { Datastore } from '../remote/datastore';
+import { Datastore, newDatastore } from '../remote/datastore';
 import { User } from '../auth/user';
 import { PersistenceSettings } from './firestore_client';
 import { debugAssert } from '../util/assert';
@@ -55,10 +55,11 @@ import {
   MemoryEagerDelegate,
   MemoryPersistence
 } from '../local/memory_persistence';
-import { newConnectivityMonitor } from '../platform/connection';
+import { newConnection, newConnectivityMonitor } from '../platform/connection';
 import { newSerializer } from '../platform/serializer';
 import { getDocument, getWindow } from '../platform/dom';
-
+import { CredentialsProvider } from '../api/credentials';
+import { Connection } from '../remote/connection';
 const MEMORY_ONLY_PERSISTENCE_ERROR_MESSAGE =
   'You are using the memory-only build of Firestore. Persistence support is ' +
   'only available via the @firebase/firestore bundle or the ' +
@@ -67,7 +68,7 @@ const MEMORY_ONLY_PERSISTENCE_ERROR_MESSAGE =
 export interface ComponentConfiguration {
   asyncQueue: AsyncQueue;
   databaseInfo: DatabaseInfo;
-  datastore: Datastore;
+  credentials: CredentialsProvider;
   clientId: ClientId;
   initialUser: User;
   maxConcurrentLimboResolutions: number;
@@ -84,6 +85,7 @@ export interface ComponentProvider {
   localStore: LocalStore;
   syncEngine: SyncEngine;
   gcScheduler: GarbageCollectionScheduler | null;
+  datastore: Datastore;
   remoteStore: RemoteStore;
   eventManager: EventManager;
 
@@ -105,6 +107,7 @@ export class MemoryComponentProvider implements ComponentProvider {
   localStore!: LocalStore;
   syncEngine!: SyncEngine;
   gcScheduler!: GarbageCollectionScheduler | null;
+  datastore!: Datastore;
   remoteStore!: RemoteStore;
   eventManager!: EventManager;
 
@@ -114,6 +117,11 @@ export class MemoryComponentProvider implements ComponentProvider {
     await this.persistence.start();
     this.gcScheduler = this.createGarbageCollectionScheduler(cfg);
     this.localStore = this.createLocalStore(cfg);
+
+    this.datastore = this.createDatastore(cfg);
+    const connection = await this.loadConnection(cfg);
+    this.datastore.start(connection);
+
     this.remoteStore = this.createRemoteStore(cfg);
     this.syncEngine = this.createSyncEngine(cfg);
     this.eventManager = this.createEventManager(cfg);
@@ -132,6 +140,10 @@ export class MemoryComponentProvider implements ComponentProvider {
     await this.remoteStore.applyPrimaryState(this.syncEngine.isPrimaryClient);
   }
 
+  protected loadConnection(cfg: ComponentConfiguration): Promise<Connection> {
+    return newConnection(cfg.databaseInfo);
+  }
+
   createEventManager(cfg: ComponentConfiguration): EventManager {
     return new EventManager(this.syncEngine);
   }
@@ -160,10 +172,15 @@ export class MemoryComponentProvider implements ComponentProvider {
     return new MemoryPersistence(MemoryEagerDelegate.factory);
   }
 
+  createDatastore(cfg: ComponentConfiguration): Datastore {
+    const serializer = newSerializer(cfg.databaseInfo.databaseId);
+    return newDatastore(cfg.credentials, serializer);
+  }
+
   createRemoteStore(cfg: ComponentConfiguration): RemoteStore {
     return new RemoteStore(
       this.localStore,
-      cfg.datastore,
+      this.datastore,
       cfg.asyncQueue,
       onlineState =>
         this.syncEngine.applyOnlineStateChange(
@@ -182,7 +199,7 @@ export class MemoryComponentProvider implements ComponentProvider {
     return newSyncEngine(
       this.localStore,
       this.remoteStore,
-      cfg.datastore,
+      this.datastore,
       this.sharedClientState,
       cfg.initialUser,
       cfg.maxConcurrentLimboResolutions
@@ -218,7 +235,7 @@ export class IndexedDbComponentProvider extends MemoryComponentProvider {
     return newSyncEngine(
       this.localStore,
       this.remoteStore,
-      cfg.datastore,
+      this.datastore,
       this.sharedClientState,
       cfg.initialUser,
       cfg.maxConcurrentLimboResolutions
@@ -315,7 +332,7 @@ export class MultiTabIndexedDbComponentProvider extends IndexedDbComponentProvid
     const syncEngine = newMultiTabSyncEngine(
       this.localStore,
       this.remoteStore,
-      cfg.datastore,
+      this.datastore,
       this.sharedClientState,
       cfg.initialUser,
       cfg.maxConcurrentLimboResolutions

packages/firestore/src/core/firestore_client.ts

@@ -22,7 +22,6 @@ import { GarbageCollectionScheduler, Persistence } from '../local/persistence';
 import { Document, NoDocument } from '../model/document';
 import { DocumentKey } from '../model/document_key';
 import { Mutation } from '../model/mutation';
-import { newDatastore } from '../remote/datastore';
 import { RemoteStore } from '../remote/remote_store';
 import { AsyncQueue, wrapInUserErrorIfRecoverable } from '../util/async_queue';
 import { Code, FirestoreError } from '../util/error';
@@ -47,8 +46,6 @@ import {
   ComponentProvider,
   MemoryComponentProvider
 } from './component_provider';
-import { newConnection } from '../platform/connection';
-import { newSerializer } from '../platform/serializer';
 
 const LOG_TAG = 'FirestoreClient';
 const MAX_CONCURRENT_LIMBO_RESOLUTIONS = 100;
@@ -236,19 +233,11 @@ export class FirestoreClient {
     persistenceResult: Deferred<void>
   ): Promise<void> {
     try {
-      // TODO(mrschmidt): Ideally, ComponentProvider would also initialize
-      // Datastore (without duplicating the initializing logic once per
-      // provider).
-
-      const connection = await newConnection(this.databaseInfo);
-      const serializer = newSerializer(this.databaseInfo.databaseId);
-      const datastore = newDatastore(connection, this.credentials, serializer);
-
       await componentProvider.initialize({
         asyncQueue: this.asyncQueue,
         databaseInfo: this.databaseInfo,
-        datastore,
         clientId: this.clientId,
+        credentials: this.credentials,
         initialUser: user,
         maxConcurrentLimboResolutions: MAX_CONCURRENT_LIMBO_RESOLUTIONS,
         persistenceSettings

packages/firestore/src/remote/datastore.ts

@@ -20,7 +20,7 @@ import { Document, MaybeDocument } from '../model/document';
 import { DocumentKey } from '../model/document_key';
 import { Mutation } from '../model/mutation';
 import * as api from '../protos/firestore_proto_api';
-import { debugCast, hardAssert } from '../util/assert';
+import { debugAssert, debugCast, hardAssert } from '../util/assert';
 import { Code, FirestoreError } from '../util/error';
 import { Connection } from './connection';
 import {
@@ -46,28 +46,27 @@ import { Query } from '../core/query';
  * Cloud Datastore grpc API, which provides an interface that is more convenient
  * for the rest of the client SDK architecture to consume.
  */
-export class Datastore {
-  // Make sure that the structural type of `Datastore` is unique.
-  // See https://github.com/microsoft/TypeScript/issues/5451
-  private _ = undefined;
+export abstract class Datastore {
+  abstract start(connection: Connection): void;
 }
 
 /**
  * An implementation of Datastore that exposes additional state for internal
  * consumption.
  */
 class DatastoreImpl extends Datastore {
+  connection!: Connection;
   terminated = false;
 
   constructor(
-    readonly connection: Connection,
     readonly credentials: CredentialsProvider,
     readonly serializer: JsonProtoSerializer
   ) {
     super();
   }
 
-  private verifyNotTerminated(): void {
+  verifyInitialized(): void {
+    debugAssert(!!this.connection, 'Datastore.start() not called');
     if (this.terminated) {
       throw new FirestoreError(
         Code.FAILED_PRECONDITION,
@@ -76,9 +75,14 @@ class DatastoreImpl extends Datastore {
     }
   }
 
+  start(connection: Connection): void {
+    debugAssert(!this.connection, 'Datastore.start() already called');
+    this.connection = connection;
+  }
+
   /** Gets an auth token and invokes the provided RPC. */
   invokeRPC<Req, Resp>(rpcName: string, request: Req): Promise<Resp> {
-    this.verifyNotTerminated();
+    this.verifyInitialized();
     return this.credentials
       .getToken()
       .then(token => {
@@ -97,7 +101,7 @@ class DatastoreImpl extends Datastore {
     rpcName: string,
     request: Req
   ): Promise<Resp[]> {
-    this.verifyNotTerminated();
+    this.verifyInitialized();
     return this.credentials
       .getToken()
       .then(token => {
@@ -117,11 +121,10 @@ class DatastoreImpl extends Datastore {
 }
 
 export function newDatastore(
-  connection: Connection,
   credentials: CredentialsProvider,
   serializer: JsonProtoSerializer
 ): Datastore {
-  return new DatastoreImpl(connection, credentials, serializer);
+  return new DatastoreImpl(credentials, serializer);
 }
 
 export async function invokeCommitRpc(
@@ -200,6 +203,7 @@ export function newPersistentWriteStream(
   listener: WriteStreamListener
 ): PersistentWriteStream {
   const datastoreImpl = debugCast(datastore, DatastoreImpl);
+  datastoreImpl.verifyInitialized();
   return new PersistentWriteStream(
     queue,
     datastoreImpl.connection,
@@ -215,6 +219,7 @@ export function newPersistentWatchStream(
   listener: WatchStreamListener
 ): PersistentListenStream {
   const datastoreImpl = debugCast(datastore, DatastoreImpl);
+  datastoreImpl.verifyInitialized();
   return new PersistentListenStream(
     queue,
     datastoreImpl.connection,

packages/firestore/test/integration/util/internal_helpers.ts

@@ -55,7 +55,8 @@ export function withTestDatastore(
   const databaseInfo = getDefaultDatabaseInfo();
   return newConnection(databaseInfo).then(conn => {
     const serializer = newSerializer(databaseInfo.databaseId);
-    const datastore = newDatastore(conn, credentialsProvider, serializer);
+    const datastore = newDatastore(credentialsProvider, serializer);
+    datastore.start(conn);
     return fn(datastore);
   });
 }

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

@@ -61,6 +61,8 @@ import {
 } from '../../../src/local/shared_client_state';
 import { WindowLike } from '../../../src/util/types';
 import { newSerializer } from '../../../src/platform/serializer';
+import { Datastore, newDatastore } from '../../../src/remote/datastore';
+import { JsonProtoSerializer } from '../../../src/remote/serializer';
 
 /**
  * A test-only MemoryPersistence implementation that is able to inject
@@ -119,6 +121,7 @@ function failTransactionIfNeeded(
 
 export class MockIndexedDbComponentProvider extends MultiTabIndexedDbComponentProvider {
   persistence!: MockIndexedDbPersistence;
+  connection!: MockConnection;
 
   constructor(
     private readonly window: WindowLike,
@@ -127,6 +130,19 @@ export class MockIndexedDbComponentProvider extends MultiTabIndexedDbComponentPr
     super();
   }
 
+  async loadConnection(cfg: ComponentConfiguration): Promise<Connection> {
+    this.connection = new MockConnection(cfg.asyncQueue);
+    return this.connection;
+  }
+
+  createDatastore(cfg: ComponentConfiguration): Datastore {
+    const serializer = new JsonProtoSerializer(
+      cfg.databaseInfo.databaseId,
+      /* useProto3Json= */ true
+    );
+    return newDatastore(cfg.credentials, serializer);
+  }
+
   createGarbageCollectionScheduler(
     cfg: ComponentConfiguration
   ): GarbageCollectionScheduler | null {
@@ -176,11 +192,25 @@ export class MockIndexedDbComponentProvider extends MultiTabIndexedDbComponentPr
 
 export class MockMemoryComponentProvider extends MemoryComponentProvider {
   persistence!: MockMemoryPersistence;
+  connection!: MockConnection;
 
   constructor(private readonly gcEnabled: boolean) {
     super();
   }
 
+  async loadConnection(cfg: ComponentConfiguration): Promise<Connection> {
+    this.connection = new MockConnection(cfg.asyncQueue);
+    return this.connection;
+  }
+
+  createDatastore(cfg: ComponentConfiguration): Datastore {
+    const serializer = new JsonProtoSerializer(
+      cfg.databaseInfo.databaseId,
+      /* useProto3Json= */ true
+    );
+    return newDatastore(cfg.credentials, serializer);
+  }
+
   createGarbageCollectionScheduler(
     cfg: ComponentConfiguration
   ): GarbageCollectionScheduler | null {

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

@@ -51,7 +51,6 @@ import { DocumentKey } from '../../../src/model/document_key';
 import { JsonObject } from '../../../src/model/object_value';
 import { Mutation } from '../../../src/model/mutation';
 import * as api from '../../../src/protos/firestore_proto_api';
-import { Datastore, newDatastore } from '../../../src/remote/datastore';
 import { ExistenceFilter } from '../../../src/remote/existence_filter';
 import { RemoteStore } from '../../../src/remote/remote_store';
 import { mapCodeFromRpcCode } from '../../../src/remote/rpc_error';
@@ -178,7 +177,6 @@ abstract class TestRunner {
   private networkEnabled = true;
 
   // Initialized asynchronously via start().
-  private datastore!: Datastore;
   private localStore!: LocalStore;
   private remoteStore!: RemoteStore;
   private persistence!: MockMemoryPersistence | MockIndexedDbPersistence;
@@ -233,18 +231,11 @@ abstract class TestRunner {
   }
 
   async start(): Promise<void> {
-    this.connection = new MockConnection(this.queue);
-    this.datastore = newDatastore(
-      this.connection,
-      new EmptyCredentialsProvider(),
-      this.serializer
-    );
-
     const componentProvider = await this.initializeComponentProvider(
       {
         asyncQueue: this.queue,
         databaseInfo: this.databaseInfo,
-        datastore: this.datastore,
+        credentials: new EmptyCredentialsProvider(),
         clientId: this.clientId,
         initialUser: this.user,
         maxConcurrentLimboResolutions:
@@ -257,6 +248,7 @@ abstract class TestRunner {
     this.sharedClientState = componentProvider.sharedClientState;
     this.persistence = componentProvider.persistence;
     this.localStore = componentProvider.localStore;
+    this.connection = componentProvider.connection;
     this.remoteStore = componentProvider.remoteStore;
     this.syncEngine = componentProvider.syncEngine;
     this.eventManager = componentProvider.eventManager;