Don't support concurrent waiters

Author: simolus3

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -215,7 +215,7 @@ export abstract class AbstractStreamingSyncImplementation
   protected options: AbstractStreamingSyncImplementationOptions;
   protected abortController: AbortController | null;
   protected crudUpdateListener?: () => void;
-  protected streamingSyncPromise?: Promise<void>;
+  protected streamingSyncPromise?: Promise<[void, void]>;
   protected logger: ILogger;
   private activeStreams: SubscribedStream[];
   private connectionMayHaveChanged = false;
@@ -409,7 +409,10 @@ The next upload iteration will be delayed.`);
 
     const controller = new AbortController();
     this.abortController = controller;
-    this.streamingSyncPromise = this.streamingSync(this.abortController.signal, options);
+    this.streamingSyncPromise = Promise.all([
+      this.crudUploadLoop(controller.signal).catch((ex) => this.logger.error('Error in crud upload loop', ex)),
+      this.streamingSync(controller.signal, options)
+    ]);
 
     // Return a promise that resolves when the connection status is updated to indicate that we're connected.
     return new Promise<void>((resolve) => {
@@ -484,8 +487,6 @@ The next upload iteration will be delayed.`);
       });
     });
 
-    this.crudUploadLoop(signal).catch((ex) => this.logger.error('Error in crud upload loop', ex));
-
     /**
      * This loops runs until [retry] is false or the abort signal is set to aborted.
      * Aborting the nestedAbortController will:

packages/common/src/utils/async.ts

@@ -27,26 +27,31 @@ export interface AsyncNotifier {
   waitForNotification(signal: AbortSignal): Promise<void>;
 
   /**
-   * Notifies any pending listener, or makes the next {@link waitForNotification} complete immediately if no listener
+   * Notifies a pending listener, or makes the next {@link waitForNotification} complete immediately if no listener
    * is currently active.
    */
   notify(): void;
 }
 
 export function asyncNotifier(): AsyncNotifier {
-  let waitingConsumers: (() => void)[] = [];
+  let waitingConsumer: (() => void) | null = null;
   let hasPendingNotification = false;
 
   return {
     notify() {
-      if (waitingConsumers.length > 0) {
-        waitingConsumers.splice(0, 1)[0]();
+      if (waitingConsumer != null) {
+        waitingConsumer();
+        waitingConsumer = null;
       } else {
         hasPendingNotification = true;
       }
     },
     waitForNotification(signal: AbortSignal) {
       return new Promise((resolve) => {
+        if (waitingConsumer != null) {
+          throw new Error('Illegal call to waitForNotification, already has a waiter.');
+        }
+
         if (signal.aborted) {
           resolve();
         } else if (hasPendingNotification) {
@@ -59,14 +64,11 @@ export function asyncNotifier(): AsyncNotifier {
           }
 
           function onAbort() {
-            const i = waitingConsumers.indexOf(complete);
-            if (i > -1) {
-              waitingConsumers.splice(i, 1);
-            }
+            waitingConsumer = null;
             resolve();
           }
 
-          waitingConsumers.push(complete);
+          waitingConsumer = complete;
           signal.addEventListener('abort', onAbort);
         }
       });