feat(client): consumeEventIterator utility

apps/content/docs/client/event-iterator.md

@@ -68,3 +68,35 @@ catch (error) {
 ::: info
 Errors thrown by the server can be instances of `ORPCError`.
 :::
+
+## Using `consumeEventIterator`
+
+oRPC provides a utility function `consumeEventIterator` to consume an event iterator with lifecycle callbacks.
+
+```ts
+import { consumeEventIterator } from '@orpc/client'
+
+const cancel = consumeEventIterator(client.streaming(), {
+  onEvent: (event) => {
+    console.log(event.message)
+  },
+  onError: (error) => {
+    console.error(error)
+  },
+  onSuccess: (value) => {
+    console.log(value)
+  },
+  onFinish: (state) => {
+    console.log(state)
+  },
+})
+
+setTimeout(async () => {
+  // Stop the stream after 1 second
+  await cancel()
+}, 1000)
+```
+
+:::info
+This utility accepts both promises and event iterators. Passing a promise directly lets it infer correct error type.
+:::

packages/client/src/utils.test-d.ts

@@ -1,7 +1,8 @@
+import type { OnFinishState } from '@orpc/shared'
 import type { ORPCError } from './error'
-import type { Client, ClientContext } from './types'
+import type { Client, ClientContext, ClientPromiseResult } from './types'
 import { isDefinedError } from './error'
-import { safe } from './utils'
+import { consumeEventIterator, safe } from './utils'
 
 describe('safe', async () => {
   const client = {} as Client<ClientContext, string, number, Error | ORPCError<'BAD_GATEWAY', { val: string }>>
@@ -65,3 +66,39 @@ describe('safe', async () => {
     expectTypeOf(data).toEqualTypeOf<number | undefined>()
   })
 })
+
+describe('consumeEventIterator', () => {
+  it('can infer types from ClientPromiseResult + AsyncGenerator', () => {
+    void consumeEventIterator({} as ClientPromiseResult<AsyncGenerator<'message-value', 'done-value'>, 'error-value'>, {
+      onEvent: (message) => {
+        expectTypeOf(message).toEqualTypeOf<'message-value'>()
+      },
+      onError: (error) => {
+        expectTypeOf(error).toEqualTypeOf<'error-value'>()
+      },
+      onSuccess: (value) => {
+        expectTypeOf(value).toEqualTypeOf<'done-value' | undefined>()
+      },
+      onFinish: (state) => {
+        expectTypeOf(state).toEqualTypeOf<OnFinishState<'done-value' | undefined, 'error-value'>>()
+      },
+    })
+  })
+
+  it('can infer types from AsyncIterator', () => {
+    void consumeEventIterator({} as AsyncIterator<'message-value', 'done-value'>, {
+      onEvent: (message) => {
+        expectTypeOf(message).toEqualTypeOf<'message-value'>()
+      },
+      onError: (error) => {
+        expectTypeOf(error).toEqualTypeOf<Error>()
+      },
+      onSuccess: (value) => {
+        expectTypeOf(value).toEqualTypeOf<'done-value' | undefined>()
+      },
+      onFinish: (state) => {
+        expectTypeOf(state).toEqualTypeOf<OnFinishState<'done-value' | undefined, Error>>()
+      },
+    })
+  })
+})

packages/client/src/utils.test.ts

@@ -1,5 +1,5 @@
 import { ORPCError } from './error'
-import { resolveFriendlyClientOptions, safe } from './utils'
+import { consumeEventIterator, resolveFriendlyClientOptions, safe } from './utils'
 
 it('safe', async () => {
   const r1 = await safe(Promise.resolve(1))
@@ -27,3 +27,218 @@ it('resolveFriendlyClientOptions', () => {
   expect(resolveFriendlyClientOptions({ context: { a: 1 } })).toEqual({ context: { a: 1 } })
   expect(resolveFriendlyClientOptions({ lastEventId: '123' })).toEqual({ context: {}, lastEventId: '123' })
 })
+
+describe('consumeEventIterator', () => {
+  it('on success', async () => {
+    const iterator = (async function* () {
+      yield 1
+      yield 2
+      return 3
+    }())
+
+    const onEvent = vi.fn()
+    const onError = vi.fn()
+    const onSuccess = vi.fn()
+    const onFinish = vi.fn()
+
+    const unsubscribe = consumeEventIterator(iterator, {
+      onEvent,
+      onError,
+      onSuccess,
+      onFinish,
+    })
+
+    await vi.waitFor(() => {
+      expect(onEvent).toHaveBeenCalledTimes(2)
+      expect(onEvent).toHaveBeenNthCalledWith(1, 1)
+      expect(onEvent).toHaveBeenNthCalledWith(2, 2)
+
+      expect(onSuccess).toHaveBeenCalledTimes(1)
+      expect(onSuccess).toHaveBeenNthCalledWith(1, 3)
+      expect(onFinish).toHaveBeenCalledTimes(1)
+      expect(onFinish).toHaveBeenNthCalledWith(1, [null, 3, true])
+
+      expect(onError).toHaveBeenCalledTimes(0)
+    })
+  })
+
+  it('on error', async () => {
+    const error = new Error('TEST')
+    const iterator = (async function* () {
+      yield 1
+      yield 2
+      throw error
+    }())
+
+    const onEvent = vi.fn()
+    const onError = vi.fn()
+    const onSuccess = vi.fn()
+    const onFinish = vi.fn()
+
+    const unsubscribe = consumeEventIterator(iterator, {
+      onEvent,
+      onError,
+      onSuccess,
+      onFinish,
+    })
+
+    await vi.waitFor(() => {
+      expect(onEvent).toHaveBeenCalledTimes(2)
+      expect(onEvent).toHaveBeenNthCalledWith(1, 1)
+      expect(onEvent).toHaveBeenNthCalledWith(2, 2)
+
+      expect(onError).toHaveBeenCalledTimes(1)
+      expect(onError).toHaveBeenNthCalledWith(1, error)
+
+      expect(onFinish).toHaveBeenCalledTimes(1)
+      expect(onFinish).toHaveBeenNthCalledWith(1, [error, undefined, false])
+
+      expect(onSuccess).toHaveBeenCalledTimes(0)
+    })
+  })
+
+  it('on error without onError and onFinish', async () => {
+    const unhandledRejectionHandler = vi.fn()
+    process.on('unhandledRejection', unhandledRejectionHandler)
+
+    afterEach(() => {
+      process.off('unhandledRejection', unhandledRejectionHandler)
+    })
+
+    const error = new Error('TEST')
+    const iterator = (async function* () {
+      yield 1
+      yield 2
+      throw error
+    }())
+
+    const onEvent = vi.fn()
+
+    const unsubscribe = consumeEventIterator(iterator, {
+      onEvent,
+    })
+
+    await vi.waitFor(() => {
+      expect(onEvent).toHaveBeenCalledTimes(2)
+      expect(onEvent).toHaveBeenNthCalledWith(1, 1)
+      expect(onEvent).toHaveBeenNthCalledWith(2, 2)
+    })
+
+    expect(unhandledRejectionHandler).toHaveBeenCalledTimes(1)
+    expect(unhandledRejectionHandler).toHaveBeenNthCalledWith(1, error, expect.anything())
+  })
+
+  it('unsubscribe', async () => {
+    let cleanup = false
+    const iterator = (async function* () {
+      try {
+        await new Promise(resolve => setTimeout(resolve, 25))
+        yield 1
+        yield 2
+        return 3
+      }
+      finally {
+        cleanup = true
+      }
+    }())
+
+    const onEvent = vi.fn()
+    const onSuccess = vi.fn()
+    const onFinish = vi.fn()
+    const onError = vi.fn()
+
+    const unsubscribe = consumeEventIterator(iterator, {
+      onEvent,
+      onError,
+      onSuccess,
+      onFinish,
+    })
+
+    await new Promise(resolve => setTimeout(resolve, 1))
+    await unsubscribe()
+    expect(cleanup).toBe(true)
+    // side-effect of async generator - waiting for .next resolve before .return effect
+    expect(onEvent).toHaveBeenCalledTimes(1)
+    expect(onEvent).toHaveBeenNthCalledWith(1, 1)
+
+    expect(onError).toHaveBeenCalledTimes(0)
+    expect(onSuccess).toHaveBeenCalledTimes(1)
+    expect(onFinish).toHaveBeenCalledTimes(1)
+    // undefined can be passed on success because iterator can be canceled
+    expect(onSuccess).toHaveBeenNthCalledWith(1, undefined)
+    expect(onFinish).toHaveBeenNthCalledWith(1, [null, undefined, true])
+  })
+
+  it('error on unsubscribe', async () => {
+    const error = new Error('TEST')
+    let cleanup = false
+    const iterator = (async function* () {
+      try {
+        await new Promise(resolve => setTimeout(resolve, 25))
+        yield 1
+        yield 2
+        return 3
+      }
+      finally {
+        cleanup = true
+        // eslint-disable-next-line no-unsafe-finally
+        throw error
+      }
+    }())
+
+    const onEvent = vi.fn()
+    const onError = vi.fn()
+    const onSuccess = vi.fn()
+    const onFinish = vi.fn()
+
+    const unsubscribe = consumeEventIterator(iterator, {
+      onEvent,
+      onError,
+      onSuccess,
+      onFinish,
+    })
+
+    await new Promise(resolve => setTimeout(resolve, 1))
+    await expect(unsubscribe()).rejects.toThrow(error)
+    expect(cleanup).toBe(true)
+    // side-effect of async generator - waiting for .next resolve before .return effect
+    expect(onEvent).toHaveBeenCalledTimes(1)
+    expect(onEvent).toHaveBeenNthCalledWith(1, 1)
+
+    expect(onError).toHaveBeenCalledTimes(0)
+    expect(onSuccess).toHaveBeenCalledTimes(1)
+    expect(onFinish).toHaveBeenCalledTimes(1)
+    // undefined can be passed on success because iterator can be canceled
+    expect(onSuccess).toHaveBeenNthCalledWith(1, undefined)
+    expect(onFinish).toHaveBeenNthCalledWith(1, [null, undefined, true])
+  })
+
+  it('on iterator promise rejection', async () => {
+    const error = new Error('TEST')
+    const iterator = Promise.reject(error)
+
+    const onEvent = vi.fn()
+    const onError = vi.fn()
+    const onSuccess = vi.fn()
+    const onFinish = vi.fn()
+
+    void consumeEventIterator(iterator, {
+      onEvent,
+      onError,
+      onSuccess,
+      onFinish,
+    })
+
+    await vi.waitFor(() => {
+      expect(onEvent).toHaveBeenCalledTimes(0)
+
+      expect(onError).toHaveBeenCalledTimes(1)
+      expect(onError).toHaveBeenNthCalledWith(1, error)
+
+      expect(onFinish).toHaveBeenCalledTimes(1)
+      expect(onFinish).toHaveBeenNthCalledWith(1, [error, undefined, false])
+
+      expect(onSuccess).toHaveBeenCalledTimes(0)
+    })
+  })
+})

packages/client/src/utils.ts

@@ -1,4 +1,4 @@
-import type { ThrowableError } from '@orpc/shared'
+import type { OnFinishState, ThrowableError } from '@orpc/shared'
 import type { ORPCError } from './error'
 import type { ClientContext, ClientOptions, ClientPromiseResult, FriendlyClientOptions } from './types'
 import { isDefinedError } from './error'
@@ -48,3 +48,79 @@ export function resolveFriendlyClientOptions<T extends ClientContext>(options: F
     context: options.context ?? {} as T, // Context only optional if all fields are optional
   }
 }
+
+export interface ConsumeEventIteratorOptions<T, TReturn, TError> {
+  /**
+   * Called on each event
+   */
+  onEvent: (event: T) => void
+  /**
+   * Called once error happens
+   */
+  onError?: (error: TError) => void
+  /**
+   * Called once event iterator is done
+   *
+   * @info If iterator is canceled, `undefined` can be passed on success
+   */
+  onSuccess?: (value: TReturn | undefined) => void
+  /**
+   * Called once after onError or onSuccess
+   *
+   * @info If iterator is canceled, `undefined` can be passed on success
+   */
+  onFinish?: (state: OnFinishState<TReturn | undefined, TError>) => void
+}
+
+/**
+ * Consumes an event iterator with lifecycle callbacks
+ *
+ * @warning If no `onError` or `onFinish` is provided, unhandled rejections will be thrown
+ * @return unsubscribe callback
+ */
+export function consumeEventIterator<T, TReturn, TError = ThrowableError>(
+  iterator: AsyncIterator<T, TReturn> | ClientPromiseResult<AsyncIterator<T, TReturn>, TError>,
+  options: ConsumeEventIteratorOptions<T, TReturn, TError>,
+): () => Promise<void> {
+  void (async () => {
+    let onFinishState: OnFinishState<TReturn | undefined, TError>
+
+    try {
+      const resolvedIterator = await iterator
+
+      while (true) {
+        const { done, value } = await resolvedIterator.next()
+
+        if (done) {
+          // if iterator is canceled, value can be undefined
+          const realValue = value as typeof value | undefined
+          onFinishState = [null, realValue, true]
+          options.onSuccess?.(realValue)
+          break
+        }
+
+        options.onEvent(value)
+      }
+    }
+    catch (error) {
+      onFinishState = [error as TError, undefined, false]
+
+      /**
+       * If no `onError` or `onFinish` is provided, unhandled rejections will be thrown
+       * This is best practice for error handling - error should always be handled
+       */
+      if (!options.onError && !options.onFinish) {
+        throw error
+      }
+
+      options.onError?.(error as TError)
+    }
+    finally {
+      options.onFinish?.(onFinishState!)
+    }
+  })()
+
+  return async () => {
+    await (await iterator)?.return?.()
+  }
+}