CachingPubSub
CachingPubSub 包装任意 PubSub 实现，并添加事件缓存和 replay。它在 cache 中记录每个 topic 的每个已发布事件，因此晚连接或断开后重新连接的订阅者可以先 replay 错过的事件，再继续接收实时事件。
将它用于在不保留历史记录的 transport（例如 EventEmitterPubSub）上构建可恢复流。已经持久化事件的 transport（例如 RedisStreamsPubSub）不需要此包装器。
CachingPubSub 对批处理透明：subscribe() 将 options.batch 转发给内部 pub/sub，且 supportsNativeBatching 镜像内部值。即使传入 options.batch，包装不支持批处理的内部实现仍会导致非批量投递。
使用示例使用示例的直接链接
包装内部 pub/sub，并提供用于存储事件的 server cache。
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 helper 返回同一个实例，且可读性更好：
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 contract。它会重写 replay 方法以读取缓存事件。以下方法描述缓存行为。
publish(topic, event)publishtopic-event的直接链接
使用顺序 index 缓存事件，然后将其发布到内部 pub/sub。
await pubsub.publish('my-topic', {
type: 'example',
data: { value: 1 },
runId: 'run-123',
})
subscribeWithReplay(topic, cb)subscribewithreplaytopic-cb的直接链接
replay topic 的所有缓存事件，然后订阅实时事件。
await pubsub.subscribeWithReplay('my-topic', event => {
console.log(event)
})
subscribeFromOffset(topic, offset, cb)subscribefromoffsettopic-offset-cb的直接链接
从 offset 开始 replay 缓存事件，然后订阅实时事件。当 client 知道最后位置时使用它，以避免 replay 整个历史记录。
await pubsub.subscribeFromOffset('my-topic', 42, event => {
console.log(event)
})
getHistory(topic, offset?)gethistorytopic-offset的直接链接
返回 topic 从 offset 开始的缓存事件。
const events = await pubsub.getHistory('my-topic', 0)
返回：Promise<Event[]>
clearTopic(topic)cleartopictopic的直接链接
移除 topic 的缓存事件和 offset 计数器，并将调用转发给内部 pub/sub，以便持久 transport（如 RedisStreamsPubSub）也能删除其保留状态。Mastra 的 run lifecycle 会在 run 完成时自动调用此方法。
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