feat: add AsyncIterable support to stream() (fixes #426)

.README/QUERY_METHODS.md

@@ -189,6 +189,27 @@ await connection.stream(sql.typeAlias('foo')`SELECT foo`, (stream) => {
 });
 ```
 
+You can also using the [AsyncIterable](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/AsyncIterator) interface:
+
+```ts
+await connection.stream(sql.typeAlias('foo')`SELECT foo`, async (stream) => {
+  for await (const row of stream) {
+    row;
+    // {
+    //   data: {
+    //     foo: 'bar'
+    //   },
+    //   fields: [
+    //     {
+    //       name: 'foo',
+    //       dataTypeId: 23,
+    //     }
+    //   ]
+    // }
+  }
+});
+```
+
 ### `transaction`
 
 `transaction` method is used wrap execution of queries in `START TRANSACTION` and `COMMIT` or `ROLLBACK`. `COMMIT` is called if the transaction handler returns a promise that resolves; `ROLLBACK` is called otherwise.

README.md

@@ -2486,6 +2486,27 @@ await connection.stream(sql.typeAlias('foo')`SELECT foo`, (stream) => {
 });
 ```
 
+You can also using the [AsyncIterable](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/AsyncIterator) interface:
+
+```ts
+await connection.stream(sql.typeAlias('foo')`SELECT foo`, async (stream) => {
+  for await (const row of stream) {
+    row;
+    // {
+    //   data: {
+    //     foo: 'bar'
+    //   },
+    //   fields: [
+    //     {
+    //       name: 'foo',
+    //       dataTypeId: 23,
+    //     }
+    //   ]
+    // }
+  }
+});
+```
+
 <a name="user-content-slonik-query-methods-transaction"></a>
 <a name="slonik-query-methods-transaction"></a>
 ### <code>transaction</code>

src/types.ts

@@ -57,15 +57,16 @@ export type QueryId = string;
 
 export type MaybePromise<T> = Promise<T> | T;
 
+type StreamDataEvent<T> = { data: T; fields: readonly Field[] };
+
 // eslint-disable-next-line @typescript-eslint/consistent-type-definitions
 interface TypedReadable<T> extends Readable {
   // eslint-disable-next-line @typescript-eslint/method-signature-style
-  on(
-    event: 'data',
-    listener: (chunk: { data: T; fields: readonly Field[] }) => void,
-  ): this;
+  on(event: 'data', listener: (chunk: StreamDataEvent<T>) => void): this;
   // eslint-disable-next-line @typescript-eslint/method-signature-style
   on(event: string | symbol, listener: (...args: any[]) => void): this;
+
+  [Symbol.asyncIterator]: () => AsyncIterableIterator<StreamDataEvent<T>>;
 }
 
 export type StreamHandler<T> = (stream: TypedReadable<T>) => void;

test/slonik/integration/stream.ts

@@ -189,6 +189,37 @@ test('streams rows (check types)', async (t) => {
   await pool.end();
 });
 
+test('streams rows using AsyncIterator', async (t) => {
+  const pool = await createPool(t.context.dsn);
+
+  await pool.query(sql.unsafe`
+    INSERT INTO person (name)
+    VALUES ('foo'), ('bar'), ('baz')
+  `);
+
+  const names: string[] = [];
+
+  await pool.stream(
+    sql.type(
+      z.object({
+        name: z.string(),
+      }),
+    )`
+      SELECT name
+      FROM person
+    `,
+    async (stream) => {
+      for await (const row of stream) {
+        names.push(row.data.name);
+      }
+    },
+  );
+
+  t.deepEqual(names, ['foo', 'bar', 'baz']);
+
+  await pool.end();
+});
+
 test('streams rows with different batchSize', async (t) => {
   const pool = await createPool(t.context.dsn);