캐싱PubSub

CachingPubSub무엇이든 래핑PubSub구현하고 이벤트 캐싱 및 재생을 추가합니다. 주제별로 게시된 모든 이벤트를 캐시에 기록하므로 늦게 연결하거나 연결이 끊긴 후 다시 연결하는 구독자는 라이브 이벤트를 계속하기 전에 놓친 이벤트를 재생할 수 있습니다.

이를 사용하여 EventEmitterPubSub처럼 기록을 유지하지 않는 전송 위에 재개 가능한 스트림을 구축하세요. RedisStreamsPubSub처럼 이미 이벤트를 영구 저장하는 전송에는 이 래퍼가 필요하지 않습니다. CachingPubSub은 배치 처리에 투명하게 작동합니다. subscribe()는 options.batch를 내부 pub/sub에 전달하고, supportsNativeBatching은 내부 구현의 값을 반영합니다. 배치를 지원하지 않는 내부 구현을 래핑하면 options.batch가 전달되더라도 배치 전달이 활성화되지 않습니다.

사용예

내부 pub/sub를 래핑하고 이벤트를 저장할 서버 캐시를 제공합니다.

src/mastra/index.ts

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 헬퍼는 동일한 인스턴스를 반환하며 인라인으로 래핑할 때 더 읽기 좋습니다.

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

재생을 위해 주제별 이벤트를 저장하는 데 사용하는 캐시입니다.

options?:

CachingPubSubOptions

선택적 구성입니다.

string

IMastraLogger

속성

supportsNativeBatching:

boolean

내부 pub/sub의 동작을 반영합니다. 래핑된 구현이 일괄 처리를 지원하는 경우에만 true를 반환합니다.

행동 양식

CachingPubSub는 PubSub 계약을 구현합니다. 캐시된 이벤트를 읽도록 재생 메서드를 재정의합니다. 아래 메서드는 캐싱 동작을 설명합니다.

publish(topic, event)

순차 인덱스로 이벤트를 캐시한 다음 내부 pub/sub에 게시합니다.

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

subscribeWithReplay(topic, cb)

주제에 대해 캐시된 모든 이벤트를 재생한 다음 라이브 이벤트를 구독합니다.

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

subscribeFromOffset(topic, offset, cb)

offset부터 캐시된 이벤트를 재생한 후 실시간 이벤트를 구독합니다. 전체 기록을 다시 재생하지 않도록 클라이언트가 마지막 위치를 알고 있을 때 사용하세요.

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

getHistory(topic, offset?)

다음에서 시작하여 주제에 대해 캐시된 이벤트를 반환합니다.offset.

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

보고:Promise<Event[]>

clearTopic(topic)

주제의 캐시된 이벤트와 오프셋 카운터를 제거하고 호출을 내부 pub/sub로 전달하므로 RedisStreamsPubSub 같은 영구 전송 구현도 보존된 상태를 삭제할 수 있습니다. Mastra의 실행 수명 주기는 실행이 끝날 때 이를 자동으로 호출합니다.

await pubsub.clearTopic('my-topic')

기능

withCaching(pubsub, 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