review feedback, queue outstanding gets

Author: jmwski

.changeset/many-snails-kneel.md

@@ -1,5 +1,5 @@
 ---
-"@firebase/database": patch
+"@firebase/database": minor
 ---
 
 Add a `get` method for database queries that returns server result when connected

packages/database/CHANGELOG.md

@@ -1,7 +1,5 @@
 # Unreleased
 
-- [changed] Added support for query get operation.
-
 ## 0.6.13
 
 ### Patch Changes

packages/database/src/api/Query.ts

@@ -299,7 +299,7 @@ export class Query {
    * Get the server-value for this query, or return a cached value if not connected.
    */
   get(): Promise<DataSnapshot> {
-    return this.repo.get(this);
+    return this.repo.getValue(this);
   }
 
   /**

packages/database/src/core/PersistentConnection.ts

@@ -45,6 +45,7 @@ import { SDK_VERSION } from './version';
 
 const RECONNECT_MIN_DELAY = 1000;
 const RECONNECT_MAX_DELAY_DEFAULT = 60 * 5 * 1000; // 5 minutes in milliseconds (Case: 1858)
+const GET_CONNECT_TIMEOUT = 3 * 1000;
 const RECONNECT_MAX_DELAY_FOR_ADMINS = 30 * 1000; // 30 seconds for admin clients (likely to be a backend server)
 const RECONNECT_DELAY_MULTIPLIER = 1.3;
 const RECONNECT_DELAY_RESET_TIMEOUT = 30000; // Reset delay back to MIN_DELAY after being connected for 30sec.
@@ -76,6 +77,12 @@ interface OutstandingPut {
   onComplete: (a: string, b?: string) => void;
 }
 
+interface OutstandingGet {
+  action: string;
+  request: object;
+  onComplete: (response: { [k: string]: unknown }) => void;
+}
+
 /**
  * Firebase connection.  Abstracts wire protocol and handles reconnecting.
  *
@@ -94,7 +101,9 @@ export class PersistentConnection extends ServerActions {
     Map</* queryId */ string, ListenSpec>
   > = new Map();
   private outstandingPuts_: OutstandingPut[] = [];
+  private outstandingGets_: OutstandingGet[] = [];
   private outstandingPutCount_ = 0;
+  private outstandingGetCount_ = 0;
   private onDisconnectRequestQueue_: OnDisconnectRequest[] = [];
   private connected_ = false;
   private reconnectDelay_ = RECONNECT_MIN_DELAY;
@@ -186,15 +195,55 @@ export class PersistentConnection extends ServerActions {
   }
 
   get(query: Query): Promise<string> {
-    const req: { [k: string]: unknown } = {
+    const deferred = new Deferred<string>();
+    const request = {
       p: query.path.toString(),
       q: query.queryObject()
     };
+    const outstandingGet = {
+      action: 'g',
+      request,
+      onComplete: (message: { [k: string]: unknown }) => {
+        const payload = message['d'] as string;
+        if (message['s'] === 'ok') {
+          this.onDataUpdate_(
+            request['p'],
+            payload,
+            /*isMerge*/ false,
+            /*tag*/ null
+          );
+          deferred.resolve(payload);
+        } else {
+          deferred.reject(payload);
+        }
+      }
+    };
+    this.outstandingGets_.push(outstandingGet);
+    this.outstandingGetCount_++;
+    const index = this.outstandingGets_.length - 1;
+
+    if (!this.connected_) {
+      const self = this;
+      setTimeout(function () {
+        const get = self.outstandingGets_[index];
+        if (get === undefined || outstandingGet !== get) {
+          return;
+        }
+        delete self.outstandingGets_[index];
+        self.outstandingGetCount_--;
+        if (self.outstandingGetCount_ === 0) {
+          self.outstandingGets_ = [];
+        }
+        self.log_('get ' + index + ' timed out on connection');
+        deferred.reject(new Error('Client is offline.'));
+      }, GET_CONNECT_TIMEOUT);
+    }
+
     if (this.connected_) {
-      return this.sendGet_(req);
-    } else {
-      return Promise.reject(new Error('Client is offline'));
+      this.sendGet_(index);
     }
+
+    return deferred.promise;
   }
 
   /**
@@ -214,7 +263,7 @@ export class PersistentConnection extends ServerActions {
     }
     assert(
       query.getQueryParams().isDefault() ||
-        !query.getQueryParams().loadsAllData(),
+      !query.getQueryParams().loadsAllData(),
       'listen() called for non-default but complete query'
     );
     assert(
@@ -234,23 +283,22 @@ export class PersistentConnection extends ServerActions {
     }
   }
 
-  private sendGet_(request: object): Promise<string> {
-    return new Promise((resolve, reject) => {
-      this.sendRequest('g', request, (message: { [k: string]: unknown }) => {
-        const payload = message['d'] as string;
-        if (message['s'] === 'ok') {
-          this.onDataUpdate_(
-            request['p'],
-            payload,
-            /*isMerge*/ false,
-            /*tag*/ null
-          );
-          resolve(payload);
-        } else {
-          reject(payload);
+  private sendGet_(index: number) {
+    const get = this.outstandingGets_[index];
+    this.sendRequest(
+      get.action,
+      get.request,
+      (message: { [k: string]: unknown }) => {
+        delete this.outstandingGets_[index];
+        this.outstandingGetCount_--;
+        if (this.outstandingGetCount_ === 0) {
+          this.outstandingGets_ = [];
         }
-      });
-    });
+        if (get.onComplete) {
+          get.onComplete(message);
+        }
+      }
+    );
   }
 
   private sendListen_(listenSpec: ListenSpec) {
@@ -305,8 +353,8 @@ export class PersistentConnection extends ServerActions {
         const indexPath = query.path.toString();
         warn(
           `Using an unspecified index. Your data will be downloaded and ` +
-            `filtered on the client. Consider adding ${indexSpec} at ` +
-            `${indexPath} to your security rules for better performance.`
+          `filtered on the client. Consider adding ${indexSpec} at ` +
+          `${indexPath} to your security rules for better performance.`
         );
       }
     }
