1 files changed, 62 insertions, 0 deletions
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
index 69f0e5f0b..fb0bd9f62 100644
--- a/packages/taler-wallet-core/src/wallet.ts
+++ b/packages/taler-wallet-core/src/wallet.ts
@@ -1916,6 +1916,16 @@ class WalletDbTriggerSpec implements TriggerSpec {
   }
 }
 
+type LongpollRunFn<T> = (timeoutMs: number) => Promise<T>;
+type ResolveFn = () => void;
+
+/**
+ * Per-hostname state for longpolling.
+ */
+interface LongpollState {
+  queue: Array<ResolveFn>;
+}
+
 /**
  * Internal state of the wallet.
  *
@@ -1977,6 +1987,58 @@ export class InternalWalletState {
     return this._dbAccessHandle;
   }
 
+  private longpollStatePerHostname: Map<string, LongpollState> = new Map();
+
+  private longpollRequestIdCounter = 1;
+
+  /**
+   * Run a long-polling request, potentially queueing the request
+   * if too many other long-polling requests against the same hostname
+   * (or too many overall) are active.
+   */
+  async runLongpollQueueing<T>(
+    hostname: string,
+    f: LongpollRunFn<T>,
+  ): Promise<T> {
+    let rid = this.longpollRequestIdCounter++;
+    const doRun: () => Promise<T> = async () => {
+      const st = this.longpollStatePerHostname.get(hostname);
+      const numWaiting = st?.queue.length ?? 0;
+      logger.info(
+        `running long-poll ${rid} to ${hostname} with ${numWaiting} waiting`,
+      );
+      try {
+        const timeoutMs = Math.round(Math.max(10000, 30000 / (numWaiting + 1)));
+        return await f(timeoutMs);
+      } finally {
+        logger.info(
+          `cleaning up after long-poll ${rid} request to ${hostname}`,
+        );
+        if (st) {
+          const next = st.queue.shift();
+          if (next) {
+            next();
+          } else {
+            this.longpollStatePerHostname.delete(hostname);
+          }
+        }
+      }
+    };
+    const state = this.longpollStatePerHostname.get(hostname);
+    if (state) {
+      logger.info(`long-poll request ${rid} to ${hostname} queued`);
+      const promcap = openPromise<void>();
+      state.queue.push(promcap.resolve);
+      return promcap.promise.then(doRun);
+    } else {
+      logger.info(`directly running long-poll request ${rid} to ${hostname}`);
+      this.longpollStatePerHostname.set(hostname, {
+        queue: [],
+      });
+      return Promise.resolve().then(doRun);
+    }
+  }
+
   devExperimentState: DevExperimentState = {};
 
   clientCancellationMap: Map<string, CancellationToken.Source> = new Map();