WIP

Author: schmidt-sebastian

packages/firestore/exp/dependencies.json

@@ -19,7 +19,12 @@ import { debugAssert } from '../util/assert';
 import { EventHandler } from '../util/misc';
 import { ObjectMap } from '../util/obj_map';
 import { canonifyQuery, Query, queryEquals, stringifyQuery } from './query';
-import { SyncEngine, SyncEngineListener } from './sync_engine';
+import {
+  SyncEngine,
+  SyncEngineListener,
+  listen as syncEngineListen,
+  unlisten as syncEngineUnlisten
+} from './sync_engine';
 import { OnlineState } from './types';
 import { ChangeType, DocumentViewChange, ViewSnapshot } from './view_snapshot';
 import { wrapInUserErrorIfRecoverable } from '../util/async_queue';
@@ -47,79 +52,19 @@ export interface Observer<T> {
  * backend.
  */
 export class EventManager implements SyncEngineListener {
-  private queries = new ObjectMap<Query, QueryListenersInfo>(
+  queries = new ObjectMap<Query, QueryListenersInfo>(
     q => canonifyQuery(q),
     queryEquals
   );
 
-  private onlineState = OnlineState.Unknown;
+  onlineState = OnlineState.Unknown;
 
   private snapshotsInSyncListeners: Set<Observer<void>> = new Set();
 
-  constructor(private syncEngine: SyncEngine) {
+  constructor(public syncEngine: SyncEngine) {
     this.syncEngine.subscribe(this);
   }
 
-  async listen(listener: QueryListener): Promise<void> {
-    const query = listener.query;
-    let firstListen = false;
-
-    let queryInfo = this.queries.get(query);
-    if (!queryInfo) {
-      firstListen = true;
-      queryInfo = new QueryListenersInfo();
-    }
-
-    if (firstListen) {
-      try {
-        queryInfo.viewSnap = await this.syncEngine.listen(query);
-      } catch (e) {
-        const firestoreError = wrapInUserErrorIfRecoverable(
-          e,
-          `Initialization of query '${stringifyQuery(listener.query)}' failed`
-        );
-        listener.onError(firestoreError);
-        return;
-      }
-    }
-
-    this.queries.set(query, queryInfo);
-    queryInfo.listeners.push(listener);
-
-    // Run global snapshot listeners if a consistent snapshot has been emitted.
-    const raisedEvent = listener.applyOnlineStateChange(this.onlineState);
-    debugAssert(
-      !raisedEvent,
-      "applyOnlineStateChange() shouldn't raise an event for brand-new listeners."
-    );
-
-    if (queryInfo.viewSnap) {
-      const raisedEvent = listener.onViewSnapshot(queryInfo.viewSnap);
-      if (raisedEvent) {
-        this.raiseSnapshotsInSyncEvent();
-      }
-    }
-  }
-
-  async unlisten(listener: QueryListener): Promise<void> {
-    const query = listener.query;
-    let lastListen = false;
-
-    const queryInfo = this.queries.get(query);
-    if (queryInfo) {
-      const i = queryInfo.listeners.indexOf(listener);
-      if (i >= 0) {
-        queryInfo.listeners.splice(i, 1);
-        lastListen = queryInfo.listeners.length === 0;
-      }
-    }
-
-    if (lastListen) {
-      this.queries.delete(query);
-      return this.syncEngine.unlisten(query);
-    }
-  }
-
   onWatchChange(viewSnaps: ViewSnapshot[]): void {
     let raisedEvent = false;
     for (const viewSnap of viewSnaps) {
@@ -180,7 +125,7 @@ export class EventManager implements SyncEngineListener {
   }
 
   // Call all global snapshot listeners that have been set.
-  private raiseSnapshotsInSyncEvent(): void {
+  raiseSnapshotsInSyncEvent(): void {
     this.snapshotsInSyncListeners.forEach(observer => {
       observer.next();
     });
@@ -357,3 +302,72 @@ export class QueryListener {
     this.queryObserver.next(snap);
   }
 }
+
+export async function listen(
+  eventManager: EventManager,
+  listener: QueryListener
+): Promise<void> {
+  const query = listener.query;
+  let firstListen = false;
+
+  let queryInfo = eventManager.queries.get(query);
+  if (!queryInfo) {
+    firstListen = true;
+    queryInfo = new QueryListenersInfo();
+  }
+
+  if (firstListen) {
+    try {
+      queryInfo.viewSnap = await syncEngineListen(
+        eventManager.syncEngine,
+        query
+      );
+    } catch (e) {
+      const firestoreError = wrapInUserErrorIfRecoverable(
+        e,
+        `Initialization of query '${stringifyQuery(listener.query)}' failed`
+      );
+      listener.onError(firestoreError);
+      return;
+    }
+  }
+
+  eventManager.queries.set(query, queryInfo);
+  queryInfo.listeners.push(listener);
+
+  // Run global snapshot listeners if a consistent snapshot has been emitted.
+  const raisedEvent = listener.applyOnlineStateChange(eventManager.onlineState);
+  debugAssert(
+    !raisedEvent,
+    "applyOnlineStateChange() shouldn't raise an event for brand-new listeners."
+  );
+
+  if (queryInfo.viewSnap) {
+    const raisedEvent = listener.onViewSnapshot(queryInfo.viewSnap);
+    if (raisedEvent) {
+      eventManager.raiseSnapshotsInSyncEvent();
+    }
+  }
+}
+
+export async function unlisten(
+  eventManager: EventManager,
+  listener: QueryListener
+): Promise<void> {
+  const query = listener.query;
+  let lastListen = false;
+
+  const queryInfo = eventManager.queries.get(query);
+  if (queryInfo) {
+    const i = queryInfo.listeners.indexOf(listener);
+    if (i >= 0) {
+      queryInfo.listeners.splice(i, 1);
+      lastListen = queryInfo.listeners.length === 0;
+    }
+  }
+
+  if (lastListen) {
+    eventManager.queries.delete(query);
+    return syncEngineUnlisten(eventManager.syncEngine, query);
+  }
+}

packages/firestore/src/core/event_manager.ts

@@ -33,7 +33,9 @@ import {
   EventManager,
   ListenOptions,
   Observer,
-  QueryListener
+  QueryListener,
+  listen,
+  unlisten
 } from './event_manager';
 import { SyncEngine } from './sync_engine';
 import { View } from './view';
@@ -403,10 +405,10 @@ export class FirestoreClient {
     this.verifyNotTerminated();
     const wrappedObserver = new AsyncObserver(observer);
     const listener = new QueryListener(query, wrappedObserver, options);
-    this.asyncQueue.enqueueAndForget(() => this.eventMgr.listen(listener));
+    this.asyncQueue.enqueueAndForget(() => listen(this.eventMgr, listener));
     return () => {
       wrappedObserver.mute();
-      this.asyncQueue.enqueueAndForget(() => this.eventMgr.unlisten(listener));
+      this.asyncQueue.enqueueAndForget(() => unlisten(this.eventMgr, listener));
     };
   }
 
@@ -569,10 +571,10 @@ export function enqueueListen(
 ): Unsubscribe {
   const wrappedObserver = new AsyncObserver(observer);
   const listener = new QueryListener(query, wrappedObserver, options);
-  asyncQueue.enqueueAndForget(() => eventManger.listen(listener));
+  asyncQueue.enqueueAndForget(() => listen(eventManger, listener));
   return () => {
     wrappedObserver.mute();
-    asyncQueue.enqueueAndForget(() => eventManger.unlisten(listener));
+    asyncQueue.enqueueAndForget(() => unlisten(eventManger, listener));
   };
 }