RabbitMQStore

RabbitMQStore Reference documentation for RabbitMQStore, the ORM's RabbitMQ / AMQP 0-9-1 store. Overview RabbitMQ is a message broker built on AMQP 0-9-1, not a database. There is no query language, no relational schema, and no query(sql) shape. Producers publish messages to exchanges, which route them (by binding key/pattern) into queues, from which consumers read. Because of this, RabbitMQStore does not implement the SQL-shaped Dialect interface. Instead it implements the minimal NoSqlStore marker interface (src/nosql/store.ts), which requires only connection lifecycle (connect/disconnect/isConnected) plus a getClient() escape hatch. RabbitMQ's real operations — assert/bind topology, publish, and consume — are exposed directly on the class, mapped onto a single AMQP channel. Key facts (from src/nosql/rabbitmq/index.ts): name = 'rabbitmq' library = 'amqplib' Uses the official amqplib driver. amqplib is an optional peer of this ORM. It is lazy-loaded via require('amqplib') inside connect(), so importing this module does not require amqplib to be installed unless you actually connect over a URL. An injected connection can be supplied via RabbitMQStoreOptions.connection. When provided, connect() adopts it and skips the require entirely — useful for sharing one AMQP connection across stores, or for injecting a mock in tests. The store holds both a connection and a single channel. The connection comes from amqp.connect(...) (or is injected); the channel is created with connection.createChannel(). All topology/produce/consume operations run on that one channel. Message serialization AMQP message bodies are raw bytes (Buffer). For convenience, both sendToQueue() and publish() accept any value and normalize it via an internal toBuffer(): A Buffer is passed through as-is (same object). A Uint8Array is wrapped with Buffer.from(...). A string is UTF-8 encoded with Buffer.from(...). Anything else is serialized with JSON.stringify, then wrapped in a Buffer. Consumers receive the raw amqplib message object; you decode msg.content (a Buffer) yourself, e.g. JSON.parse(msg.content.toString()). Connection Real connection (via URL) url defaults to 'amqp://localhost' when omitted. amqps://host (TLS) URLs are also supported. connect() is idempotent: if already connected with a live channel, it returns immediately without creating another channel. On any failure during connect, the channel/connection are reset to null and a ConnectionError is thrown. ECONNREFUSED maps to the message 'Connection refused'; other failures are wrapped as Unable to connect to RabbitMQ: <message>. Injected connection When connection is provided, connect() does not require('amqplib') or open a new connection; it just calls createChannel() on the supplied connection. Options Note: amqplib's connection and channel are typed as any. The repo's tsconfig sets noImplicitAny: false, the intended escape hatch for optional, untyped drivers. Methods Constructor Both options and its fields are optional; defaults to {} (URL 'amqp://localhost', no injected connection). No connection is opened until connect() is called. Lifecycle Opens the connection (or adopts the injected one) and creates a single channel. Idempotent when already connected. Throws ConnectionError on failure. Closes the channel, then the connection, then resets internal state (channel/connection to null, connected to false). Errors from channel.close() and connection.close() are swallowed — teardown proceeds regardless. Returns true only when connected and a channel is present. Returns the underlying amqplib channel for operations not wrapped here. Throws ConnectionError('Not connected to RabbitMQ') if there is no channel. Returns the underlying amqplib connection. Throws ConnectionError('Not connected to RabbitMQ') if there is no connection. Topology (queues, exchanges, bindings) Declares a queue, creating it if it doesn't exist (idempotent). Returns amqplib's assertion reply ({ queue, messageCount, consumerCount }). Pass '' as queue to have the broker generate a unique name. options is passed straight to amqplib (e.g. { durable: true }). Declares an exchange of the given type ('direct', 'fanout', 'topic', 'headers'), creating it if it doesn't exist (idempotent). options (e.g. { durable: false }) is passed through. Binds queue to exchange so messages published to exchange matching pattern (the routing/binding key) are routed into queue. Producing Sends a message straight to a named queue (via the default exchange). The message is normalized through toBuffer() (see Message serialization). Returns amqplib's boolean write result — false when the channel's write buffer is full and you should wait for a 'drain' event before writing more. options (e.g. { persistent: true }) is passed through. Publishes a message to an exchange with a routingKey; the exchange routes it to bound queues. Message serialization matches sendToQueue(). Returns amqplib's boolean write result. options (e.g. { contentType: 'application/json' }) is passed through. The message value type is: Consuming Registers handler to receive each message delivered from queue. The handler is called with the raw amqplib message object (or null if the consumer is cancelled by the broker); decode msg.content yourself. Unless options.noAck is set, call ack(msg) / nack(msg) when done. Returns amqplib's { consumerTag } reply. Acknowledges a delivered message so the broker can drop it. Synchronous (delegates directly to the channel; not a promise). Rejects a delivered message. By default RabbitMQ requeues it (requeue defaults to true in amqplib); pass requeue = false to drop or dead-letter it. Synchronous. Sets the consumer prefetch (QoS) — the max number of unacknowledged messages the broker will deliver to consumers on this channel at once. Error handling Connection failures during connect() (and the pre-connect guard in getClient()/getConnection()) throw ConnectionError. All channel-backed operations (assertQueue, assertExchange, bindQueue, sendToQueue, publish, consume, prefetch) run through an internal exec() wrapper that catches driver errors and rethrows them as DatabaseError with a message of the form RabbitMQ <action> failed: <message>. Example Exchange-based routing: Verification status Unit / mock-verified only. The tests in tests/nosql/rabbitmq.test.ts are pure unit tests: The real amqplib driver is not installed and is never loaded. A mock amqplib connection — whose createChannel() returns a fully-spied channel — is injected via the store's connection option, so connect() adopts it and skips require('amqplib') entirely. No network is used; no real broker is contacted. The tests confirm delegation and behavior against the mock: name/library values; injected-connection connect and channel creation; getClient()/getConnection() returning the raw objects; idempotent connect; disconnect closing channel then connection; ConnectionError on channel-creation failure and on pre-connect use; topology delegation (assertQueue, assertExchange, bindQueue); serialization (Buffer passthrough, string UTF-8, object JSON) for sendToQueue/publish; consume registering a callback that forwards delivered messages; and ack/nack/prefetch delegating with the expected arguments. Behavior against a real RabbitMQ broker over the network is not covered by these tests. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories