Skip to content

Commit dda04c5

Browse files
dinwwwhdinwwwh
authored
feat(trpc): support event iterator for toORPCRouter (#716)
<!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Added support for async iterable outputs in procedures, enabling streaming events with event metadata. * Subscription procedures can now accept and utilize a `lastEventId` parameter for event streaming. * **Tests** * Introduced new tests to verify event iterator behavior, including event metadata and cleanup logic in subscriptions. * Enhanced test data to yield tracked events with identifiers and metadata. * **Chores** * Updated dependencies and TypeScript project references to include the client package. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
1 parent 083a799 commit dda04c5

7 files changed

Lines changed: 147 additions & 15 deletions

File tree

‎packages/trpc/package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
"@trpc/server": ">=11.4.2"
3939
},
4040
"dependencies": {
41+
"@orpc/client": "workspace:*",
4142
"@orpc/server": "workspace:*",
4243
"@orpc/shared": "workspace:*"
4344
},

‎packages/trpc/src/to-orpc-router.test-d.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import type { ContractRouter, InferRouterInitialContext, Procedure, Router, Schema } from '@orpc/server'
2+
import type { AsyncIteratorClass } from '@orpc/shared'
23
import type { inferRouterContext } from '@trpc/server'
3-
import type { inferRouterMeta } from '@trpc/server/unstable-core-do-not-import'
4+
import type { inferRouterMeta, TrackedData } from '@trpc/server/unstable-core-do-not-import'
45
import type { TRPCContext, TRPCMeta, trpcRouter } from '../tests/shared'
56
import type { experimental_ToORPCRouterResult as ToORPCRouterResult } from './to-orpc-router'
67

