wallet-core: unify handling of run-until-done, simplify waiter implementation

1 files changed, 124 insertions, 173 deletions

diff --git a/packages/taler-wallet-core/src/testing.ts b/packages/taler-wallet-core/src/testing.ts
index 32c0765b4..2f149cfa8 100644
--- a/packages/taler-wallet-core/src/testing.ts
+++ b/packages/taler-wallet-core/src/testing.ts
@@ -39,8 +39,6 @@ import {
   j2s,
   Logger,
   NotificationType,
-  OpenedPromise,
-  openPromise,
   parsePaytoUri,
   PreparePayResultType,
   TalerCorebankApiClient,
@@ -58,6 +56,7 @@ import {
   readSuccessResponseJsonOrThrow,
 } from "@gnu-taler/taler-util/http";
 import { getBalances } from "./balance.js";
+import { genericWaitForState } from "./common.js";
 import { createDepositGroup } from "./deposits.js";
 import { fetchFreshExchange } from "./exchanges.js";
 import {
@@ -402,52 +401,56 @@ export async function waitUntilAllTransactionsFinal(
   wex: WalletExecutionContext,
 ): Promise<void> {
   logger.info("waiting until all transactions are in a final state");
-  wex.taskScheduler.ensureRunning();
-  let p: OpenedPromise<void> | undefined = undefined;
-  const cancelNotifs = wex.ws.addNotificationListener((notif) => {
-    if (!p) {
-      return;
-    }
-    if (notif.type === NotificationType.TransactionStateTransition) {
+  await wex.taskScheduler.ensureRunning();
+  await genericWaitForState(wex, {
+    filterNotification(notif) {
+      if (notif.type !== NotificationType.TransactionStateTransition) {
+        return false;
+      }
       switch (notif.newTxState.major) {
         case TransactionMajorState.Pending:
         case TransactionMajorState.Aborting:
-          break;
+          return false;
         default:
-          p.resolve();
+          return true;
       }
-    }
-  });
-  while (1) {
-    p = openPromise();
-    const txs = await getTransactions(wex, {
-      includeRefreshes: true,
-      filterByState: "nonfinal",
-    });
-    let finished = true;
-    for (const tx of txs.transactions) {
-      switch (tx.txState.major) {
-        case TransactionMajorState.Pending:
-        case TransactionMajorState.Aborting:
-        case TransactionMajorState.Suspended:
-        case TransactionMajorState.SuspendedAborting:
-          finished = false;
-          logger.info(
-            `continuing waiting, ${tx.transactionId} in ${tx.txState.major}(${tx.txState.minor})`,
-          );
-          break;
+    },
+    async checkState() {
+      const txs = await getTransactions(wex, {
+        includeRefreshes: true,
+        filterByState: "nonfinal",
+      });
+      for (const tx of txs.transactions) {
+        switch (tx.txState.major) {
+          case TransactionMajorState.Pending:
+          case TransactionMajorState.Aborting:
+          case TransactionMajorState.Suspended:
+          case TransactionMajorState.SuspendedAborting:
+            logger.info(
+              `continuing waiting, ${tx.transactionId} in ${tx.txState.major}(${tx.txState.minor})`,
+            );
+            return false;
+        }
       }
-    }
-    if (finished) {
-      break;
-    }
-    // Wait until transaction state changed
-    await p.promise;
-  }
-  cancelNotifs();
+      return true;
+    },
+  });
   logger.info("done waiting until all transactions are in a final state");
 }
 
+export async function waitTasksDone(
+  wex: WalletExecutionContext,
+): Promise<void> {
+  await genericWaitForState(wex, {
+    async checkState() {
+      return wex.taskScheduler.isIdle();
+    },
+    filterNotification(notif) {
+      return notif.type === NotificationType.Idle;
+    },
+  });
+}
+
 /**
  * Wait until all chosen transactions are in a final state.
  */
@@ -462,59 +465,51 @@ export async function waitUntilGivenTransactionsFinal(
   if (transactionIds.length === 0) {
     return;
   }
-  wex.taskScheduler.ensureRunning();
+
   const txIdSet = new Set(transactionIds);
-  let p: OpenedPromise<void> | undefined = undefined;
-  const cancelNotifs = wex.ws.addNotificationListener((notif) => {
-    if (!p) {
-      return;
-    }
-    if (notif.type === NotificationType.TransactionStateTransition) {
+
+  await genericWaitForState(wex, {
+    filterNotification(notif) {
+      if (notif.type !== NotificationType.TransactionStateTransition) {
+        return false;
+      }
       if (!txIdSet.has(notif.transactionId)) {
-        return;
+        return false;
       }
       switch (notif.newTxState.major) {
         case TransactionMajorState.Pending:
         case TransactionMajorState.Aborting:
         case TransactionMajorState.Suspended:
         case TransactionMajorState.SuspendedAborting:
-          break;
-        default:
-          p.resolve();
-      }
-    }
-  });
-  while (1) {
-    p = openPromise();
-    const txs = await getTransactions(wex, {
-      includeRefreshes: true,
-      filterByState: "nonfinal",
-    });
-    let finished = true;
-    for (const tx of txs.transactions) {
-      if (!txIdSet.has(tx.transactionId)) {
-        // Don't look at this transaction, we're not interested in it.
-        continue;
+          return false;
       }
-      switch (tx.txState.major) {
-        case TransactionMajorState.Pending:
-        case TransactionMajorState.Aborting:
-        case TransactionMajorState.Suspended:
-        case TransactionMajorState.SuspendedAborting:
-          finished = false;
-          logger.info(
-            `continuing waiting, ${tx.transactionId} in ${tx.txState.major}(${tx.txState.minor})`,
-          );
-          break;
+      return true;
+    },
+    async checkState() {
+      const txs = await getTransactions(wex, {
+        includeRefreshes: true,
+        filterByState: "nonfinal",
+      });
+      for (const tx of txs.transactions) {
+        if (!txIdSet.has(tx.transactionId)) {
+          // Don't look at this transaction, we're not interested in it.
+          continue;
+        }
+        switch (tx.txState.major) {
+          case TransactionMajorState.Pending:
+          case TransactionMajorState.Aborting:
+          case TransactionMajorState.Suspended:
+          case TransactionMajorState.SuspendedAborting:
+            logger.info(
+              `continuing waiting, ${tx.transactionId} in ${tx.txState.major}(${tx.txState.minor})`,
+            );
+            return false;
+        }
       }
-    }
-    if (finished) {
-      break;
-    }
-    // Wait until transaction state changed
-    await p.promise;
-  }
-  cancelNotifs();
+      // No transaction is pending, we're done waiting!
+      return true;
+    },
+  });
   logger.info("done waiting until given transactions are in a final state");
 }
 
@@ -522,52 +517,43 @@ export async function waitUntilRefreshesDone(
   wex: WalletExecutionContext,
 ): Promise<void> {
   logger.info("waiting until all refresh transactions are in a final state");
-  wex.taskScheduler.ensureRunning();
-  let p: OpenedPromise<void> | undefined = undefined;
-  const cancelNotifs = wex.ws.addNotificationListener((notif) => {
-    if (!p) {
-      return;
-    }
-    if (notif.type === NotificationType.TransactionStateTransition) {
+
+  await genericWaitForState(wex, {
+    filterNotification(notif) {
+      if (notif.type !== NotificationType.TransactionStateTransition) {
+        return false;
+      }
       switch (notif.newTxState.major) {
         case TransactionMajorState.Pending:
         case TransactionMajorState.Aborting:
-          break;
+          return false;
         default:
-          p.resolve();
-      }
-    }
-  });
-  while (1) {
-    p = openPromise();
-    const txs = await getTransactions(wex, {
-      includeRefreshes: true,
-      filterByState: "nonfinal",
-    });
-    let finished = true;
-    for (const tx of txs.transactions) {
-      if (tx.type !== TransactionType.Refresh) {
-        continue;
+          return true;
       }
-      switch (tx.txState.major) {
-        case TransactionMajorState.Pending:
-        case TransactionMajorState.Aborting:
-        case TransactionMajorState.Suspended:
-        case TransactionMajorState.SuspendedAborting:
-          finished = false;
-          logger.info(
-            `continuing waiting, ${tx.transactionId} in ${tx.txState.major}(${tx.txState.minor})`,
-          );
-          break;
+    },
+    async checkState() {
+      const txs = await getTransactions(wex, {
+        includeRefreshes: true,
+        filterByState: "nonfinal",
+      });
+      for (const tx of txs.transactions) {
+        if (tx.type !== TransactionType.Refresh) {
+          continue;
+        }
+        switch (tx.txState.major) {
+          case TransactionMajorState.Pending:
+          case TransactionMajorState.Aborting:
+          case TransactionMajorState.Suspended:
+          case TransactionMajorState.SuspendedAborting:
+            logger.info(
+              `continuing waiting, ${tx.transactionId} in ${tx.txState.major}(${tx.txState.minor})`,
+            );
+            return false;
+        }
       }
-    }
-    if (finished) {
-      break;
-    }
-    // Wait until transaction state changed
-    await p.promise;
-  }
-  cancelNotifs();
+      return true;
+    },
+  });
   logger.info("done waiting until all refreshes are in a final state");
 }
 
@@ -575,33 +561,10 @@ async function waitUntilTransactionPendingReady(
   wex: WalletExecutionContext,
   transactionId: string,
 ): Promise<void> {
-  logger.info(`starting waiting for ${transactionId} to be in pending(ready)`);
-  wex.taskScheduler.ensureRunning();
-  let p: OpenedPromise<void> | undefined = undefined;
-  const cancelNotifs = wex.ws.addNotificationListener((notif) => {
-    if (!p) {
-      return;
-    }
-    if (notif.type === NotificationType.TransactionStateTransition) {
-      p.resolve();
-    }
+  return await waitTransactionState(wex, transactionId, {
+    major: TransactionMajorState.Pending,
+    minor: TransactionMinorState.Ready,
   });
-  while (1) {
-    p = openPromise();
-    const tx = await getTransactionById(wex, {
-      transactionId,
-    });
-    if (
-      tx.txState.major == TransactionMajorState.Pending &&
-      tx.txState.minor === TransactionMinorState.Ready
-    ) {
-      break;
-    }
-    // Wait until transaction state changed
-    await p.promise;
-  }
-  logger.info(`done waiting for ${transactionId} to be in pending(ready)`);
-  cancelNotifs();
 }
 
 /**
@@ -617,34 +580,22 @@ export async function waitTransactionState(
       txState,
     )})`,
   );
-  wex.taskScheduler.ensureRunning();
-  let p: OpenedPromise<void> | undefined = undefined;
-  const cancelNotifs = wex.ws.addNotificationListener((notif) => {
-    if (!p) {
-      return;
-    }
-    if (notif.type === NotificationType.TransactionStateTransition) {
-      p.resolve();
-    }
+  await genericWaitForState(wex, {
+    async checkState() {
+      const tx = await getTransactionById(wex, {
+        transactionId,
+      });
+      return (
+        tx.txState.major === txState.major && tx.txState.minor === txState.minor
+      );
+    },
+    filterNotification(notif) {
+      return notif.type === NotificationType.TransactionStateTransition;
+    },
   });
-  while (1) {
-    p = openPromise();
-    const tx = await getTransactionById(wex, {
-      transactionId,
-    });
-    if (
-      tx.txState.major === txState.major &&
-      tx.txState.minor === txState.minor
-    ) {
-      break;
-    }
-    // Wait until transaction state changed
-    await p.promise;
-  }
   logger.info(
     `done waiting for ${transactionId} to be in ${JSON.stringify(txState)}`,
   );
-  cancelNotifs();
 }
 
 export async function waitUntilTransactionWithAssociatedRefreshesFinal(
@@ -669,7 +620,7 @@ export async function runIntegrationTest2(
   wex: WalletExecutionContext,
   args: IntegrationTestV2Args,
 ): Promise<void> {
-  wex.taskScheduler.ensureRunning();
+  await wex.taskScheduler.ensureRunning();
   logger.info("running test with arguments", args);
 
   const exchangeInfo = await fetchFreshExchange(wex, args.exchangeBaseUrl);