GoogleCloudPubSub

GoogleCloudPubSub est une implémentation de PubSub fondée sur Google Cloud Pub/Sub. Elle distribue les événements entre plusieurs processus et hôtes à l’aide de topics et d’abonnements Google Cloud, avec une livraison ordonnée et un acquittement des messages.

Utilisez-la pour les déploiements distribués sur Google Cloud. Pour une distribution au sein d’un seul processus, utilisez EventEmitterPubSub. Pour Redis, utilisez RedisStreamsPubSub.

Chaque topic correspond à un topic Google Cloud. Les abonnements associés à un groupe partagent un même abonnement ; les membres se répartissent donc les événements. Les abonnements sans groupe créent un abonnement par instance, si bien que chaque instance reçoit tous les événements.

Installation

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

Exemple d’utilisation

Transmettez une configuration de client Google Cloud, telle qu’un ID de projet.

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',
  }),
})

Paramètres du constructeur

config:

ClientConfig

Configuration du client Google Cloud Pub/Sub, comprenant les identifiants d’authentification et l’ID du projet. Consultez la documentation du client @google-cloud/pubsub pour connaître tous les champs.

Méthodes

GoogleCloudPubSub implémente le contrat PubSub. Les méthodes ci-dessous sont propres à cette implémentation.

init(topicName, group?)

Crée le topic et un abonnement s’ils n’existent pas encore, puis renvoie l’abonnement. subscribe appelle cette méthode en interne ; vous avez donc rarement besoin de l’appeler directement.

await pubsub.init('workflow.events')

subscribe(topic, cb, options?)

S’abonne à un topic. Avec options.group, les membres du groupe partagent un abonnement et se répartissent les événements. Sans groupe, l’instance reçoit chaque événement par l’intermédiaire de son propre abonnement.

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

flush()

Attend la fin des acquittements en attente.

await pubsub.flush()

destroy(topicName)

Supprime l’abonnement et le topic correspondant à un nom de topic. Utilisez cette méthode pour nettoyer les ressources Google Cloud.

await pubsub.destroy('workflow.events')

Acquittement

Chaque événement distribué comprend les fonctions ack et nack. Appelez ack après un traitement réussi afin de retirer l’événement de l’abonnement. Si aucune des deux fonctions n’est appelée, Google Cloud distribue à nouveau l’événement à l’expiration de son délai d’acquittement.