PulsarStore — Apache Pulsar pub/sub & streaming

PulsarStore — Apache Pulsar pub/sub & streaming Overview Apache Pulsar is a distributed pub/sub messaging and streaming platform. It has no query language, no rows, and no SQL-shaped access pattern — data flows as messages (raw byte payloads) published to topics and consumed via subscriptions or readers. Because none of that fits the SQL-shaped Dialect interface, PulsarStore implements the minimal NoSqlStore marker interface (src/nosql/store.ts) — connection lifecycle plus a getClient() escape hatch — and exposes Pulsar's own operations (produce / subscribe / createReader) directly rather than forcing them into a query(sql) shape. Identity: Property | Value | --------- | ---------------- | name | 'pulsar' | library | 'pulsar-client' | The store is built on the pulsar-client driver: it constructs a Pulsar.Client({ serviceUrl }), then creates producers, consumers, and readers from that client. Message serialization Pulsar sends and receives raw bytes. produce() accepts a string or a Buffer: strings are UTF-8 encoded to a Buffer, Buffers are passed through untouched, and the result is handed to the producer as { data: Buffer }. Consumers receive Pulsar Message objects; use msg.getData() to read the raw bytes back. Lazy loading — not a hard dependency pulsar-client is not a hard dependency of this package. The driver is loaded lazily via require('pulsar-client') 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 Pulsar store is actually connected against a real broker. Injected client PulsarStoreOptions accepts a pre-built client (a pulsar-client Client instance, or a compatible mock). When provided, connect() uses it directly and does not require('pulsar-client'). This is how the test suite injects a mock client (no driver, no network), and how callers can supply a custom-configured Client. Connection Build a store from connection options and call connect(): PulsarStoreOptions: Option | Type | Purpose | ------------ | -------- | -------------------------------------------------------------------------------- | serviceUrl | string | Pulsar broker service URL, e.g. 'pulsar://localhost:6650'. | client | any | A pre-built pulsar-client Client. When set, the driver is not required. | [key] | unknown | Any additional pulsar-client Client options, passed straight through. | connect() is idempotent: if already connected with a live client, it returns immediately. Connection failures are wrapped in a ConnectionError (an ECONNREFUSED message is normalized to 'Connection refused'). Injected-client form Supply your own client (or a mock) to bypass driver-based client construction: Methods Using any operation before connect() (or after disconnect()) throws a ConnectionError via the internal requireClient() guard. Operation failures are wrapped in a DatabaseError (preserving the original error). Lifecycle Method | Signature | Behavior | ------------- | ----------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | connect | connect(): Promise<void> | Uses an injected client if provided, otherwise lazy-requires pulsar-client and builds a Pulsar.Client({ serviceUrl }). Idempotent. Wraps failures in ConnectionError. | disconnect | disconnect(): Promise<void> | Stops receive loops, then best-effort closes every tracked consumer, reader, and cached producer, then closes the client. Clears all state. | isConnected | isConnected(): boolean | true only when connected and a client is present. | getClient | getClient(): any | Returns the underlying pulsar-client Client for operations not wrapped here. Throws ConnectionError if not connected. | Producing Method | Signature | Behavior | --------- | ------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------- | produce | produce(topic: string, message: PulsarMessage): Promise<any> | Creates the producer for topic on first use and caches one producer per topic, so repeated calls reuse it. Serializes message (string \| Buffer) to bytes and sends { data: Buffer }. Returns the producer's send result. | Consuming Method | Signature | Behavior | -------------- | ------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | subscribe | subscribe(topic: string, subscription: string, handler: (msg: any) => Promise<void> \| void, options?: SubscribeOptions): Promise<any> | Subscribes a consumer to topic under subscription, then runs a background receive loop: consumer.receive() → handler(msg) → consumer.acknowledge(msg). Handler/ack failures don't kill the loop (move to next message). The loop stops on disconnect(). The consumer is tracked so disconnect() closes it. Returns the created consumer. | createReader | createReader(topic: string, options?: ReaderOptions): Promise<any> | Creates a Pulsar reader over topic for non-destructive, position-controlled reads (no subscription/acks). Tracked so disconnect() closes it. Returns the created reader. | SubscribeOptions and ReaderOptions are spread into the client call ({ topic, subscription, ...options } / { topic, ...options }), so any extra pulsar-client field passes straight through. Example Verification status Unit / mock-verified only. The tests in tests/nosql/pulsar.test.ts are fully mock-driven: a fake Client whose createProducer/subscribe/createReader/close are jest spies (with spy-backed producer, consumer, and reader sub-objects) is injected via PulsarStoreOptions.client. The real pulsar-client package is not installed, and there is no live broker and no network in the test run — connect() never reaches the require. What this proves: produce() creates a producer for the topic, sends a Buffer (string encoded to UTF-8, Buffer passed through untouched), and caches one producer per topic across repeated calls. subscribe() wires the receive → handler → acknowledge loop, delivering each queued message to the handler and acking the same message, and passes subscribe options through. createReader() creates and tracks a reader. Lifecycle: connecting via an injected client without loading the driver, disconnect() closing the client and tracked consumers/readers, and ConnectionError from getClient() / operations before connect(). Error handling: produce/subscribe failures wrapped so the message matches /produce.failed/i / /subscribe.failed/i. What this does not prove: live execution against a real Apache Pulsar broker. The call shapes are verified against the pulsar-client API contract, but end-to-end streaming 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