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

# CachingPubSub

`CachingPubSub` 會包裝任何 [`PubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/base) 實作，並新增 event 快取與重播功能。它會按 topic 在 cache 中記錄每個發布的 event，因此較晚連線的 subscriber，或斷線後重新連線的 subscriber，可以先重播錯過的 event，再繼續接收即時 event。

適合在不保留歷史記錄的 transport（例如 [`EventEmitterPubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/event-emitter)）之上建立可恢復串流。已持久化 event 的 transport（例如 [`RedisStreamsPubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/redis-streams)）不需要此 wrapper。

`CachingPubSub` 不會干預批次處理：`subscribe()` 會將 `options.batch` 轉送至內部 pub/sub，而 `supportsNativeBatching` 會反映內部 instance 的值。即使傳入 `options.batch`，包裝不支援批次處理的內部 instance 仍會以非批次方式傳遞。

## 使用範例

包裝內部 pub/sub，並提供 server cache 以儲存 event。

```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` 輔助函式會傳回相同的 instance，inline 包裝時也更易讀：

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

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

## Constructor 參數

**inner** (`PubSub`): 要包裝的 pub/sub 實作。所有 publish 與即時 subscription 都會通過此 instance。

**cache** (`MastraServerCache`): 用於按 topic 儲存 event 以供重播的 cache。

**options** (`CachingPubSubOptions`): 選填設定。

## 屬性

**supportsNativeBatching** (`boolean`): 反映內部 pub/sub 的值。只有包裝的實作支援批次處理時才傳回 true。

## 方法

`CachingPubSub` 實作 [`PubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/base) contract。它會覆寫重播方法，以讀取快取的 event。以下方法說明快取行為。

### `publish(topic, event)`

以連續 index 快取 event，再將它發布至內部 pub/sub。

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

### `subscribeWithReplay(topic, cb)`

重播 topic 的所有快取 event，再訂閱即時 event。

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

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

從 `offset` 開始重播快取 event，再訂閱即時 event。client 知道最後位置時請使用此方法，以免重播完整歷史記錄。

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

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

傳回 topic 從 `offset` 開始的快取 event。

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

傳回：`Promise<Event[]>`

### `clearTopic(topic)`

移除 topic 的快取 event 與 offset counter，並將呼叫轉送至內部 pub/sub，讓持久化 transport（例如 [`RedisStreamsPubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/redis-streams)）也能刪除保留的 state。run 完成時，Mastra 的 run 生命週期會自動呼叫此方法。

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

## 函式

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

用於建構 `CachingPubSub` 的便利 wrapper。接受與 constructor 相同的引數，並傳回新的 instance。

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

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

傳回：`CachingPubSub`