> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt

# CachingPubSub

`CachingPubSub` 包装任意 [`PubSub`](https://mastra.zisheng.pro/reference/pubsub/base) 实现，并添加事件缓存和 replay。它在 cache 中记录每个 topic 的每个已发布事件，因此晚连接或断开后重新连接的订阅者可以先 replay 错过的事件，再继续接收实时事件。

将它用于在不保留历史记录的 transport（例如 [`EventEmitterPubSub`](https://mastra.zisheng.pro/reference/pubsub/event-emitter)）上构建可恢复流。已经持久化事件的 transport（例如 [`RedisStreamsPubSub`](https://mastra.zisheng.pro/reference/pubsub/redis-streams)）不需要此包装器。

`CachingPubSub` 对批处理透明：`subscribe()` 将 `options.batch` 转发给内部 pub/sub，且 `supportsNativeBatching` 镜像内部值。即使传入 `options.batch`，包装不支持批处理的内部实现仍会导致非批量投递。

## 使用示例

包装内部 pub/sub，并提供用于存储事件的 server cache。

```typescript
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 返回同一个实例，且可读性更好：

```typescript
import { withCaching, EventEmitterPubSub } from '@mastra/core/events'
import { InMemoryServerCache } from '@mastra/core/cache'

const pubsub = withCaching(new EventEmitterPubSub(), new InMemoryServerCache())
```

## 构造函数参数

**inner** (`PubSub`): 要包装的 pub/sub 实现。所有发布和实时订阅都会传递给此实例。

**cache** (`MastraServerCache`): 用于按 topic 存储事件以供 replay 的 cache。

**options** (`CachingPubSubOptions`): 可选配置。

## 属性

**supportsNativeBatching** (`boolean`): 镜像内部 pub/sub。仅当包装的实现支持批处理时才返回 true。

## 方法

`CachingPubSub` 实现 [`PubSub`](https://mastra.zisheng.pro/reference/pubsub/base) contract。它会重写 replay 方法以读取缓存事件。以下方法描述缓存行为。

### `publish(topic, event)`

使用顺序 index 缓存事件，然后将其发布到内部 pub/sub。

```typescript
await pubsub.publish('my-topic', {
  type: 'example',
  data: { value: 1 },
  runId: 'run-123',
})
```

### `subscribeWithReplay(topic, cb)`

replay topic 的所有缓存事件，然后订阅实时事件。

```typescript
await pubsub.subscribeWithReplay('my-topic', event => {
  console.log(event)
})
```

### `subscribeFromOffset(topic, offset, cb)`

从 `offset` 开始 replay 缓存事件，然后订阅实时事件。当 client 知道最后位置时使用它，以避免 replay 整个历史记录。

```typescript
await pubsub.subscribeFromOffset('my-topic', 42, event => {
  console.log(event)
})
```

### `getHistory(topic, offset?)`

返回 topic 从 `offset` 开始的缓存事件。

```typescript
const events = await pubsub.getHistory('my-topic', 0)
```

返回：`Promise<Event[]>`

### `clearTopic(topic)`

移除 topic 的缓存事件和 offset 计数器，并将调用转发给内部 pub/sub，以便持久 transport（如 [`RedisStreamsPubSub`](https://mastra.zisheng.pro/reference/pubsub/redis-streams)）也能删除其保留状态。Mastra 的 run lifecycle 会在 run 完成时自动调用此方法。

```typescript
await pubsub.clearTopic('my-topic')
```

## 函数

### `withCaching(pubsub, cache, options?)`

构造 `CachingPubSub` 的便捷包装器。接受与构造函数相同的参数，并返回新实例。

```typescript
import { withCaching, EventEmitterPubSub } from '@mastra/core/events'
import { InMemoryServerCache } from '@mastra/core/cache'

const pubsub = withCaching(new EventEmitterPubSub(), new InMemoryServerCache(), {
  keyPrefix: 'events:',
})
```

返回：`CachingPubSub`