diff --git a/apps/content/docs/client/event-iterator.md b/apps/content/docs/client/event-iterator.md index d6f89059e..cce766c73 100644 --- a/apps/content/docs/client/event-iterator.md +++ b/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. +::: diff --git a/packages/client/src/utils.test-d.ts b/packages/client/src/utils.test-d.ts index 5a5472fa6..dba101670 100644 --- a/packages/client/src/utils.test-d.ts +++ b/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> @@ -65,3 +66,39 @@ describe('safe', async () => { expectTypeOf(data).toEqualTypeOf() }) }) + +describe('consumeEventIterator', () => { + it('can infer types from ClientPromiseResult + AsyncGenerator', () => { + void consumeEventIterator({} as ClientPromiseResult, '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>() + }, + }) + }) + + 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() + }, + onSuccess: (value) => { + expectTypeOf(value).toEqualTypeOf<'done-value' | undefined>() + }, + onFinish: (state) => { + expectTypeOf(state).toEqualTypeOf>() + }, + }) + }) +}) diff --git a/packages/client/src/utils.test.ts b/packages/client/src/utils.test.ts index 31c78fdf9..4e1d5bad2 100644 --- a/packages/client/src/utils.test.ts +++ b/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) + }) + }) +}) diff --git a/packages/client/src/utils.ts b/packages/client/src/utils.ts index 0752614a0..822cb4360 100644 --- a/packages/client/src/utils.ts +++ b/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(options: F context: options.context ?? {} as T, // Context only optional if all fields are optional } } + +export interface ConsumeEventIteratorOptions { + /** + * 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) => 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( + iterator: AsyncIterator | ClientPromiseResult, TError>, + options: ConsumeEventIteratorOptions, +): () => Promise { + void (async () => { + let onFinishState: OnFinishState + + 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?.() + } +}