CachingPubSub
CachingPubSub 會包裝任何 PubSub 實作，並加入事件快取和重播功能。它會按主題在快取中記錄每個已發佈事件，因此較遲連線或中斷後重新連線的訂閱者，可以先重播錯過的事件，再繼續接收即時事件。
你可以在不保留歷史記錄的傳輸（例如 EventEmitterPubSub）之上，使用它建立可恢復串流。已持久保存事件的傳輸（例如 RedisStreamsPubSub）則不需要此包裝器。
CachingPubSub 對批次處理透明：subscribe() 會將 options.batch 轉交內部 pub/sub，而 supportsNativeBatching 會反映內部實作的值。若內部實作不支援批次處理，即使傳入 options.batch，事件仍會逐一傳送。
使用範例使用範例 的直接連結
包裝內部 pub/sub，並提供用於儲存事件的伺服器快取。
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,
})
以 inline 方式包裝時，withCaching 輔助函數會傳回相同執行個體，令程式碼更易讀：
import { withCaching, EventEmitterPubSub } from '@mastra/core/events'
import { InMemoryServerCache } from '@mastra/core/cache'
const pubsub = withCaching(new EventEmitterPubSub(), new InMemoryServerCache())
建構函數參數建構函數參數 的直接連結
inner:
cache:
options?:
屬性屬性 的直接連結
supportsNativeBatching:
true。方法方法 的直接連結
CachingPubSub 實作 PubSub 合約。它會覆寫重播方法以讀取快取事件。以下方法說明快取行為。
publish(topic, event)publishtopic-event 的直接連結
以連續索引快取事件，然後將它發佈至內部 pub/sub。
await pubsub.publish('my-topic', {
type: 'example',
data: { value: 1 },
runId: 'run-123',
})
subscribeWithReplay(topic, cb)subscribewithreplaytopic-cb 的直接連結
重播主題的所有快取事件，然後訂閱即時事件。
await pubsub.subscribeWithReplay('my-topic', event => {
console.log(event)
})
subscribeFromOffset(topic, offset, cb)subscribefromoffsettopic-offset-cb 的直接連結
從 offset 開始重播快取事件，然後訂閱即時事件。當用戶端知道上次位置時使用此方法，可避免重播完整歷史記錄。
await pubsub.subscribeFromOffset('my-topic', 42, event => {
console.log(event)
})
getHistory(topic, offset?)gethistorytopic-offset 的直接連結
傳回主題從 offset 開始的快取事件。
const events = await pubsub.getHistory('my-topic', 0)
傳回：Promise<Event[]>
clearTopic(topic)cleartopictopic 的直接連結
移除主題的快取事件及 offset 計數器，並將呼叫轉交內部 pub/sub，讓持久傳輸（例如 RedisStreamsPubSub）亦可刪除保留的狀態。Mastra 的執行生命週期會在一次執行完成時自動呼叫此方法。
await pubsub.clearTopic('my-topic')
函數函數 的直接連結
withCaching(pubsub, cache, options?)withcachingpubsub-cache-options 的直接連結
用於建構 CachingPubSub 的便利包裝器。它接受與建構函數相同的引數，並傳回新執行個體。
import { withCaching, EventEmitterPubSub } from '@mastra/core/events'
import { InMemoryServerCache } from '@mastra/core/cache'
const pubsub = withCaching(new EventEmitterPubSub(), new InMemoryServerCache(), {
keyPrefix: 'events:',
})
傳回：CachingPubSub