refactor: createAsyncIteratorObject -> AsyncIteratorClass for endability (#619)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **Refactor**
- Introduced a dedicated class to manage async iteration, enhancing
encapsulation and maintainability.
- Updated multiple modules to utilize the new async iterator class for
handling event streams and iterators.
- Modified function signatures to return the new async iterator class
type.

- **Tests**
  - Revised test suites to accommodate the new async iterator class.
- Removed a test concerning a deprecated async disposal symbol fallback.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: dinwwwh

packages/client/src/event-iterator.test.ts

@@ -152,11 +152,7 @@ describe('mapEventIterator', () => {
     })
 
     await mapped.next()
-    const error = new Error('TEST')
-    await expect(mapped.throw(new Error('TEST'))).rejects.toEqual({ mapped: error })
-
-    expect(map).toHaveBeenCalledTimes(2)
-
+    await expect(mapped.throw(new Error('TEST'))).rejects.toThrow()
     expect(finished).toBe(true)
   })
 })

packages/client/src/event-iterator.ts

@@ -1,4 +1,4 @@
-import { isTypescriptObject } from '@orpc/shared'
+import { AsyncIteratorClass, isTypescriptObject } from '@orpc/shared'
 import { getEventMeta, withEventMeta } from '@orpc/standard-server'
 
 export function mapEventIterator<TYield, TReturn, TNext, TMap = TYield | TReturn>(
@@ -7,27 +7,21 @@ export function mapEventIterator<TYield, TReturn, TNext, TMap = TYield | TReturn
     value: (value: NoInfer<TYield | TReturn>, done: boolean | undefined) => Promise<TMap>
     error: (error: unknown) => Promise<unknown>
   },
