feat(client, server): Event-Source Ping Interval (#183)

* feat: support encode event comment

* feat: event-source ping internal

* sync

* clear ping internal on error

* docs

* update tests

* improve naming

* improve

Author: dinwwwh

apps/content/docs/client/rpc-link.md

@@ -89,7 +89,7 @@ const link = new RPCLink({
 
 ## Event Source Configuration
 
-Customize the retry logic for event sources using these options:
+Customize the retry logic for event sources (the mechanism behind [Event Iterator](/docs/event-iterator)) using these options:
 
 - **eventSourceMaxNumberOfRetries:** Maximum retry attempts.
 - **eventSourceRetryDelay:** Delay between retries.
@@ -115,3 +115,23 @@ const link = new RPCLink<ClientContext>({
 :::tip
 You should disable event source retries when streaming results from a chatbot AI.
 :::
+
+## Event-Source Ping Interval
+
+To keep EventSource connections alive (the mechanism behind [Event Iterator](/docs/event-iterator)), `RPCLink` periodically sends a ping comment to the server. You can configure this behavior using the following options:
+
+- `eventSourcePingEnabled` (default: `true`) – Enables or disables pings.
+- `eventSourcePingInterval` (default: `5000`) – Time between pings (in milliseconds).
+- `eventSourcePingContent` (default: `''`) – Custom content for ping messages.
+
+```ts
+const link = new RPCLink({
+  eventSourcePingEnabled: true,
+  eventSourcePingInterval: 5000, // 5 seconds
+  eventSourcePingContent: '',
+})
+```
+
+:::warning
+These options for sending [Event Iterator](/docs/event-iterator) from client to the server, not from the server to client as used in [RPCHandler](/docs/rpc-handler#event-source-ping-interval) or [OpenAPIHandler](/docs/openapi/openapi-handler#event-source-ping-interval).
+:::

apps/content/docs/openapi/openapi-handler.md

@@ -86,3 +86,19 @@ export default async function fetch(request: Request) {
   return new Response('Not Found', { status: 404 })
 }
 ```
+
+## Event-Source Ping Interval
+
+To keep EventSource connections alive (the mechanism behind [Event Iterator](/docs/event-iterator)), `OpenAPIHandler` periodically sends a ping comment to the client. You can configure this behavior using the following options:
+
+- `eventSourcePingEnabled` (default: `true`) – Enables or disables pings.
+- `eventSourcePingInterval` (default: `5000`) – Time between pings (in milliseconds).
+- `eventSourcePingContent` (default: `''`) – Custom content for ping messages.
+
+```ts
+const result = await handler.handle(request, {
+  eventSourcePingEnabled: true,
+  eventSourcePingInterval: 5000, // 5 seconds
+  eventSourcePingContent: '',
+})
+```

apps/content/docs/rpc-handler.md

@@ -65,3 +65,19 @@ export default async function fetch(request: Request) {
   return new Response('Not Found', { status: 404 })
 }
 ```
+
+## Event-Source Ping Interval
+
+To keep EventSource connections alive (the mechanism behind [Event Iterator](/docs/event-iterator)), `RPCHandler` periodically sends a ping comment to the client. You can configure this behavior using the following options:
+
+- `eventSourcePingEnabled` (default: `true`) – Enables or disables pings.
+- `eventSourcePingInterval` (default: `5000`) – Time between pings (in milliseconds).
+- `eventSourcePingContent` (default: `''`) – Custom content for ping messages.
+
+```ts
+const result = await handler.handle(request, {
+  eventSourcePingEnabled: true,
+  eventSourcePingInterval: 5000, // 5 seconds
+  eventSourcePingContent: '',
+})
+```

packages/client/src/adapters/fetch/rpc-link.ts

@@ -1,5 +1,6 @@
 import type { Value } from '@orpc/shared'
 import type { StandardBody } from '@orpc/standard-server'
+import type { ToFetchBodyOptions } from '@orpc/standard-server-fetch'
 import type { ClientContext, ClientLink, ClientOptionsOut } from '../../types'
 import type { FetchWithContext } from './types'
 import { isAsyncIteratorObject, stringifyJSON, trim, value } from '@orpc/shared'
@@ -12,7 +13,7 @@ type HTTPMethod = 'GET' | 'POST' | 'PUT' | 'DELETE' | 'PATCH'
 
 export class InvalidEventSourceRetryResponse extends Error { }
 
-export interface RPCLinkOptions<TClientContext extends ClientContext> {
+export interface RPCLinkOptions<TClientContext extends ClientContext> extends ToFetchBodyOptions {
   /**
    * Base url for all requests.
    */
@@ -105,6 +106,7 @@ export class RPCLink<TClientContext extends ClientContext> implements ClientLink
   private readonly eventSourceMaxNumberOfRetries: Exclude<RPCLinkOptions<TClientContext>['eventSourceMaxNumberOfRetries'], undefined>
   private readonly eventSourceRetryDelay: Exclude<RPCLinkOptions<TClientContext>['eventSourceRetryDelay'], undefined>
   private readonly eventSourceRetry: Exclude<RPCLinkOptions<TClientContext>['eventSourceRetry'], undefined>
+  private readonly toFetchBodyOptions: ToFetchBodyOptions
 
   constructor(options: RPCLinkOptions<TClientContext>) {
     this.fetch = options.fetch ?? globalThis.fetch.bind(globalThis)
@@ -120,6 +122,8 @@ export class RPCLink<TClientContext extends ClientContext> implements ClientLink
 
     this.eventSourceRetryDelay = options.eventSourceRetryDelay
       ?? (({ retryTimes, lastRetry }) => lastRetry ?? (1000 * 2 ** retryTimes))
+
+    this.toFetchBodyOptions = options
   }
 
   async call(path: readonly string[], input: unknown, options: ClientOptionsOut<TClientContext>): Promise<unknown> {
@@ -160,7 +164,7 @@ export class RPCLink<TClientContext extends ClientContext> implements ClientLink
   ): Promise<unknown> {
     const encoded = await this.encodeRequest(path, input, options)
 
-    const fetchBody = toFetchBody(encoded.body, encoded.headers)
+    const fetchBody = toFetchBody(encoded.body, encoded.headers, this.toFetchBodyOptions)
 
     if (options.lastEventId !== undefined) {
       encoded.headers.set('last-event-id', options.lastEventId)

packages/openapi/src/adapters/fetch/openapi-handler.test.ts

@@ -44,7 +44,8 @@ describe('openAPIHandler', () => {
       },
     })
 
-    const result = await handler.handle(request, { prefix: '/api/v1', context: { db: 'postgres' } })
+    const options = { prefix: '/api/v1', context: { db: 'postgres' } } as const
+    const result = await handler.handle(request, options)
 
     expect(result).toEqual({
       matched: true,
@@ -54,7 +55,7 @@ describe('openAPIHandler', () => {
     expect(handle).toHaveBeenCalledOnce()
     expect(handle).toHaveBeenCalledWith(
       vi.mocked(toStandardRequest).mock.results[0]!.value,
-      { prefix: '/api/v1', context: { db: 'postgres' } },
+      options,
     )
 
     expect(vi.mocked(toStandardRequest)).toHaveBeenCalledOnce()
@@ -65,7 +66,7 @@ describe('openAPIHandler', () => {
       status: 200,
       headers: {},
       body: '__body__',
-    })
+    }, options)
   })
 
   it('on mismatch', async () => {

packages/openapi/src/adapters/fetch/openapi-handler.ts

@@ -2,6 +2,7 @@ import type { Context, Router } from '@orpc/server'
 import type { FetchHandler, FetchHandleResult } from '@orpc/server/fetch'
 import type { StandardHandleOptions } from '@orpc/server/standard'
 import type { MaybeOptionalOptions } from '@orpc/shared'
+import type { ToFetchResponseOptions } from '@orpc/standard-server-fetch'
 import type { OpenAPIHandlerOptions } from '../standard'
 import { StandardHandler } from '@orpc/server/standard'
 import { toFetchResponse, toStandardRequest } from '@orpc/standard-server-fetch'
@@ -17,18 +18,21 @@ export class OpenAPIHandler<T extends Context> implements FetchHandler<T> {
     this.standardHandler = new StandardHandler(router, matcher, codec, options)
   }
 
-  async handle(request: Request, ...rest: MaybeOptionalOptions<StandardHandleOptions<T>>): Promise<FetchHandleResult> {
+  async handle(
+    request: Request,
+    ...[options]: MaybeOptionalOptions<StandardHandleOptions<T> & ToFetchResponseOptions>
+  ): Promise<FetchHandleResult> {
     const standardRequest = toStandardRequest(request)
 
-    const result = await this.standardHandler.handle(standardRequest, ...rest)
+    const result = await this.standardHandler.handle(standardRequest, options as any)
 
     if (!result.matched) {
       return result
     }
 
     return {
       matched: true,
-      response: toFetchResponse(result.response),
+      response: toFetchResponse(result.response, options),
     }
   }
 }

packages/openapi/src/adapters/node/openapi-handler.test.ts

@@ -61,7 +61,8 @@ describe('openapiHandler', async () => {
       },
     })
 
-    const result = await handler.handle(req, res, { prefix: '/api/v1', context: { db: 'postgres' } })
+    const options = { prefix: '/api/v1', context: { db: 'postgres' } } as const
+    const result = await handler.handle(req, res, options)
 
     expect(result).toEqual({
       matched: true,
@@ -70,7 +71,7 @@ describe('openapiHandler', async () => {
     expect(handle).toHaveBeenCalledOnce()
     expect(handle).toHaveBeenCalledWith(
       standardRequest,
-      { prefix: '/api/v1', context: { db: 'postgres' } },
+      options,
     )
 
     expect(toStandardRequest).toHaveBeenCalledOnce()
@@ -81,7 +82,7 @@ describe('openapiHandler', async () => {
       status: 200,
       headers: {},
       body: '__body__',
-    })
+    }, options)
   })
 
   it('on mismatch', async () => {

packages/openapi/src/adapters/node/openapi-handler.ts

@@ -2,6 +2,7 @@ import type { Context, Router } from '@orpc/server'
 import type { NodeHttpHandler, NodeHttpHandleResult, NodeHttpRequest, NodeHttpResponse } from '@orpc/server/node'
 import type { StandardHandleOptions } from '@orpc/server/standard'
 import type { MaybeOptionalOptions } from '@orpc/shared'
+import type { SendStandardResponseOptions } from '@orpc/standard-server-node'
 import type { OpenAPIHandlerOptions } from '../standard'
 import { StandardHandler } from '@orpc/server/standard'
 import { sendStandardResponse, toStandardRequest } from '@orpc/standard-server-node'
@@ -20,17 +21,17 @@ export class OpenAPIHandler<T extends Context> implements NodeHttpHandler<T> {
   async handle(
     req: NodeHttpRequest,
     res: NodeHttpResponse,
-    ...rest: MaybeOptionalOptions<StandardHandleOptions<T>>
+    ...[options]: MaybeOptionalOptions<StandardHandleOptions<T> & SendStandardResponseOptions>
   ): Promise<NodeHttpHandleResult> {
     const standardRequest = toStandardRequest(req, res)
 
-    const result = await this.standardHandler.handle(standardRequest, ...rest)
+    const result = await this.standardHandler.handle(standardRequest, options as any)
 
     if (!result.matched) {
       return { matched: false }
     }
 
-    await sendStandardResponse(res, result.response)
+    await sendStandardResponse(res, result.response, options)
 
     return { matched: true }
   }

packages/server/src/adapters/fetch/rpc-handler.test.ts

@@ -39,7 +39,8 @@ describe('rpcHandler', () => {
       body: '__body__',
     } })
 
-    const result = await handler.handle(request, { prefix: '/api/v1', context: { db: 'postgres' } })
+    const options = { prefix: '/api/v1', context: { db: 'postgres' } } as const
+    const result = await handler.handle(request, options)
 
     expect(result).toEqual({
       matched: true,
@@ -49,7 +50,7 @@ describe('rpcHandler', () => {
     expect(handle).toHaveBeenCalledOnce()
     expect(handle).toHaveBeenCalledWith(
       vi.mocked(toStandardRequest).mock.results[0]!.value,
-      { prefix: '/api/v1', context: { db: 'postgres' } },
+      options,
     )
 
     expect(vi.mocked(toStandardRequest)).toHaveBeenCalledOnce()
@@ -60,7 +61,7 @@ describe('rpcHandler', () => {
       status: 200,
       headers: {},
       body: '__body__',
-    })
+    }, options)
   })
 
   it('on mismatch', async () => {

packages/server/src/adapters/fetch/rpc-handler.ts

@@ -1,4 +1,5 @@
 import type { MaybeOptionalOptions } from '@orpc/shared'
+import type { ToFetchResponseOptions } from '@orpc/standard-server-fetch'
 import type { Context } from '../../context'
 import type { Router } from '../../router'
 import type { RPCHandlerOptions, StandardHandleOptions } from '../standard'
@@ -15,18 +16,21 @@ export class RPCHandler<T extends Context> implements FetchHandler<T> {
     this.standardHandler = new StandardHandler(router, matcher, codec, options)
   }
 
-  async handle(request: Request, ...rest: MaybeOptionalOptions<StandardHandleOptions<T>>): Promise<FetchHandleResult> {
+  async handle(
+    request: Request,
+    ...[options]: MaybeOptionalOptions<StandardHandleOptions<T> & ToFetchResponseOptions>
+  ): Promise<FetchHandleResult> {
     const standardRequest = toStandardRequest(request)
 
-    const result = await this.standardHandler.handle(standardRequest, ...rest)
+    const result = await this.standardHandler.handle(standardRequest, options as any)
 
     if (!result.matched) {
       return result
     }
 
     return {
       matched: true,
-      response: toFetchResponse(result.response),
+      response: toFetchResponse(result.response, options),
     }
   }
 }