GoogleCloudPubSub

GoogleCloudPubSub 是以 Google Cloud Pub/Sub 為基礎的 PubSub 實作。它使用 Google Cloud topic 與 subscription，在不同 process 與主機之間傳遞 event，並支援依序傳遞與訊息確認。

適用於 Google Cloud 上的分散式部署。單一 process 傳遞請使用 EventEmitterPubSub；Redis 請使用 RedisStreamsPubSub。

每個 topic 都會對應至 Google Cloud topic。設定 group 的 subscription 會共用同一 subscription，因此成員會競爭 event。未設定 group 的 subscription 會建立各 instance 專用的 subscription，因此每個 instance 都會收到每個 event。

安裝

npm

npm install @mastra/google-cloud-pubsub

pnpm

pnpm add @mastra/google-cloud-pubsub

Yarn

yarn add @mastra/google-cloud-pubsub

Bun

bun add @mastra/google-cloud-pubsub

使用範例

傳入 Google Cloud client 設定，例如 project ID。

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { GoogleCloudPubSub } from '@mastra/google-cloud-pubsub'

export const mastra = new Mastra({
  pubsub: new GoogleCloudPubSub({
    projectId: 'my-project',
  }),
})

Constructor 參數

config:

ClientConfig

Google Cloud Pub/Sub client 的設定，包括憑證與 project ID。所有欄位請參閱 @google-cloud/pubsub client 文件。

方法

GoogleCloudPubSub 實作 PubSub contract。以下是此實作專屬的方法。

init(topicName, group?)

如果 topic 與 subscription 尚不存在，此方法會加以建立，並傳回 subscription。subscribe 會在內部呼叫此方法，因此你很少需要直接呼叫。

await pubsub.init('workflow.events')

subscribe(topic, cb, options?)

訂閱 topic。設定 options.group 後，group 成員會共用 subscription 並競爭 event。未設定 group 時，此 instance 會透過自己的 subscription 收到每個 event。

await pubsub.subscribe('workflow.events', (event, ack, nack) => {
  console.log(event)
})

flush()

等待待處理的確認完成。

await pubsub.flush()

destroy(topicName)

移除 topic 名稱所對應的 subscription 與 topic。使用此方法清理 Google Cloud 資源。

await pubsub.destroy('workflow.events')

確認

每個傳遞的 event 都包含 ack 與 nack 函式。處理成功後呼叫 ack，即可從 subscription 移除 event。如果兩者都未呼叫，Google Cloud 會在確認期限到期後重新傳遞 event。