Untangle Datastore

Author: schmidt-sebastian

packages/firestore/src/core/component_provider.ts

@@ -28,7 +28,6 @@ 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';
@@ -42,6 +41,9 @@ 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 ' +
@@ -52,7 +54,9 @@ export interface ComponentConfiguration {
   asyncQueue: AsyncQueue;
   databaseInfo: DatabaseInfo;
   platform: Platform;
-  datastore: Datastore;
+  connection: Connection;
+  credentials: CredentialsProvider;
+  serializer: JsonProtoSerializer;
   clientId: ClientId;
   initialUser: User;
   maxConcurrentLimboResolutions: number;
@@ -143,7 +147,9 @@ export class MemoryComponentProvider implements ComponentProvider {
   createRemoteStore(cfg: ComponentConfiguration): RemoteStore {
     return new RemoteStore(
       this.localStore,
-      cfg.datastore,
+      cfg.connection,
+      cfg.credentials,
+      cfg.serializer,
       cfg.asyncQueue,
       onlineState =>
         this.syncEngine.applyOnlineStateChange(

packages/firestore/src/core/firestore_client.ts

@@ -23,7 +23,6 @@ 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';
@@ -238,18 +237,14 @@ export class FirestoreClient {
       const serializer = this.platform.newSerializer(
         this.databaseInfo.databaseId
       );
-      const datastore = new Datastore(
-        this.asyncQueue,
-        connection,
-        this.credentials,
-        serializer
-      );
 
       await componentProvider.initialize({
         asyncQueue: this.asyncQueue,
         databaseInfo: this.databaseInfo,
         platform: this.platform,
-        datastore,
+        connection,
+        credentials: this.credentials,
+        serializer,
         clientId: this.clientId,
         initialUser: user,
         maxConcurrentLimboResolutions: MAX_CONCURRENT_LIMBO_RESOLUTIONS,

packages/firestore/src/remote/datastore.ts

@@ -16,21 +16,13 @@
  */
 
 import { CredentialsProvider } from '../api/credentials';
-import { maybeDocumentMap } from '../model/collections';
 import { MaybeDocument } from '../model/document';
 import { DocumentKey } from '../model/document_key';
 import { Mutation, MutationResult } from '../model/mutation';
 import * as api from '../protos/firestore_proto_api';
 import { hardAssert } from '../util/assert';
-import { AsyncQueue } from '../util/async_queue';
 import { Code, FirestoreError } from '../util/error';
 import { Connection } from './connection';
-import {
-  WatchStreamListener,
-  WriteStreamListener,
-  PersistentListenStream,
-  PersistentWriteStream
-} from './persistent_stream';
 
 import { JsonProtoSerializer } from './serializer';
 
@@ -51,36 +43,11 @@ interface CommitRequest extends api.CommitRequest {
  */
 export class Datastore {
   constructor(
-    private queue: AsyncQueue,
     private connection: Connection,
     private credentials: CredentialsProvider,
     private serializer: JsonProtoSerializer
   ) {}
 
-  newPersistentWriteStream(
-    listener: WriteStreamListener
-  ): PersistentWriteStream {
-    return new PersistentWriteStream(
-      this.queue,
-      this.connection,
-      this.credentials,
-      this.serializer,
-      listener
-    );
-  }
-
-  newPersistentWatchStream(
-    listener: WatchStreamListener
-  ): PersistentListenStream {
-    return new PersistentListenStream(
-      this.queue,
-      this.connection,
-      this.credentials,
-      this.serializer,
-      listener
-    );
-  }
-
   commit(mutations: Mutation[]): Promise<MutationResult[]> {
     const params: CommitRequest = {
       database: this.serializer.encodedDatabaseId,
@@ -106,14 +73,14 @@ export class Datastore {
       BatchGetDocumentsRequest,
       api.BatchGetDocumentsResponse
     >('BatchGetDocuments', params).then(response => {
-      let docs = maybeDocumentMap();
+      const docs = new Map<string, MaybeDocument>();
       response.forEach(proto => {
         const doc = this.serializer.fromMaybeDocument(proto);
-        docs = docs.insert(doc.key, doc);
+        docs.set(doc.key.toString(), doc);
       });
       const result: MaybeDocument[] = [];
       keys.forEach(key => {
-        const doc = docs.get(key);
+        const doc = docs.get(key.toString());
         hardAssert(!!doc, 'Missing entity in write response for ' + key);
         result.push(doc);
       });

packages/firestore/src/remote/remote_store.ts

@@ -50,6 +50,9 @@ 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';
 
@@ -106,6 +109,7 @@ export class RemoteStore implements TargetMetadataProvider {
    */
   private listenTargets = new Map<TargetId, TargetData>();
 
+  private datastore: Datastore;
   private connectivityMonitor: ConnectivityMonitor;
   private watchStream: PersistentListenStream;
   private writeStream: PersistentWriteStream;
@@ -122,12 +126,10 @@ export class RemoteStore implements TargetMetadataProvider {
   private onlineStateTracker: OnlineStateTracker;
 
   constructor(
-    /**
-     * The local store, used to fill the write pipeline with outbound mutations.
-     */
     private localStore: LocalStore,
-    /** The client-side proxy for interacting with the backend. */
-    private datastore: Datastore,
+    connection: Connection,
+    credentials: CredentialsProvider,
+    serializer: JsonProtoSerializer,
     asyncQueue: AsyncQueue,
     onlineStateHandler: (onlineState: OnlineState) => void,
     connectivityMonitor: ConnectivityMonitor
@@ -150,19 +152,33 @@ export class RemoteStore implements TargetMetadataProvider {
       onlineStateHandler
     );
 
+    this.datastore = new Datastore(connection, credentials, serializer);
+
     // Create streams (but note they're not started yet).
-    this.watchStream = this.datastore.newPersistentWatchStream({
-      onOpen: this.onWatchStreamOpen.bind(this),
-      onClose: this.onWatchStreamClose.bind(this),
-      onWatchChange: this.onWatchStreamChange.bind(this)
-    });
+    this.watchStream = new PersistentListenStream(
+      asyncQueue,
+      connection,
+      credentials,
+      serializer,
+      {
+        onOpen: this.onWatchStreamOpen.bind(this),
+        onClose: this.onWatchStreamClose.bind(this),
+        onWatchChange: this.onWatchStreamChange.bind(this)
+      }
+    );
 
-    this.writeStream = this.datastore.newPersistentWriteStream({
-      onOpen: this.onWriteStreamOpen.bind(this),
-      onClose: this.onWriteStreamClose.bind(this),
-      onHandshakeComplete: this.onWriteHandshakeComplete.bind(this),
-      onMutationResult: this.onMutationResult.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)
+      }
+    );
   }
 
   /**