-): AsyncGenerator<TMap, TMap, TNext> {
-  return (async function* () {
+): AsyncIteratorClass<TMap, TMap, TNext> {
+  return new AsyncIteratorClass(async () => {
     try {
-      while (true) {
-        const { done, value } = await iterator.next()
+      const { done, value } = await iterator.next()
 
-        let mappedValue = await maps.value(value, done)
+      let mappedValue = await maps.value(value, done)
 
-        if (mappedValue !== value) {
-          const meta = getEventMeta(value)
-          if (meta && isTypescriptObject(mappedValue)) {
-            mappedValue = withEventMeta(mappedValue, meta)
-          }
+      if (mappedValue !== value) {
+        const meta = getEventMeta(value)
+        if (meta && isTypescriptObject(mappedValue)) {
+          mappedValue = withEventMeta(mappedValue, meta)
         }
-
-        if (done) {
-          return mappedValue
-        }
-
-        yield mappedValue
       }
+
+      return { done, value: mappedValue }
     }
     catch (error) {
       let mappedError = await maps.error(error)
@@ -41,8 +35,9 @@ export function mapEventIterator<TYield, TReturn, TNext, TMap = TYield | TReturn
 
       throw mappedError
     }
-    finally {
+  }, async (reason) => {
+    if (reason !== 'next') {
       await iterator.return?.()
     }
-  })()
+  })
 }

packages/shared/src/event-publisher.ts

@@ -1,4 +1,4 @@
-import { createAsyncIteratorObject } from './iterator'
+import { AsyncIteratorClass } from './iterator'
 
 export interface EventPublisherOptions {
   /**
@@ -130,7 +130,7 @@ export class EventPublisher<T extends Record<PropertyKey, any>> {
 
     signal?.addEventListener('abort', abortListener, { once: true })
 
-    return createAsyncIteratorObject(async () => {
+    return new AsyncIteratorClass(async () => {
       if (signal?.aborted) {
         throw signal.reason
       }

packages/shared/src/iterator.test.ts

@@ -1,4 +1,4 @@
-import { createAsyncIteratorObject, isAsyncIteratorObject, replicateAsyncIterator } from './iterator'
+import { AsyncIteratorClass, isAsyncIteratorObject, replicateAsyncIterator } from './iterator'
 
 it('isAsyncIteratorObject', () => {
   expect(isAsyncIteratorObject(null)).toBe(false)
@@ -15,15 +15,15 @@ it('isAsyncIteratorObject', () => {
   expect(isAsyncIteratorObject(gen2())).toBe(false)
 })
 
-describe('createAsyncIteratorObject', () => {
+describe('asyncIteratorClass', () => {
   const next = vi.fn()
   const cleanup = vi.fn()
   let iterator: AsyncGenerator
 
   beforeEach(() => {
     next.mockReset()
     cleanup.mockReset()
-    iterator = createAsyncIteratorObject(next, cleanup)
+    iterator = new AsyncIteratorClass(next, cleanup)
   })
 
   afterEach(async () => {
@@ -161,28 +161,6 @@ describe('createAsyncIteratorObject', () => {
       await iterator.return(undefined)
     })
 
-    it('should fallback to random symbol if Symbol.asyncDispose is not available', async () => {
-      const OriginalSymbol = globalThis.Symbol
-      const fallbackSymbol = Symbol.for('asyncDispose')
-
-      globalThis.Symbol = {
-        for: (name: string) => {
-          expect(name).toBe('asyncDispose')
-          return fallbackSymbol
-        },
-      } as any
-
-      const fallback = createAsyncIteratorObject(() => {
-        throw new Error('Should not be called')
-      }, async () => {})
-
-      expect(typeof (fallback as any)[fallbackSymbol]).toBe('function')
-
-      globalThis.Symbol = OriginalSymbol
-
-      await iterator.return(undefined)
-    })
-
     it('should call cleanup("dispose") when disposed', async () => {
       await Promise.all([
         (iterator as any)[Symbol.asyncDispose](),

packages/shared/src/iterator.ts

@@ -9,86 +9,99 @@ export function isAsyncIteratorObject(maybe: unknown): maybe is AsyncIteratorObj
   return Symbol.asyncIterator in maybe && typeof maybe[Symbol.asyncIterator] === 'function'
 }
 
-export interface CreateAsyncIteratorObjectCleanupFn {
+export interface AsyncIteratorClassNextFn<T, TReturn> {
+  (): Promise<IteratorResult<T, TReturn>>
+}
+
+export interface AsyncIteratorClassCleanupFn {
   (reason: 'return' | 'throw' | 'next' | 'dispose'): Promise<void>
 }
 
-export function createAsyncIteratorObject<T, TReturn, TNext>(
-  next: () => Promise<IteratorResult<T, TReturn>>,
-  cleanup: CreateAsyncIteratorObjectCleanupFn,
-): AsyncIteratorObject<T, TReturn, TNext> & AsyncGenerator<T, TReturn, TNext> {
-  let isExecuteComplete = false
-  let isDone = false
+const fallbackAsyncDisposeSymbol: unique symbol = Symbol.for('asyncDispose')
+const asyncDisposeSymbol: typeof Symbol extends { asyncDispose: infer T } ? T : typeof fallbackAsyncDisposeSymbol = (Symbol as any).asyncDispose ?? fallbackAsyncDisposeSymbol
+
+export class AsyncIteratorClass<T, TReturn = unknown, TNext = unknown> implements AsyncIteratorObject<T, TReturn, TNext>, AsyncGenerator<T, TReturn, TNext> {
+  #isDone = false
+  #isExecuteComplete = false
+  #cleanup: AsyncIteratorClassCleanupFn
+  #next: AsyncIteratorClassNextFn<T, TReturn>
 
-  const iterator = {
-    next: sequential(async () => {
-      if (isDone) {
+  constructor(next: AsyncIteratorClassNextFn<T, TReturn>, cleanup: AsyncIteratorClassCleanupFn) {
+    this.#cleanup = cleanup
+    this.#next = sequential(async () => {
+      if (this.#isDone) {
         return { done: true, value: undefined as any }
       }
 
       try {
         const result = await next()
 
         if (result.done) {
-          isDone = true
+          this.#isDone = true
         }
 
         return result
       }
       catch (err) {
-        isDone = true
+        this.#isDone = true
         throw err
       }
       finally {
-        if (isDone && !isExecuteComplete) {
-          isExecuteComplete = true
-          await cleanup('next')
+        if (this.#isDone && !this.#isExecuteComplete) {
+          this.#isExecuteComplete = true
+          await this.#cleanup('next')
         }
       }
-    }),
-    async return(value: any) {
-      isDone = true
-      if (!isExecuteComplete) {
-        isExecuteComplete = true
-        await cleanup('return')
-      }
+    })
+  }
 
-      return { done: true, value }
-    },
-    async throw(err: any) {
-      isDone = true
-      if (!isExecuteComplete) {
-        isExecuteComplete = true
-        await cleanup('throw')
-      }
+  next(): Promise<IteratorResult<T, TReturn>> {
+    return this.#next()
+  }
 
-      throw err
-    },
-    /**
-     * asyncDispose symbol only available in esnext, we should fallback to Symbol.for('asyncDispose')
-     */
-    async [(Symbol as any).asyncDispose as typeof Symbol extends { asyncDispose: infer T } ? T : any ?? Symbol.for('asyncDispose')]() {
-      isDone = true
-      if (!isExecuteComplete) {
-        isExecuteComplete = true
-        await cleanup('dispose')
-      }
-    },
-    [Symbol.asyncIterator]() {
-      return iterator
-    },
+  async return(value?: any): Promise<IteratorResult<T, TReturn>> {
+    this.#isDone = true
+    if (!this.#isExecuteComplete) {
+      this.#isExecuteComplete = true
+      await this.#cleanup('return')
+    }
+
+    return { done: true, value }
+  }
+
+  async throw(err: any): Promise<IteratorResult<T, TReturn>> {
+    this.#isDone = true
+    if (!this.#isExecuteComplete) {
+      this.#isExecuteComplete = true
+      await this.#cleanup('throw')
+    }
+
+    throw err
   }
 
-  return iterator
+  /**
+   * asyncDispose symbol only available in esnext, we should fallback to Symbol.for('asyncDispose')
+   */
+  async [asyncDisposeSymbol](): Promise<void> {
+    this.#isDone = true
+    if (!this.#isExecuteComplete) {
+      this.#isExecuteComplete = true
+      await this.#cleanup('dispose')
+    }
+  }
+
+  [Symbol.asyncIterator](): this {
+    return this
+  }
 }
 
 export function replicateAsyncIterator<T, TReturn, TNext>(
   source: AsyncIterator<T, TReturn, TNext>,
   count: number,
-): (AsyncIteratorObject<T, TReturn, TNext> & AsyncGenerator<T, TReturn, TNext>)[] {
+): (AsyncIteratorClass<T, TReturn, TNext>)[] {
   const queue = new AsyncIdQueue<IteratorResult<T, TReturn>>()
 
-  const replicated: (AsyncIteratorObject<T, TReturn, TNext> & AsyncGenerator<T, TReturn, TNext>)[] = []
+  const replicated: AsyncIteratorClass<T, TReturn, TNext>[] = []
 
   let error: undefined | { value: unknown }
 
@@ -115,7 +128,7 @@ export function replicateAsyncIterator<T, TReturn, TNext>(
 
   for (let id = 0; id < count; id++) {
     queue.open(id)
-    replicated.push(createAsyncIteratorObject(
+    replicated.push(new AsyncIteratorClass(
       () => {
         start()
 

packages/standard-server-fetch/src/event-iterator.ts

@@ -1,4 +1,4 @@
-import { createAsyncIteratorObject, isTypescriptObject, parseEmptyableJSON, stringifyJSON } from '@orpc/shared'
+import { AsyncIteratorClass, isTypescriptObject, parseEmptyableJSON, stringifyJSON } from '@orpc/shared'
 import {
   encodeEventMessage,
   ErrorEvent,
@@ -9,14 +9,14 @@ import {
 
 export function toEventIterator(
   stream: ReadableStream<Uint8Array> | null,
-): AsyncIteratorObject<unknown | void, unknown | void, void> & AsyncGenerator<unknown | void, unknown | void, void> {
+): AsyncIteratorClass<unknown> {
   const eventStream = stream
     ?.pipeThrough(new TextDecoderStream())
     .pipeThrough(new EventDecoderStream())
 
   const reader = eventStream?.getReader()
 
-  return createAsyncIteratorObject(async () => {
+  return new AsyncIteratorClass(async () => {
     while (true) {
       if (reader === undefined) {
         return { done: true, value: undefined }

packages/standard-server-peer/src/event-iterator.ts

@@ -1,15 +1,15 @@
-import type { CreateAsyncIteratorObjectCleanupFn } from '@orpc/shared'
+import type { AsyncIteratorClassCleanupFn } from '@orpc/shared'
 import type { AsyncIdQueue } from '../../shared/src/queue'
 import type { EventIteratorPayload } from './codec'
-import { createAsyncIteratorObject, isTypescriptObject } from '@orpc/shared'
+import { AsyncIteratorClass, isTypescriptObject } from '@orpc/shared'
 import { ErrorEvent, getEventMeta, withEventMeta } from '@orpc/standard-server'
 
 export function toEventIterator(
   queue: AsyncIdQueue<EventIteratorPayload>,
   id: number,
-  cleanup: CreateAsyncIteratorObjectCleanupFn,
-): AsyncGenerator {
-  return createAsyncIteratorObject(async () => {
+  cleanup: AsyncIteratorClassCleanupFn,
+): AsyncIteratorClass<unknown> {
+  return new AsyncIteratorClass(async () => {
     const item = await queue.pull(id)
 
     switch (item.event) {