@@ -24,7 +25,7 @@ it('ToORPCRouterResult', () => {
2425
>()
2526

2627
expectTypeOf(orpcRouter.subscribe).toEqualTypeOf<
27-
Procedure<TRPCContext, object, Schema<{ u: string }, unknown>, Schema<unknown, AsyncIterable<string, void, any>>, object, TRPCMeta>
28+
Procedure<TRPCContext, object, Schema<{ u: string }, unknown>, Schema<unknown, AsyncIteratorClass<'pong' | TrackedData<{ order: number }>, void, any>>, object, TRPCMeta>
2829
>()
2930

3031
expectTypeOf(orpcRouter.nested).toEqualTypeOf<
@@ -35,7 +36,7 @@ it('ToORPCRouterResult', () => {
3536

3637
expectTypeOf(orpcRouter.lazy).toEqualTypeOf<
3738
{
38-
subscribe: Procedure<TRPCContext, object, Schema<void, unknown>, Schema<unknown, AsyncIterable<string, void, any>>, object, TRPCMeta>
39+
subscribe: Procedure<TRPCContext, object, Schema<void, unknown>, Schema<unknown, AsyncIteratorClass<string, void, any>>, object, TRPCMeta>
3940
lazy: {
4041
throw: Procedure<TRPCContext, object, Schema<{ input: number }, unknown>, Schema<unknown, { output: string }>, object, TRPCMeta>
4142
}

‎packages/trpc/src/to-orpc-router.test.ts‎

Lines changed: 93 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,15 @@
1-
import { call, createRouterClient, isProcedure, ORPCError, unlazy } from '@orpc/server'
1+
import { call, createRouterClient, getEventMeta, isProcedure, ORPCError, unlazy } from '@orpc/server'
22
import { isAsyncIteratorObject } from '@orpc/shared'
3+
import { tracked } from '@trpc/server'
4+
import { z } from 'zod'
35
import { inputSchema, outputSchema } from '../../contract/tests/shared'
4-
import { trpcRouter } from '../tests/shared'
6+
import { t, trpcRouter } from '../tests/shared'
57
import { experimental_toORPCRouter as toORPCRouter } from './to-orpc-router'
68

9+
beforeEach(() => {
10+
vi.clearAllMocks()
11+
})
12+
713
describe('toORPCRouter', async () => {
814
const orpcRouter = toORPCRouter(trpcRouter)
915

@@ -73,4 +79,89 @@ describe('toORPCRouter', async () => {
7379
})
7480
})
7581
})
82+
83+
describe('event iterators', () => {
84+
it('subscribe & tracked', async () => {
85+
const output = await call(orpcRouter.subscribe, { u: '2' }, { lastEventId: 'id-1', context: { a: 'test' } }) as any
86+
expect(output).toSatisfy(isAsyncIteratorObject)
87+
await expect(output.next()).resolves.toEqual({ done: false, value: 'pong' })
88+
await expect(output.next()).resolves.toSatisfy((result) => {
89+
expect(result.done).toEqual(false)
90+
expect(result.value).toEqual({ id: 'id-1', data: { order: 1 } })
91+
expect(getEventMeta(result.value)).toEqual({ id: 'id-1' })
92+
93+
return true
94+
})
95+
await expect(output.next()).resolves.toSatisfy((result) => {
96+
expect(result.done).toEqual(false)
97+
expect(result.value).toEqual({ id: 'id-2', data: { order: 2 } })
98+
expect(getEventMeta(result.value)).toEqual({ id: 'id-2' })
99+
100+
return true
101+
})
102+
await expect(output.next()).resolves.toEqual({ done: true, value: undefined })
103+
})
104+
105+
it('lastEventId', async () => {
106+
const trackedSubscription = vi.fn(async function* () {
107+
yield { order: 1 }
108+
yield tracked('id-2', { order: 2 })
109+
})
110+
111+
const trpcRouter = t.router({
112+
tracked: t.procedure
113+
.input(z.any())
114+
.subscription(trackedSubscription),
115+
})
116+
117+
const orpcRouter = toORPCRouter(trpcRouter)
118+
119+
await call(orpcRouter.tracked, { u: 'u' }, { lastEventId: 'id-1', context: { a: 'test' } })
120+
expect(trackedSubscription).toHaveBeenNthCalledWith(1, expect.objectContaining({
121+
input: { u: 'u', lastEventId: 'id-1' },
122+
}))
123+
124+
await call(orpcRouter.tracked, undefined, { lastEventId: 'id-2', context: { a: 'test' } })
125+
expect(trackedSubscription).toHaveBeenNthCalledWith(2, expect.objectContaining({
126+
input: { lastEventId: 'id-2' },
127+
}))
128+
129+
await call(orpcRouter.tracked, 1234, { lastEventId: 'id-3', context: { a: 'test' } })
130+
expect(trackedSubscription).toHaveBeenNthCalledWith(3, expect.objectContaining({
131+
input: 1234,
132+
}))
133+
})
134+
135+
it('works with AsyncIterable & cleanup', async () => {
136+
let cleanupCalled = false
137+
138+
const trackedSubscription = vi.fn(async () => {
139+
return {
140+
async* [Symbol.asyncIterator]() {
141+
try {
142+
yield { order: 1 }
143+
yield tracked('id-2', { order: 2 })
144+
}
145+
finally {
146+
cleanupCalled = true
147+
}
148+
},
149+
}
150+
})
151+
152+
const trpcRouter = t.router({
153+
tracked: t.procedure
154+
.input(z.any())
155+
.subscription(trackedSubscription),
156+
})
157+
158+
const orpcRouter = toORPCRouter(trpcRouter)
159+
160+
const output = await call(orpcRouter.tracked, { u: 'u' }, { lastEventId: 'id-1', context: { a: 'test' } })
161+
162+
await expect(output.next()).resolves.toEqual({ done: false, value: { order: 1 } })
163+
await expect(output.return?.()).resolves.toEqual({ done: true, value: undefined })
164+
expect(cleanupCalled).toBe(true)
165+
})
166+
})
76167
})

‎packages/trpc/src/to-orpc-router.ts‎

Lines changed: 42 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,21 @@
1+
import type { AsyncIteratorClass } from '@orpc/shared'
12
import type { AnyProcedure, AnyRouter, inferRouterContext } from '@trpc/server'
2-
import type { inferRouterMeta, Parser } from '@trpc/server/unstable-core-do-not-import'
3+
import type { inferRouterMeta, Parser, TrackedData } from '@trpc/server/unstable-core-do-not-import'
4+
import { mapEventIterator } from '@orpc/client'
35
import * as ORPC from '@orpc/server'
4-
import { isTypescriptObject } from '@orpc/shared'
5-
import { TRPCError } from '@trpc/server'
6-
import { getHTTPStatusCodeFromError } from '@trpc/server/unstable-core-do-not-import'
6+
import { isObject, isTypescriptObject } from '@orpc/shared'
7+
import { isTrackedEnvelope, TRPCError } from '@trpc/server'
8+
import { getHTTPStatusCodeFromError, isAsyncIterable } from '@trpc/server/unstable-core-do-not-import'
79

810
export interface experimental_ORPCMeta extends ORPC.Route {
911

1012
}
1113

14+
export type experimental_ToORPCOutput<T>
15+
= T extends AsyncIterable<infer TData, infer TReturn, infer TNext>
16+
? AsyncIteratorClass<TData, TReturn, TNext>
17+
: T
18+
1219
export type experimental_ToORPCRouterResult<TContext extends ORPC.Context, TMeta extends ORPC.Meta, TRecord extends Record<string, any>>
1320
= {
1421
[K in keyof TRecord]:
@@ -17,7 +24,7 @@ export type experimental_ToORPCRouterResult<TContext extends ORPC.Context, TMeta
1724
TContext,
1825
object,
1926
ORPC.Schema<TRecord[K]['_def']['$types']['input'], unknown>,
20-
ORPC.Schema<unknown, TRecord[K]['_def']['$types']['output']>,
27+
ORPC.Schema<unknown, experimental_ToORPCOutput<TRecord[K]['_def']['$types']['output']>>,
2128
object,
2229
TMeta
2330
>
@@ -88,16 +95,42 @@ function toORPCProcedure(procedure: AnyProcedure) {
8895
middlewares: [],
8996
inputSchema: toDisabledStandardSchema(procedure._def.inputs.at(-1)),
9097
outputSchema: toDisabledStandardSchema((procedure as any)._def.output),
91-
handler: async ({ context, signal, path, input }) => {
98+
handler: async ({ context, signal, path, input, lastEventId }) => {
9299
try {
93-
return await procedure({
100+
const trpcInput = lastEventId !== undefined && (input === undefined || isObject(input))
101+
? { ...input, lastEventId }
102+
: input
103+
104+
const output = await procedure({
94105
ctx: context,
95106
signal,
96107
path: path.join('.'),
97108
type: procedure._def.type,
98-
input,
99-
getRawInput: () => input,
109+
input: trpcInput,
110+
getRawInput: () => trpcInput,
100111
})
112+
113+
if (isAsyncIterable(output)) {
114+
return mapEventIterator(output[Symbol.asyncIterator](), {
115+
error: async error => error,
116+
value: (value) => {
117+
if (isTrackedEnvelope(value)) {
118+
const [id, data] = value
119+
120+
return ORPC.withEventMeta({
121+
id,
122+
data,
123+
} satisfies TrackedData<unknown>, {
124+
id,
125+
})
126+
}
127+
128+
return value
129+
},
130+
})
131+
}
132+
133+
return output
101134
}
102135
catch (cause) {
103136
if (cause instanceof TRPCError) {

‎packages/trpc/tests/shared.ts‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import type { experimental_ORPCMeta as ORPCMeta } from '../src/to-orpc-router'
2-
import { initTRPC, lazy, TRPCError } from '@trpc/server'
2+
import { initTRPC, lazy, tracked, TRPCError } from '@trpc/server'
33
import { z } from 'zod/v4'
44
import { inputSchema, outputSchema } from '../../contract/tests/shared'
55

@@ -34,6 +34,8 @@ export const trpcRouter = t.router({
3434
.input(z.object({ u: z.string() }))
3535
.subscription(async function* () {
3636
yield 'pong'
37+
yield tracked('id-1', { order: 1 })
38+
yield tracked('id-2', { order: 2 })
3739
}),
3840

3941
nested: {

‎packages/trpc/tsconfig.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
"extends": "../../tsconfig.lib.json",
33
"references": [
44
{ "path": "../server" },
5+
{ "path": "../client" },
56
{ "path": "../shared" }
67
],
78
"include": ["src"],

‎pnpm-lock.yaml‎

Lines changed: 3 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)