replace streamToPromise with Array.fromAsync (#644)

apps/webapp/package.json

@@ -33,6 +33,7 @@
     "zustand": "^4.5.1"
   },
   "devDependencies": {
+    "@penumbra-zone/polyfills": "workspace:*",
     "@testing-library/jest-dom": "^6.4.2",
     "@testing-library/react": "^14.2.1",
     "@types/node": "^20.11.22",

apps/webapp/src/fetchers/assets.ts

@@ -1,9 +1,9 @@
-import { streamToPromise } from './stream';
+import Array from '@penumbra-zone/polyfills/Array.fromAsync';
 import { AssetsRequest } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/view/v1/view_pb';
 import { viewClient } from '../clients';
 
 export const getAllAssets = () => {
   const req = new AssetsRequest();
   const iterable = viewClient.assets(req);
-  return streamToPromise(iterable);
+  return Array.fromAsync(iterable);
 };

apps/webapp/src/fetchers/balances/index.ts

@@ -2,7 +2,7 @@ import { BalancesRequest } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbr
 import { AssetId } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/asset/v1/asset_pb';
 import { AddressIndex } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/keys/v1/keys_pb';
 import { viewClient } from '../../clients';
-import { streamToPromise } from '../stream';
+import Array from '@penumbra-zone/polyfills/Array.fromAsync';
 
 interface BalancesProps {
   accountFilter?: AddressIndex;
@@ -15,5 +15,5 @@ export const getBalances = ({ accountFilter, assetIdFilter }: BalancesProps = {}
   if (assetIdFilter) req.assetIdFilter = assetIdFilter;
 
   const iterable = viewClient.balances(req);
-  return streamToPromise(iterable);
+  return Array.fromAsync(iterable);
 };

apps/webapp/src/fetchers/stream.ts

@@ -51,24 +51,3 @@ export const useStream = <T>(query: AsyncIterable<T>): StreamQueryResult<T> => {
 export const useCollectedStream = <T>(query: AsyncIterable<T>): CollectedStreamQueryResult<T> => {
   return useStreamCommon(query, [] as T[], (prevData, newData) => [...prevData, newData]);
 };
-
-// Meant to convert a stream into a promise of the completed result
-// Note: If the stream is unending, this will not resolve.
-//       This is only useful if you are collecting all of the fixed set of results together.
-export const streamToPromise = <T>(query: AsyncIterable<T>): Promise<T[]> => {
-  return new Promise<T[]>((resolve, reject) => {
-    void (async function () {
-      const result: T[] = [];
-      try {
-        for await (const res of query) {
-          result.push(res);
-        }
-        resolve(result);
-      } catch (e) {
-        if (e instanceof Error) reject(e);
-        else if (typeof e === 'string') reject(new Error(e));
-        else reject(new Error('Unknown error in `streamToPromise`'));
-      }
-    })();
-  });
-};

apps/webapp/src/fetchers/transactions.ts

@@ -1,5 +1,5 @@
 import { viewClient } from '../clients';
-import { streamToPromise } from './stream';
+import Array from '@penumbra-zone/polyfills/Array.fromAsync';
 import { getTransactionClassificationLabel, uint8ArrayToHex } from '@penumbra-zone/types';
 
 export interface TransactionSummary {
@@ -9,7 +9,7 @@ export interface TransactionSummary {
 }
 
 export const getAllTransactions = async (): Promise<TransactionSummary[]> => {
-  const responses = await streamToPromise(viewClient.transactionInfo({}));
+  const responses = await Array.fromAsync(viewClient.transactionInfo({}));
   return responses
     .map(tx => {
       return {

apps/webapp/src/fetchers/unclaimed-swaps.ts

@@ -1,9 +1,9 @@
 import { viewClient } from '../clients';
-import { streamToPromise } from './stream';
+import Array from '@penumbra-zone/polyfills/Array.fromAsync';
 import { getUnclaimedSwaps } from '@penumbra-zone/getters';
 import { SwapRecord } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/view/v1/view_pb';
 
 export const fetchUnclaimedSwaps = async (): Promise<SwapRecord[]> => {
-  const responses = await streamToPromise(viewClient.unclaimedSwaps({}));
+  const responses = await Array.fromAsync(viewClient.unclaimedSwaps({}));
   return responses.map(getUnclaimedSwaps);
 };

packages/polyfills/streamToPromise.test.ts

@@ -0,0 +1,53 @@
+import { beforeEach, describe, expect, test } from 'vitest';
+
+import Array from './Array.fromAsync';
+
+// eslint-disable-next-line @typescript-eslint/unbound-method
+const streamToPromise = Array.fromAsync;
+
+describe('streamToPromise()', () => {
+  describe('when one of the streamed items throws', () => {
+    let error: unknown;
+    const query = async function* () {
+      yield* [
+        await new Promise(() => {
+          throw error;
+        }),
+      ];
+    };
+
+    describe('when the thrown value is an instance of `Error`', () => {
+      beforeEach(() => {
+        error = new Error('oops');
+      });
+
+      test('rejects with the error', async () => {
+        await expect(streamToPromise(query())).rejects.toThrow(error as Error);
+      });
+    });
+
+    describe('old streamToPromise behavior that Array.fromAsync does not exhibit', () => {
+      describe('when the thrown value is a string', () => {
+        beforeEach(() => {
+          error = 'oops';
+        });
+
+        test.fails("don't reject with the string wrapped in an instance of `Error`", async () => {
+          await expect(streamToPromise(query())).rejects.toThrow(new Error('oops'));
+        });
+      });
+
+      describe('when the thrown value is neither an `Error` instance nor a string', () => {
+        beforeEach(() => {
+          error = 1n;
+        });
+
+        test.fails("don't reject with an unknown error", async () => {
+          await expect(streamToPromise(query())).rejects.toThrow(
+            new Error('Unknown error in `streamToPromise`'),
+          );
+        });
+      });
+    });
+  });
+});