Support async sync change listeners (#1626)

docs/sync.js

@@ -119,9 +119,10 @@ class Sync {
  * Change information passed when receiving sync `'change'` events.
  *
  * A ChangeEvent object can only be used within the callback which it is
- * supplied to, and cannot be stored for use later. In particular, this means
- * that async functions cannot be used within the callback prior to accessing
- * the change event. The Realms supplied by the change event do not need to be
+ * supplied to, and cannot be stored for use later. If the callback returns a
+ * promise, the ChangeEvent will remain valid until that promise is resolved
+ * (and no further notifications for that same Realm will be made until it is
+ * resolved). The Realms supplied by the change event do not need to be
  * explicitly closed.
  *
  * @memberof Realm.Sync

lib/index.d.ts

@@ -408,7 +408,8 @@ declare namespace Realm.Sync {
     }
 
     function addListener(serverURL: string, adminUser: Realm.Sync.User, regex: string, name: string, changeCallback: (changeEvent: ChangeEvent) => void): void;
-    function removeAllListeners(name?: string): void;
+    function addListener(serverURL: string, adminUser: Realm.Sync.User, regex: string, name: string, changeCallback: (changeEvent: ChangeEvent) => Promise<void>): void;
+    function removeAllListeners(): Promise<void>;
     function removeListener(regex: string, name: string, changeCallback: (changeEvent: ChangeEvent) => void): void;
     function setLogLevel(logLevel: 'all' | 'trace' | 'debug' | 'detail' | 'info' | 'warn' | 'error' | 'fatal' | 'off'): void;
     function initiateClientReset(path: string): void;

lib/notifier.js

@@ -57,12 +57,14 @@ class FunctionListener {
 
     onchange(changes) {
         if (this.event !== 'change' || !this.regex.test(changes.path)) {
+            changes.release();
             return;
         }
         if (changes.isEmpty) {
+            changes.release();
             return;
         }
-        this.fn(changes);
+        Promise.resolve(this.fn(changes)).then(() => changes.release());
     }
 };
 
@@ -125,12 +127,17 @@ class Listener {
         if (!changes) {
             return;
         }
+
+        let refCount = this.callbacks.length;
+        changes.release = () => {
+            if (--refCount === 0) {
+                changes.close();
+            }
+        }
+
         for (const callback of this.callbacks) {
             callback.onchange(changes);
         }
-        if (!changes.refCount) {
-            changes.close();
-        }
     }
 
     available(virtualPath) {

lib/worker.js

@@ -80,9 +80,7 @@ class Worker {
             if (m.change) {
                 const changeObj = this._changeObjects[m.change];
                 delete this._changeObjects[m.change];
-                if (--changeObj.refCount === 0) {
-                    changeObj.close();
-                }
+                changeObj.release();
             }
             this._waiting.push(child);
             this._next();