Skip to content

Commit be82b92

Browse files
dinwwwhdinwwwh
authored
refactor(publisher,bun,cloudflare)!: rename replay option to resume (#1711)
Renames the event `replay` feature to `resume` across all publisher adapters. The feature lets subscribers continue from `lastEventId` after a disconnect — `resume` matches that purpose (and the v1 naming), while `replay` suggested full event-history replay. ```ts const publisher = new MemoryPublisher<Events>({ resume: { // [!code ++] replay: { // [!code --] enabled: true, seconds: 300, }, }) ``` - `replay` option → `resume` in `MemoryPublisher`, `RedisPublisher`, `UpstashPublisher`, `BunRedisPublisher`, and `DurablePublisherObject` - `DurablePublisherObjectReplayOptions` type → `DurablePublisherObjectResumeOptions` - Docs, playgrounds, and internal names updated to match
1 parent 428ad93 commit be82b92

21 files changed

Lines changed: 278 additions & 278 deletions

‎apps/content/docs/helpers/publisher.md‎

Lines changed: 23 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# Publisher Helpers
22

3-
Publisher helpers provide a unified way to publish and subscribe to events across different storage backends in oRPC applications. They support both static and dynamic event names, along with optional replay of missed events for subscribers.
3+
Publisher helpers provide a unified way to publish and subscribe to events across different storage backends in oRPC applications. They support both static and dynamic event names, along with optional resume support so subscribers can catch up on missed events.
44

55
## Installation
66

@@ -70,7 +70,7 @@ const publisher = new MemoryPublisher<Record<string, { message: string }>>()
7070

7171
## Adapters
7272

73-
| Name | Replay Support | Adapter for |
73+
| Name | Resume Support | Adapter for |
7474
| ------------------- | -------------- | -------------------------------------------------------------------------------- |
7575
| `MemoryPublisher` | ✅ | In-memory storage |
7676
| `RedisPublisher` | ✅ | [Redis](https://github.com/redis/redis) |
@@ -84,9 +84,9 @@ const publisher = new MemoryPublisher<Record<string, { message: string }>>()
8484
import { MemoryPublisher } from '@orpc/publisher/memory'
8585

8686
const publisher = new MemoryPublisher<Events>({
87-
replay: {
87+
resume: {
8888
/**
89-
* Whether event replay support is enabled.
89+
* Whether event resume support is enabled.
9090
*
9191
* When enabled, published events are temporarily stored so new
9292
* subscribers can resume from a previous position using `lastEventId`.
@@ -96,7 +96,7 @@ const publisher = new MemoryPublisher<Events>({
9696
enabled: false,
9797

9898
/**
99-
* How long (in seconds) to retain events for replay.
99+
* How long (in seconds) to retain events for resume.
100100
*
101101
* Expired events are cleaned up lazily for performance reasons, so
102102
* some events may remain available slightly longer than this period.
@@ -142,9 +142,9 @@ const publisher = new RedisPublisher<Events>(client, {
142142
*/
143143
serializer: undefined,
144144

145-
replay: {
145+
resume: {
146146
/**
147-
* Whether event replay support is enabled.
147+
* Whether event resume support is enabled.
148148
*
149149
* When enabled, published events are temporarily stored so new
150150
* subscribers can resume from a previous position using `lastEventId`.
@@ -154,7 +154,7 @@ const publisher = new RedisPublisher<Events>(client, {
154154
enabled: false,
155155

156156
/**
157-
* How long (in seconds) to retain events for replay.
157+
* How long (in seconds) to retain events for resume.
158158
*
159159
* Expired events are cleaned up lazily for performance reasons, so
160160
* some events may remain available slightly longer than this period.
@@ -187,9 +187,9 @@ const publisher = new UpstashPublisher<Events>(redis, {
187187
*/
188188
serializer: undefined,
189189

190-
replay: {
190+
resume: {
191191
/**
192-
* Whether event replay support is enabled.
192+
* Whether event resume support is enabled.
193193
*
194194
* When enabled, published events are temporarily stored so new
195195
* subscribers can resume from a previous position using `lastEventId`.
@@ -199,7 +199,7 @@ const publisher = new UpstashPublisher<Events>(redis, {
199199
enabled: false,
200200

201201
/**
202-
* How long (in seconds) to retain events for replay.
202+
* How long (in seconds) to retain events for resume.
203203
*
204204
* Expired events are cleaned up lazily for performance reasons, so
205205
* some events may remain available slightly longer than this period.
@@ -239,9 +239,9 @@ const publisher = new BunRedisPublisher<Events>(redis, {
239239
*/
240240
serializer: undefined,
241241

242-
replay: {
242+
resume: {
243243
/**
244-
* Whether event replay support is enabled.
244+
* Whether event resume support is enabled.
245245
*
246246
* When enabled, published events are temporarily stored so new
247247
* subscribers can resume from a previous position using `lastEventId`.
@@ -251,7 +251,7 @@ const publisher = new BunRedisPublisher<Events>(redis, {
251251
enabled: false,
252252

253253
/**
254-
* How long (in seconds) to retain events for replay.
254+
* How long (in seconds) to retain events for resume.
255255
*
256256
* Expired events are cleaned up lazily for performance reasons, so
257257
* some events may remain available slightly longer than this period.
@@ -269,9 +269,9 @@ import { DurablePublisher, DurablePublisherObject } from '@orpc/cloudflare'
269269
export class PublisherDO extends DurablePublisherObject {
270270
constructor(ctx: DurableObjectState, env: Env) {
271271
super(ctx, env, {
272-
replay: {
272+
resume: {
273273
/**
274-
* Whether event replay support is enabled.
274+
* Whether event resume support is enabled.
275275
*
276276
* When enabled, published events are temporarily stored so new
277277
* subscribers can resume from a previous position using `lastEventId`.
@@ -281,7 +281,7 @@ export class PublisherDO extends DurablePublisherObject {
281281
enabled: false,
282282

283283
/**
284-
* How long (in seconds) to retain events for replay.
284+
* How long (in seconds) to retain events for resume.
285285
*
286286
* Expired events are cleaned up lazily for performance reasons, so
287287
* some events may remain available slightly longer than this period.
@@ -343,26 +343,26 @@ export default {
343343

344344
:::
345345

346-
## Replay Missing Events
346+
## Resume Missing Events
347347

348-
Some adapters can replay events missed while a subscriber is offline. This feature is usually disabled by default, but you can enable it when creating the publisher. When enabled, the publisher automatically manages event ids and attempts to replay events since the last event id provided by the subscriber.
348+
Some adapters can resume events missed while a subscriber is offline. This feature is usually disabled by default, but you can enable it when creating the publisher. When enabled, the publisher automatically manages event ids and attempts to deliver events since the last event id provided by the subscriber.
349349

350350
```ts
351351
const publisher = new MemoryPublisher({
352-
replay: {
353-
enabled: true, // Enable replaying missed events
352+
resume: {
353+
enabled: true, // Enable resuming missed events
354354
seconds: 60 * 5, // TTL in seconds
355355
}
356356
})
357357

358358
const iterator = publisher.subscribe('something-updated', {
359359
signal,
360-
lastEventId, // The publisher will attempt to replay missed events since this event id
360+
lastEventId, // The publisher will attempt to deliver missed events since this event id
361361
})
362362
```
363363

364364
::: warning
365-
When replay is enabled, the publisher manages event ids automatically. This means:
365+
When resume is enabled, the publisher manages event ids automatically. This means:
366366

367367
- Any event id provided during publishing is ignored
368368
- When subscribing, you must preserve and forward the event id when yielding custom payloads

‎packages/bun/src/redis-publisher.test.ts‎

Lines changed: 23 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
4545
return publisher
4646
}
4747

48-
it('delivers live events and meta (without replay and without prefix)', async () => {
48+
it('delivers live events and meta (without resume and without prefix)', async () => {
4949
const publisher = createTestingPublisher({ prefix: undefined })
5050
const liveEvent = `${crypto.randomUUID()}live`
5151
const ignoredEvent = `${crypto.randomUUID()}ignored`
@@ -71,19 +71,19 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
7171

7272
await unsubscribe()
7373

74-
const replayAttempt = vi.fn()
75-
const unsubscribeReplayAttempt = await publisher.subscribe(liveEvent, replayAttempt, { lastEventId: '0' })
74+
const resumeAttempt = vi.fn()
75+
const unsubscribeResumeAttempt = await publisher.subscribe(liveEvent, resumeAttempt, { lastEventId: '0' })
7676

7777
await sleep(300)
7878

79-
expect(replayAttempt).not.toHaveBeenCalled()
79+
expect(resumeAttempt).not.toHaveBeenCalled()
8080

81-
await unsubscribeReplayAttempt()
81+
await unsubscribeResumeAttempt()
8282
}, { timeout: 20_000 })
8383

84-
it('replays missed events in order and preserves event metadata while rewriting ids', async () => {
84+
it('resumes missed events in order and preserves event metadata while rewriting ids', async () => {
8585
const publisher = createTestingPublisher({
86-
replay: { enabled: true, seconds: 10 },
86+
resume: { enabled: true, seconds: 10 },
8787
})
8888
const event = 'orders'
8989
const liveListener = vi.fn()
@@ -117,26 +117,26 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
117117

118118
await unsubscribeLive()
119119

120-
const replayed = vi.fn()
121-
const unsubscribeReplay = await publisher.subscribe(event, replayed, {
120+
const resumed = vi.fn()
121+
const unsubscribeResume = await publisher.subscribe(event, resumed, {
122122
lastEventId: getEventMeta(secondDelivered)?.id,
123123
})
124124

125125
await waitFor(() => {
126-
expect(replayed).toHaveBeenCalledTimes(1)
126+
expect(resumed).toHaveBeenCalledTimes(1)
127127
})
128128

129-
const replayedPayload = replayed.mock.calls[0]![0]
129+
const resumedPayload = resumed.mock.calls[0]![0]
130130

131-
expect(replayedPayload).toEqual(thirdDelivered)
132-
expect(getEventMeta(replayedPayload)?.id).toEqual(getEventMeta(thirdDelivered)?.id)
131+
expect(resumedPayload).toEqual(thirdDelivered)
132+
expect(getEventMeta(resumedPayload)?.id).toEqual(getEventMeta(thirdDelivered)?.id)
133133

134-
await unsubscribeReplay()
134+
await unsubscribeResume()
135135
}, { timeout: 20_000 })
136136

137-
it('treats an empty replay backlog as a clean starting point and then continues with live messages', async () => {
137+
it('treats an empty resume backlog as a clean starting point and then continues with live messages', async () => {
138138
const publisher = createTestingPublisher({
139-
replay: { enabled: true, seconds: 10 },
139+
resume: { enabled: true, seconds: 10 },
140140
})
141141
const event = 'empty-backlog'
142142
const listener = vi.fn()
@@ -158,7 +158,7 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
158158
await unsubscribe()
159159
}, { timeout: 20_000 })
160160

161-
it('deduplicates events that race between replay and live delivery during reconnect', async () => {
161+
it('deduplicates events that race between resume and live delivery during reconnect', async () => {
162162
const { resolve, promise } = promiseWithResolvers<void>()
163163
const delayedRedis = new Proxy(new RedisClient(REDIS_URL), {
164164
get(target, p) {
@@ -182,7 +182,7 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
182182
})
183183

184184
const publisher = createTestingPublisher({
185-
replay: { enabled: true, seconds: 10 },
185+
resume: { enabled: true, seconds: 10 },
186186
}, { useRedis: delayedRedis })
187187
const event = 'timeline'
188188

@@ -207,12 +207,12 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
207207
await unsubscribe()
208208
}, { timeout: 20_000 })
209209

210-
it('trims stale replay history on the next publish and lets Redis expire the stream key', async () => {
210+
it('trims stale resume history on the next publish and lets Redis expire the stream key', async () => {
211211
const prefix = `retention:${crypto.randomUUID()}:`
212212
const event = 'orders'
213213
const key = `${prefix}${event}`
214214
const publisher = createTestingPublisher({
215-
replay: { enabled: true, seconds: 1 },
215+
resume: { enabled: true, seconds: 1 },
216216
prefix,
217217
})
218218

@@ -262,7 +262,7 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
262262
const prefix = `custom:${crypto.randomUUID()}:`
263263
const event = 'profile-updated'
264264
const publisher = createTestingPublisher({
265-
replay: { enabled: true, seconds: 10 },
265+
resume: { enabled: true, seconds: 10 },
266266
serializer,
267267
prefix,
268268
})
@@ -300,9 +300,9 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
300300
await unsubscribe()
301301
}, { timeout: 20_000 })
302302

303-
it('reports replay errors while subscribing', async () => {
303+
it('reports resume errors while subscribing', async () => {
304304
const publisher = createTestingPublisher({
305-
replay: { enabled: true, seconds: 10 },
305+
resume: { enabled: true, seconds: 10 },
306306
})
307307
const event = 'resume-errors'
308308
const listener = vi.fn()

‎packages/bun/src/redis-publisher.ts‎

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -32,16 +32,16 @@ export interface BunRedisPublisherOptions extends PublisherOptions {
3232
serializer?: undefined | Pick<RPCSerializer, keyof RPCSerializer>
3333

3434
/**
35-
* Configuration for event replay support.
35+
* Configuration for event resume support.
3636
*
3737
* When enabled, published events are temporarily stored so new
3838
* subscribers can resume from a previous position using `lastEventId`.
3939
*
4040
* @default { enabled: false }
4141
*/
42-
replay?: {
42+
resume?: {
4343
/**
44-
* Whether event replay support is enabled.
44+
* Whether event resume support is enabled.
4545
*
4646
* When enabled, published events are temporarily stored so new
4747
* subscribers can resume from a previous position using `lastEventId`.
@@ -51,7 +51,7 @@ export interface BunRedisPublisherOptions extends PublisherOptions {
5151
enabled: boolean
5252

5353
/**
54-
* How long (in seconds) to retain events for replay.
54+
* How long (in seconds) to retain events for resume.
5555
*
5656
* Expired events are cleaned up lazily for performance reasons, so
5757
* some events may remain available slightly longer than this period.
@@ -66,8 +66,8 @@ export class BunRedisPublisher<T extends Record<string, object>> extends Publish
6666
private subscriber: BunRedisPublisherOptions['subscriber']
6767
private readonly prefix: Exclude<BunRedisPublisherOptions['prefix'], undefined>
6868
private readonly serializer: Exclude<BunRedisPublisherOptions['serializer'], undefined>
69-
private readonly replayEnabled: boolean
70-
private readonly replaySeconds: number
69+
private readonly resumeEnabled: boolean
70+
private readonly resumeSeconds: number
7171

7272
/**
7373
* The exactness of the `XTRIM` command.
@@ -83,8 +83,8 @@ export class BunRedisPublisher<T extends Record<string, object>> extends Publish
8383

8484
this.prefix = options.prefix ?? ''
8585
this.serializer = options.serializer ?? new RPCSerializer()
86-
this.replayEnabled = options.replay?.enabled ?? false
87-
this.replaySeconds = options.replay?.seconds ?? 300
86+
this.resumeEnabled = options.resume?.enabled ?? false
87+
this.resumeSeconds = options.resume?.seconds ?? 300
8888
this.subscriber = options.subscriber
8989
}
9090

@@ -94,13 +94,13 @@ export class BunRedisPublisher<T extends Record<string, object>> extends Publish
9494
const data = this.serializePayload(payload)
9595
let id: string | undefined
9696

97-
if (this.replayEnabled) {
97+
if (this.resumeEnabled) {
9898
const now = Date.now()
9999

100-
// Remove expired replay windows.
100+
// Remove expired resume windows.
101101
// The next publish for a stale event will perform trimming again.
102102
for (const [event, firstPublishTime] of this.firstPublishTimeMap) {
103-
if (firstPublishTime + this.replaySeconds * 1000 < now) {
103+
if (firstPublishTime + this.resumeSeconds * 1000 < now) {
104104
this.firstPublishTimeMap.delete(event)
105105
}
106106
}
@@ -110,10 +110,10 @@ export class BunRedisPublisher<T extends Record<string, object>> extends Publish
110110

111111
const result = await Promise.all([
112112
this.redis.send('XADD', [redisKey, '*', 'data', stringifyJSON(data)]),
113-
this.redis.send('XTRIM', [redisKey, 'MINID', this.xTrimExactness, `${now - this.replaySeconds * 1000}-0`]),
114-
// Use a 2x TTL so events published near the end of the replay window
113+
this.redis.send('XTRIM', [redisKey, 'MINID', this.xTrimExactness, `${now - this.resumeSeconds * 1000}-0`]),
114+
// Use a 2x TTL so events published near the end of the resume window
115115
// are not expired before the next window updates the key expiration.
116-
this.redis.expire(redisKey, this.replaySeconds * 2),
116+
this.redis.expire(redisKey, this.resumeSeconds * 2),
117117
])
118118

119119
id = result[0] as string
@@ -134,7 +134,7 @@ export class BunRedisPublisher<T extends Record<string, object>> extends Publish
134134
const redisKey = `${this.prefix}${event}`
135135

136136
let pendingPayloads: T[K][] | undefined = []
137-
const replayedIds = new Set<string>()
137+
const resumedIds = new Set<string>()
138138

139139
const deduplicatingListener = (payload: T[K]) => {
140140
if (pendingPayloads) {
@@ -143,7 +143,7 @@ export class BunRedisPublisher<T extends Record<string, object>> extends Publish
143143
}
144144

145145
const id = getEventMeta(payload)?.id
146-
if (id !== undefined && replayedIds.has(id)) { // Already delivered through replay.
146+
if (id !== undefined && resumedIds.has(id)) { // Already delivered during resume.
147147
return
148148
}
149149

@@ -169,7 +169,7 @@ export class BunRedisPublisher<T extends Record<string, object>> extends Publish
169169
const subscribePromise = subscriber.subscribe(redisKey, redisListener)
170170

171171
try {
172-
if (this.replayEnabled && lastEventId !== undefined) {
172+
if (this.resumeEnabled && lastEventId !== undefined) {
173173
/**
174174
* [Object: null prototype] {
175175
* "redis:9d1536ca-8952-4e35-ae79-d466074f9436:orders": [
@@ -186,7 +186,7 @@ export class BunRedisPublisher<T extends Record<string, object>> extends Publish
186186
const rawData = message[1]
187187
const data = parseEmptyableJSON(rawData as string)
188188
const payload = this.deserializePayload(id, data as any)
189-
replayedIds.add(id)
189+
resumedIds.add(id)
190190
originalListener(payload as T[K])
191191
}
192192
}

0 commit comments

Comments
 (0)