CachingPubSub
CachingPubSub 會包裝任何 PubSub 實作，並新增 event 快取與重播功能。它會按 topic 在 cache 中記錄每個發布的 event，因此較晚連線的 subscriber，或斷線後重新連線的 subscriber，可以先重播錯過的 event，再繼續接收即時 event。
適合在不保留歷史記錄的 transport（例如 EventEmitterPubSub）之上建立可恢復串流。已持久化 event 的 transport（例如 RedisStreamsPubSub）不需要此 wrapper。
CachingPubSub 不會干預批次處理：subscribe() 會將 options.batch 轉送至內部 pub/sub，而 supportsNativeBatching 會反映內部 instance 的值。即使傳入 options.batch，包裝不支援批次處理的內部 instance 仍會以非批次方式傳遞。
使用範例「使用範例」的直接連結
包裝內部 pub/sub，並提供 server cache 以儲存 event。
import { Mastra } from '@mastra/core'
import { CachingPubSub, EventEmitterPubSub } from '@mastra/core/events'
import { InMemoryServerCache } from '@mastra/core/cache'
const cache = new InMemoryServerCache()
const pubsub = new CachingPubSub(new EventEmitterPubSub(), cache)
export const mastra = new Mastra({
pubsub,
})
withCaching 輔助函式會傳回相同的 instance，inline 包裝時也更易讀：
import { withCaching, EventEmitterPubSub } from '@mastra/core/events'
import { InMemoryServerCache } from '@mastra/core/cache'
const pubsub = withCaching(new EventEmitterPubSub(), new InMemoryServerCache())
Constructor 參數「Constructor 參數」的直接連結
inner:
cache:
options?:
屬性「屬性」的直接連結
supportsNativeBatching:
true。方法「方法」的直接連結
CachingPubSub 實作 PubSub contract。它會覆寫重播方法，以讀取快取的 event。以下方法說明快取行為。
publish(topic, event)「publishtopic-event」的直接連結
以連續 index 快取 event，再將它發布至內部 pub/sub。
await pubsub.publish('my-topic', {
type: 'example',
data: { value: 1 },
runId: 'run-123',
})
subscribeWithReplay(topic, cb)「subscribewithreplaytopic-cb」的直接連結
重播 topic 的所有快取 event，再訂閱即時 event。
await pubsub.subscribeWithReplay('my-topic', event => {
console.log(event)
})
subscribeFromOffset(topic, offset, cb)「subscribefromoffsettopic-offset-cb」的直接連結
從 offset 開始重播快取 event，再訂閱即時 event。client 知道最後位置時請使用此方法，以免重播完整歷史記錄。
await pubsub.subscribeFromOffset('my-topic', 42, event => {
console.log(event)
})
getHistory(topic, offset?)「gethistorytopic-offset」的直接連結
傳回 topic 從 offset 開始的快取 event。
const events = await pubsub.getHistory('my-topic', 0)
傳回：Promise<Event[]>
clearTopic(topic)「cleartopictopic」的直接連結
移除 topic 的快取 event 與 offset counter，並將呼叫轉送至內部 pub/sub，讓持久化 transport（例如 RedisStreamsPubSub）也能刪除保留的 state。run 完成時，Mastra 的 run 生命週期會自動呼叫此方法。
await pubsub.clearTopic('my-topic')
函式「函式」的直接連結
withCaching(pubsub, cache, options?)「withcachingpubsub-cache-options」的直接連結
用於建構 CachingPubSub 的便利 wrapper。接受與 constructor 相同的引數，並傳回新的 instance。
import { withCaching, EventEmitterPubSub } from '@mastra/core/events'
import { InMemoryServerCache } from '@mastra/core/cache'
const pubsub = withCaching(new EventEmitterPubSub(), new InMemoryServerCache(), {
keyPrefix: 'events:',
})
傳回：CachingPubSub