PubSubStore — Google Cloud Pub/Sub

PubSubStore — Google Cloud Pub/Sub Overview Google Cloud Pub/Sub is a managed, topic-and-subscription messaging service. There is no query language and no SQL-shaped access pattern: publishers send messages to a topic, and subscribers receive them on a named subscription attached to that topic. Because none of that fits the SQL-shaped Dialect interface, PubSubStore implements the minimal NoSqlStore marker interface (src/nosql/store.ts) — connection lifecycle plus a getClient() escape hatch — and exposes Pub/Sub's own operations directly rather than forcing them into a query(sql) shape. Identity: Property | Value | --------- | ----------------------- | name | 'pubsub' | library | '@google-cloud/pubsub'| The store is built on @google-cloud/pubsub: new PubSub() yields a client whose topic(name).publishMessage(...) publishes and whose subscription(name).on('message', ...) streams messages. There is no command-object pattern — the driver exposes methods directly. Lazy loading — not a hard dependency @google-cloud/pubsub is not a hard dependency of this package. The driver is loaded lazily via require() inside connect(), rather than a top-level import. Importing this module therefore does not require the driver to be installed — it is only needed when a Pub/Sub store is actually connected. Injected client PubSubStoreOptions accepts a pre-built client (a PubSub client or any object exposing the methods below, typed as PubSubClientLike). When provided, connect() uses it directly and does not require the driver. This is how the test suite injects a mock client (no driver, no network). Message encoding publish() accepts a string (UTF-8 encoded), a Buffer (sent as-is), or any other JSON-serializable value (JSON.stringifyd then UTF-8 encoded). The resulting bytes become the message data. Connection Build a store from connection options and call connect(): All connection options are optional: Option | Type | Purpose | ------------- | ------------------ | ----------------------------------------------------------------------------------- | projectId | string | GCP project id. | keyFilename | string | Path to a service-account key file. | apiEndpoint | string | Override the API endpoint, e.g. 'localhost:8085' for the emulator. | client | PubSubClientLike | A pre-built client to use directly (mock or custom config). When set, the driver is not required. | Injected-client form Supply your own client (or a mock) to bypass driver loading: Methods Driver failures are wrapped in a DatabaseError (preserving the original error); using a method before connect() (or after disconnect()) throws a ConnectionError. Lifecycle Method | Signature | Behavior | ------------- | ------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- | connect | connect(): Promise<void> | Idempotent (returns early if already connected). Uses an injected client if provided, otherwise lazy-requires @google-cloud/pubsub and builds a PubSub client. Wraps failures in ConnectionError. | disconnect | disconnect(): Promise<void> | Detaches listeners and closes each tracked subscription, closes the client, and clears state. | isConnected | isConnected(): boolean | true only when connected and a client is present. | getClient | getClient(): PubSubClientLike | Returns the underlying PubSub client for operations not wrapped here. Throws ConnectionError if not connected. | Producing Method | Signature | Driver call | Behavior | --------- | ------------------------------------------------------------------------------------------------- | ------------------------------- | ------------------------------------------------------------------------------------------------------------ | publish | publish(topic: string, message: PubSubMessage, attributes?: Record<string, string>): Promise<string> | topic(topic).publishMessage() | Encodes message to a Buffer and publishes it (with optional attributes) to topic. Returns the server-assigned message id. | Administration Method | Signature | Driver call | Behavior | -------------------- | --------------------------------------------------------------------------- | ------------------------------ | ---------------------------------------------------------------- | createTopic | createTopic(name: string): Promise<void> | client.createTopic() | Creates a new topic. | createSubscription | createSubscription(topic: string, name: string, options?: unknown): Promise<void> | client.createSubscription() | Attaches a new subscription name to topic. options passed straight through. | Consuming Method | Signature | Driver call | Behavior | ----------- | -------------------------------------------------------------------------------------------------- | ------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | subscribe | subscribe(subscription: string, handler: (message: any) => void \| Promise<void>): Promise<PubSubSubscription> | subscription(name).on('message') | Attaches a 'message' event listener that invokes handler per streamed driver Message (call .ack() / .nack() on it). Returns a handle whose stop() detaches the listener and closes the subscription. Tracked so disconnect() detaches it. | Example Verification status Unit / mock-verified only. The tests in tests/nosql/pubsub.test.ts are fully mock-driven: a fake @google-cloud/pubsub client (with jest-spy topic(), subscription(), createTopic(), createSubscription(), close()) is injected via PubSubStoreOptions.client. The real driver is not installed, and there is no live Pub/Sub access and no network in the test run. What this proves: publish routes to topic(name), encodes the message to a Buffer (string → UTF-8, object → JSON), and returns the driver's message id. createTopic / createSubscription route to the matching client methods with the expected arguments. subscribe attaches a 'message' listener and forwards messages to the handler; stop() removes listeners. Lifecycle behavior: connecting via an injected client, disconnect() closing the client, and ConnectionError before connect(). Error handling: publish and subscribe failures wrapped in DatabaseError. What this does not prove: live execution against real Google Cloud Pub/Sub (or the emulator). The method-call shape is verified against the driver's documented contract, but end-to-end execution over the wire has not been exercised here. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories