Review

Author: schmidt-sebastian

packages/firestore/src/core/component_provider.ts

@@ -28,6 +28,7 @@ import { EventManager } from './event_manager';
 import { AsyncQueue } from '../util/async_queue';
 import { DatabaseInfo } from './database_info';
 import { Platform } from '../platform/platform';
+import { Datastore } from '../remote/datastore';
 import { User } from '../auth/user';
 import { PersistenceSettings } from './firestore_client';
 import { debugAssert } from '../util/assert';
@@ -41,9 +42,6 @@ import {
   MemoryEagerDelegate,
   MemoryPersistence
 } from '../local/memory_persistence';
-import { Connection } from '../remote/connection';
-import { CredentialsProvider } from '../api/credentials';
-import { JsonProtoSerializer } from '../remote/serializer';
 
 const MEMORY_ONLY_PERSISTENCE_ERROR_MESSAGE =
   'You are using the memory-only build of Firestore. Persistence support is ' +
@@ -54,9 +52,7 @@ export interface ComponentConfiguration {
   asyncQueue: AsyncQueue;
   databaseInfo: DatabaseInfo;
   platform: Platform;
-  connection: Connection;
-  credentials: CredentialsProvider;
-  serializer: JsonProtoSerializer;
+  datastore: Datastore;
   clientId: ClientId;
   initialUser: User;
   maxConcurrentLimboResolutions: number;
@@ -147,9 +143,7 @@ export class MemoryComponentProvider implements ComponentProvider {
   createRemoteStore(cfg: ComponentConfiguration): RemoteStore {
     return new RemoteStore(
       this.localStore,
-      cfg.connection,
-      cfg.credentials,
-      cfg.serializer,
+      cfg.datastore,
       cfg.asyncQueue,
       onlineState =>
         this.syncEngine.applyOnlineStateChange(

packages/firestore/src/core/firestore_client.ts

@@ -23,6 +23,7 @@ import { Document, MaybeDocument, NoDocument } from '../model/document';
 import { DocumentKey } from '../model/document_key';
 import { Mutation } from '../model/mutation';
 import { Platform } from '../platform/platform';
+import { Datastore } from '../remote/datastore';
 import { RemoteStore } from '../remote/remote_store';
 import { AsyncQueue } from '../util/async_queue';
 import { Code, FirestoreError } from '../util/error';
@@ -237,14 +238,13 @@ export class FirestoreClient {
       const serializer = this.platform.newSerializer(
         this.databaseInfo.databaseId
       );
+      const datastore = new Datastore(connection, this.credentials, serializer);
 
       await componentProvider.initialize({
         asyncQueue: this.asyncQueue,
         databaseInfo: this.databaseInfo,
         platform: this.platform,
-        connection,
-        credentials: this.credentials,
-        serializer,
+        datastore,
         clientId: this.clientId,
         initialUser: user,
         maxConcurrentLimboResolutions: MAX_CONCURRENT_LIMBO_RESOLUTIONS,

packages/firestore/src/core/transaction.ts

@@ -26,7 +26,11 @@ import {
   Precondition,
   VerifyMutation
 } from '../model/mutation';
-import { Datastore } from '../remote/datastore';
+import {
+  Datastore,
+  invokeBatchGetDocumentsRpc,
+  invokeCommitRpc
+} from '../remote/datastore';
 import { fail, debugAssert } from '../util/assert';
 import { Code, FirestoreError } from '../util/error';
 import { SnapshotVersion } from './snapshot_version';
@@ -66,7 +70,7 @@ export class Transaction {
         'Firestore transactions require all reads to be executed before all writes.'
       );
     }
-    const docs = await this.datastore.lookup(keys);
+    const docs = await invokeBatchGetDocumentsRpc(this.datastore, keys);
     docs.forEach(doc => {
       if (doc instanceof NoDocument || doc instanceof Document) {
         this.recordVersion(doc);
@@ -112,7 +116,7 @@ export class Transaction {
     unwritten.forEach((key, _version) => {
       this.mutations.push(new VerifyMutation(key, this.precondition(key)));
     });
-    await this.datastore.commit(this.mutations);
+    await invokeCommitRpc(this.datastore, this.mutations);
     this.committed = true;
   }
 

packages/firestore/src/remote/datastore.ts

@@ -23,8 +23,14 @@ import * as api from '../protos/firestore_proto_api';
 import { hardAssert } from '../util/assert';
 import { Code, FirestoreError } from '../util/error';
 import { Connection } from './connection';
-
 import { JsonProtoSerializer } from './serializer';
+import {
+  PersistentListenStream,
+  PersistentWriteStream,
+  WatchStreamListener,
+  WriteStreamListener
+} from './persistent_stream';
+import { AsyncQueue } from '../util/async_queue';
 
 // The generated proto interfaces for these class are missing the database
 // field. So we add it here.
@@ -43,53 +49,13 @@ interface CommitRequest extends api.CommitRequest {
  */
 export class Datastore {
   constructor(
-    private connection: Connection,
-    private credentials: CredentialsProvider,
-    private serializer: JsonProtoSerializer
+    public readonly connection: Connection,
+    public readonly credentials: CredentialsProvider,
+    public readonly serializer: JsonProtoSerializer
   ) {}
 
-  commit(mutations: Mutation[]): Promise<MutationResult[]> {
-    const params: CommitRequest = {
-      database: this.serializer.encodedDatabaseId,
-      writes: mutations.map(m => this.serializer.toMutation(m))
-    };
-    return this.invokeRPC<CommitRequest, api.CommitResponse>(
-      'Commit',
-      params
-    ).then(response => {
-      return this.serializer.fromWriteResults(
-        response.writeResults,
-        response.commitTime
-      );
-    });
-  }
-
-  lookup(keys: DocumentKey[]): Promise<MaybeDocument[]> {
-    const params: BatchGetDocumentsRequest = {
-      database: this.serializer.encodedDatabaseId,
-      documents: keys.map(k => this.serializer.toName(k))
-    };
-    return this.invokeStreamingRPC<
-      BatchGetDocumentsRequest,
-      api.BatchGetDocumentsResponse
-    >('BatchGetDocuments', params).then(response => {
-      const docs = new Map<string, MaybeDocument>();
-      response.forEach(proto => {
-        const doc = this.serializer.fromMaybeDocument(proto);
-        docs.set(doc.key.toString(), doc);
-      });
-      const result: MaybeDocument[] = [];
-      keys.forEach(key => {
-        const doc = docs.get(key.toString());
-        hardAssert(!!doc, 'Missing entity in write response for ' + key);
-        result.push(doc);
-      });
-      return result;
-    });
-  }
-
   /** Gets an auth token and invokes the provided RPC. */
-  private invokeRPC<Req, Resp>(rpcName: string, request: Req): Promise<Resp> {
+  invokeRPC<Req, Resp>(rpcName: string, request: Req): Promise<Resp> {
     return this.credentials
       .getToken()
       .then(token => {
@@ -104,7 +70,7 @@ export class Datastore {
   }
 
   /** Gets an auth token and invokes the provided RPC with streamed results. */
-  private invokeStreamingRPC<Req, Resp>(
+  invokeStreamingRPC<Req, Resp>(
     rpcName: string,
     request: Req
   ): Promise<Resp[]> {
@@ -125,3 +91,79 @@ export class Datastore {
       });
   }
 }
+
+export function invokeCommitRpc(
+  datastore: Datastore,
+  mutations: Mutation[]
+): Promise<MutationResult[]> {
+  const params: CommitRequest = {
+    database: datastore.serializer.encodedDatabaseId,
+    writes: mutations.map(m => datastore.serializer.toMutation(m))
+  };
+  return datastore
+    .invokeRPC<CommitRequest, api.CommitResponse>('Commit', params)
+    .then(response => {
+      return datastore.serializer.fromWriteResults(
+        response.writeResults,
+        response.commitTime
+      );
+    });
+}
+
+export function invokeBatchGetDocumentsRpc(
+  datastore: Datastore,
+  keys: DocumentKey[]
+): Promise<MaybeDocument[]> {
+  const params: BatchGetDocumentsRequest = {
+    database: datastore.serializer.encodedDatabaseId,
+    documents: keys.map(k => datastore.serializer.toName(k))
+  };
+
+  return datastore
+    .invokeStreamingRPC<
+      BatchGetDocumentsRequest,
+      api.BatchGetDocumentsResponse
+    >('BatchGetDocuments', params)
+    .then(response => {
+      const docs = new Map<string, MaybeDocument>();
+      response.forEach(proto => {
+        const doc = datastore.serializer.fromMaybeDocument(proto);
+        docs.set(doc.key.toString(), doc);
+      });
+      const result: MaybeDocument[] = [];
+      keys.forEach(key => {
+        const doc = docs.get(key.toString());
+        hardAssert(!!doc, 'Missing entity in write response for ' + key);
+        result.push(doc);
+      });
+      return result;
+    });
+}
+
+export function newPersistentWriteStream(
+  datastore: Datastore,
+  queue: AsyncQueue,
+  listener: WriteStreamListener
+): PersistentWriteStream {
+  return new PersistentWriteStream(
+    queue,
+    datastore.connection,
+    datastore.credentials,
+    datastore.serializer,
+    listener
+  );
+}
+
+export function newPersistentWatchStream(
+  datastore: Datastore,
+  queue: AsyncQueue,
+  listener: WatchStreamListener
+): PersistentListenStream {
+  return new PersistentListenStream(
+    queue,
+    datastore.connection,
+    datastore.credentials,
+    datastore.serializer,
+    listener
+  );
+}

packages/firestore/src/remote/remote_store.ts

@@ -32,7 +32,11 @@ import { logDebug } from '../util/log';
 import { DocumentKeySet } from '../model/collections';
 import { AsyncQueue } from '../util/async_queue';
 import { ConnectivityMonitor, NetworkStatus } from './connectivity_monitor';
-import { Datastore } from './datastore';
+import {
+  Datastore,
+  newPersistentWatchStream,
+  newPersistentWriteStream
+} from './datastore';
 import { OnlineStateTracker } from './online_state_tracker';
 import {
   PersistentListenStream,
@@ -50,9 +54,6 @@ import {
   WatchTargetChangeState
 } from './watch_change';
 import { ByteString } from '../util/byte_string';
-import { Connection } from './connection';
-import { CredentialsProvider } from '../api/credentials';
-import { JsonProtoSerializer } from './serializer';
 
 const LOG_TAG = 'RemoteStore';
 
@@ -109,7 +110,6 @@ export class RemoteStore implements TargetMetadataProvider {
    */
   private listenTargets = new Map<TargetId, TargetData>();
 
-  private datastore: Datastore;
   private connectivityMonitor: ConnectivityMonitor;
   private watchStream: PersistentListenStream;
   private writeStream: PersistentWriteStream;
@@ -126,10 +126,12 @@ export class RemoteStore implements TargetMetadataProvider {
   private onlineStateTracker: OnlineStateTracker;
 
   constructor(
+    /**
+     * The local store, used to fill the write pipeline with outbound mutations.
+     */
     private localStore: LocalStore,
-    connection: Connection,
-    credentials: CredentialsProvider,
-    serializer: JsonProtoSerializer,
+    /** The client-side proxy for interacting with the backend. */
+    private datastore: Datastore,
     asyncQueue: AsyncQueue,
     onlineStateHandler: (onlineState: OnlineState) => void,
     connectivityMonitor: ConnectivityMonitor
@@ -152,33 +154,19 @@ export class RemoteStore implements TargetMetadataProvider {
       onlineStateHandler
     );
 
-    this.datastore = new Datastore(connection, credentials, serializer);
-
     // Create streams (but note they're not started yet).
-    this.watchStream = new PersistentListenStream(
-      asyncQueue,
-      connection,
-      credentials,
-      serializer,
-      {
-        onOpen: this.onWatchStreamOpen.bind(this),
-        onClose: this.onWatchStreamClose.bind(this),
-        onWatchChange: this.onWatchStreamChange.bind(this)
-      }
-    );
+    this.watchStream = newPersistentWatchStream(this.datastore, asyncQueue, {
+      onOpen: this.onWatchStreamOpen.bind(this),
+      onClose: this.onWatchStreamClose.bind(this),
+      onWatchChange: this.onWatchStreamChange.bind(this)
+    });
 
-    this.writeStream = new PersistentWriteStream(
-      asyncQueue,
-      connection,
-      credentials,
-      serializer,
-      {
-        onOpen: this.onWriteStreamOpen.bind(this),
-        onClose: this.onWriteStreamClose.bind(this),
-        onHandshakeComplete: this.onWriteHandshakeComplete.bind(this),
-        onMutationResult: this.onMutationResult.bind(this)
-      }
-    );
+    this.writeStream = newPersistentWriteStream(this.datastore, asyncQueue, {
+      onOpen: this.onWriteStreamOpen.bind(this),
+      onClose: this.onWriteStreamClose.bind(this),
+      onHandshakeComplete: this.onWriteHandshakeComplete.bind(this),
+      onMutationResult: this.onMutationResult.bind(this)
+    });
   }
 
   /**