Wait for pending notifications to complete when removing a sync listener (#1648)

With async callbacks, in-process listeners need to be able to wait for anything
currently in progress to complete just like out-of-process listeners. In
addition, removeListener() needs to be declared as returning a Promise so that
typescript code can actually use it.

lib/index.d.ts

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

lib/notifier.js

@@ -34,10 +34,11 @@ class FunctionListener {
         this.event = event;
         this.fn = fn;
         this.seen = {};
+        this.pending = [];
     }
 
     stop() {
-        return Promise.resolve();
+        return Promise.all(this.pending);
     }
 
     matches(regex, event, fn) {
@@ -64,7 +65,12 @@ class FunctionListener {
             changes.release();
             return;
         }
-        Promise.resolve(this.fn(changes)).then(() => changes.release());
+        const promise = Promise.resolve(this.fn(changes));
+        this.pending.push(promise);
+        promise.then(() => {
+            changes.release();
+            this.pending.splice(this.pending.indexOf(promise), 1);
+        });
     }
 };
 
@@ -128,7 +134,7 @@ class Listener {
             return;
         }
 
-        let refCount = this.callbacks.length;
+        let refCount = 1;
         changes.release = () => {
             if (--refCount === 0) {
                 changes.close();
@@ -136,8 +142,10 @@ class Listener {
         }
 
         for (const callback of this.callbacks) {
+            ++refCount;
             callback.onchange(changes);
         }
+        changes.release();
     }
 
     available(virtualPath) {