CachingPubSub
CachingPubSub は任意の PubSub 実装をラップし、イベントのキャッシュとリプレイを追加します。公開されたすべてのイベントをトピックごとにキャッシュへ記録するため、遅れて接続した subscriber や、切断後に再接続した subscriber は、受信できなかったイベントをリプレイしてから live イベントの受信を続けられます。
履歴を保持しない EventEmitterPubSub などの transport 上に、再開可能なストリームを構築する場合に使用します。RedisStreamsPubSub など、イベントをすでに永続化する transport にはこのラッパーは必要ありません。
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 の規約を実装します。リプレイメソッドをオーバーライドし、キャッシュされたイベントを読み取ります。以下のメソッドでは、キャッシュに関する動作を説明します。
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への直接リンク
トピックにキャッシュされたすべてのイベントをリプレイしてから、live イベントを購読します。
await pubsub.subscribeWithReplay('my-topic', event => {
console.log(event)
})
subscribeFromOffset(topic, offset, cb)subscribefromoffsettopic-offset-cbへの直接リンク
offset からキャッシュ済みイベントをリプレイし、その後 live イベントを購読します。クライアントが最後の位置を把握していて、履歴全体のリプレイを避けたい場合に使用します。
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 counter を削除し、その呼び出しを内部の pub/sub に転送します。これにより、RedisStreamsPubSub などの永続 transport も、保持している状態を削除できます。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