NsqStore — NSQ topic/channel messaging

NsqStore — NSQ topic/channel messaging Overview NSQ is a realtime distributed messaging platform, not a database. Producers publish messages to a topic via a Writer; consumers create a Reader bound to a topic + channel and handle each delivered message (calling msg.finish()/msg.requeue()). NSQ distributes a topic's messages across all channels and load-balances within a channel. There is no query language and no SQL-shaped access pattern. Because none of that fits the SQL-shaped Dialect interface, NsqStore implements the minimal NoSqlStore marker interface (src/nosql/store.ts) — connection lifecycle plus a getClient() escape hatch — and exposes NSQ's real publish/subscribe operations (enqueue/process/stats) directly rather than forcing them into a query(sql) shape. Identity: Property | Value | --------- | --------- | name | 'nsq' | library | 'nsqjs' | Reader-scoped caching The store wraps a single producer Writer plus a Map of consumer Readers keyed by topic/channel (cached on process()). A repeat process() on the same topic + channel returns the existing Reader. All readers are closed and cleared on disconnect(). Lazy loading — not a hard dependency nsqjs is not a hard dependency of this package. It is an optional peer, loaded lazily via require('nsqjs') 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 an NSQ store is actually connected. Injected client NsqStoreOptions accepts a pre-built client — the nsqjs module itself, or any stand-in exposing Writer and Reader constructors. When provided, connect() adopts it directly and does not require('nsqjs'). This is how the test suite injects mocks (no driver, no network). Connection Build a store from connection options and call connect(): connect() resolves the driver (adopting an injected client or lazy-requireing nsqjs), constructs new driver.Writer(host, port), and — only when the writer exposes both connect and on — waits for its 'ready' event (rejecting on 'error'). Failures are wrapped in a ConnectionError (a message containing ECONNREFUSED is normalized to 'Connection refused'). Option | Type | Purpose | --------------- | -------- | ------------------------------------------------------------------------------------------------------- | nsqdHost | string | nsqd host for the producer Writer. Defaults to '127.0.0.1'. | nsqdPort | number | nsqd TCP port for the producer Writer. Defaults to 4150. | readerOptions | any | Reader connection options (typically { lookupdHTTPAddresses } or { nsqdTCPAddresses }) passed to every Reader. | client | any | A pre-built nsqjs module (or stand-in with Writer/Reader). When set, the driver is not required. | Injected-client form Methods Driver failures are wrapped in a DatabaseError (message NSQ <action> failed: ..., preserving the original error). Using the store before connect() (or after disconnect()) throws a ConnectionError. Lifecycle Method | Signature | Behavior | ------------- | ----------------------------- | ------------------------------------------------------------------------------------------------------------- | connect | connect(): Promise<void> | Resolves the driver, constructs the producer Writer, and waits for 'ready' if the writer supports it. Idempotent when already connected. Wraps failures in ConnectionError. | disconnect | disconnect(): Promise<void> | close()s every cached Reader and the Writer, clears the reader map, and drops the driver. | isConnected | isConnected(): boolean | true only when connected and a Writer is present. | getClient | getClient(): any | Returns the underlying producer Writer. Throws ConnectionError if not connected. | Queue operations Method | Signature | Behavior | --------- | --------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------- | enqueue | enqueue(queue: string, jobData: unknown): Promise<null> | Publishes a message to the topic queue via writer.publish(topic, message, cb). Payload is JSON-serialized (strings/Buffers pass through). NSQ returns no client-visible job id, so this resolves to null. | process | process(queue: string, handler: (msg: any) => void \| Promise<void>, opts?: NsqProcessOptions): Promise<any> | Subscribes a Reader to topic queue on a channel (default 'default'), wires reader.on('message', ...) to handler(msg), and connect()s it. Caches one Reader per topic/channel; a repeat returns it. | getJob | getJob(queue: string, id: string): Promise<never> | Not supported — NSQ has no addressable job ids. Always throws DatabaseError. | remove | remove(queue: string, id: string): Promise<never> | Not supported — NSQ messages are acknowledged via msg.finish(), not removed by id. Always throws DatabaseError. | stats | stats(queue: string): Promise<{ topic: string; readers: number }> | Returns a lightweight local view — the number of active Readers this store has bound to the topic. (NSQ's authoritative stats live in nsqd's HTTP /stats endpoint, not the client library.) | Supporting type: Note: getJob() and remove() are intentionally unsupported because NSQ exposes no server-side lookup of a message by id. Both reject with a DatabaseError explaining the alternative (use msg.finish() in the handler). Example Verification status Unit / mock-verified only. The tests in tests/nosql/nsq.test.ts are fully mock-driven: a fake driver module whose Writer and Reader constructors are Jest spies is injected via NsqStoreOptions.client, so connect() adopts it and skips require('nsqjs'). The mock Writer exposes neither connect nor on, so the store skips the 'ready' handshake. The real nsqjs package is not installed, and there is no live nsqd and no network in the test run. What this proves: name/library identity, Writer construction, and exposure via getClient() without loading the driver. Lifecycle: disconnect() closing the Writer, and ConnectionError when using the store before connecting. enqueue publishing the JSON-serialized message and resolving null. process creating a Reader, wiring the message handler, and connecting; caching one Reader per topic/channel. stats reporting the active reader count for a topic. getJob and remove throwing DatabaseError (unsupported). Error handling: publish failures wrapped in DatabaseError. What this does not prove: live execution against a real nsqd. Constructor/method shapes are 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