@@ -324,7 +372,7 @@ export class PersistentConnection extends ServerActions {
       //If we're connected we want to let the server know to unauthenticate us. If we're not connected, simply delete
       //the credential so we dont become authenticated next time we connect.
       if (this.connected_) {
-        this.sendRequest('unauth', {}, () => {});
+        this.sendRequest('unauth', {}, () => { });
       }
     }
 
@@ -388,7 +436,7 @@ export class PersistentConnection extends ServerActions {
 
     assert(
       query.getQueryParams().isDefault() ||
-        !query.getQueryParams().loadsAllData(),
+      !query.getQueryParams().loadsAllData(),
       'unlisten() called for non-default but complete query'
     );
     const listen = this.removeListen_(pathString, queryId);
@@ -646,8 +694,8 @@ export class PersistentConnection extends ServerActions {
     } else {
       error(
         'Unrecognized action received from server: ' +
-          stringify(action) +
-          '\nAre you using the latest client?'
+        stringify(action) +
+        '\nAre you using the latest client?'
       );
     }
   }
@@ -982,6 +1030,12 @@ export class PersistentConnection extends ServerActions {
         request.onComplete
       );
     }
+
+    for (let i = 0; i < this.outstandingGets_.length; i++) {
+      if (this.outstandingGets_[i]) {
+        this.sendGet_(i);
+      }
+    }
   }
 
   /**

packages/database/src/core/Repo.ts

@@ -297,7 +297,7 @@ export class Repo {
     return this.nextWriteId_++;
   }
 
-  get(query: Query): Promise<DataSnapshot> {
+  getValue(query: Query): Promise<DataSnapshot> {
     return this.server_.get(query).then(
       payload => {
         const node = nodeFromJSON(payload as string);