Class: RabbitMQQueueDriver
The RabbitMQ driver of the queues: each group of a queue is a durable queue bound to a direct exchange by the name of the queue (every group) and by the name of the queue and the group (that group only); a delayed message goes through a delay queue of its delay, expiring it back to the exchange. The messages are published with confirmations; a delivery is acknowledged, or given back to its queue when it is abandoned. The retries are copies sent by the Queue through the delay queues, the attempts counted in their headers. The replies of the requests are the direct replies of RabbitMQ. It keeps no result and cannot list the groups (the queue keeps them in the store of its results). The client reconnects by itself and consumes again. It needs the package amqplib (loaded when it connects) and RabbitMQ 3.6 or later. The Queue creates it from RabbitMQQueueOptions; create it directly to give it to another component.
Example
import {RabbitMQQueueDriver} from '@lakutata/queue'
const driver: RabbitMQQueueDriver = new RabbitMQQueueDriver({url: 'amqp://127.0.0.1:5672', prefix: 'billing'})
await driver.connect()
console.log(driver.version)
await driver.close()Extends
Constructors
Constructor
new RabbitMQQueueDriver(
options?):RabbitMQQueueDriver
Parameters
options?
Omit<RabbitMQQueueOptions, "type">
The server, the credentials and the prefix.
Returns
RabbitMQQueueDriver
Overrides
Properties
capabilities
readonlycapabilities:QueueCapabilities
The delays, the priorities and the replies: the queue retries with copies, keeps the results and the groups in a store.
Overrides
lease?
readonlyoptionallease?:number
The lease of a delivery, in milliseconds: the time before the server delivers it again when it is not ended. The queue extends the deliveries being processed every third of it. Undefined when the deliveries have no lease.
Inherited from
version?
optionalversion?:string
The version of the server, once connected.
Methods
close()
close():
Promise<void>
Stop the consumers and close the connection: the messages not acknowledged go back to their queues.
Returns
Promise<void>
Overrides
connect()
connect():
Promise<void>
Load the client package, connect (reconnected afterwards by the client), check the version of the server, declare the exchanges and consume the direct replies.
Returns
Promise<void>
Throws
When the server cannot be reached.
Throws
When amqplib is not installed.
Throws
When the server is older than RabbitMQ 3.6.
Overrides
consume()
consume(
options,handler):Promise<QueueSubscription>
Register the group, then consume its queue on a channel of its own (consumed again after each reconnection).
Parameters
options
The queue, the group and the concurrency.
handler
(delivery) => Promise<void>
Processes a delivery.
Returns
Promise<QueueSubscription>
The subscription: stop() cancels the consumer (the messages being processed go on).
Overrides
groups()
groups():
Promise<string[] |undefined>
RabbitMQ cannot list the queues bound to an exchange: the queue keeps the groups in the store of its results.
Returns
Promise<string[] | undefined>
Undefined.
Overrides
onReply()
onReply(
handler):void
Receive the replies to this instance.
Parameters
handler
(reply) => void
Receives a reply.
Returns
void
Overrides
readResult()?
optionalreadResult(key):Promise<Uint8Array<ArrayBufferLike> |undefined>
Read a value kept by QueueDriver.saveResult (capability results).
Parameters
key
string
The key.
Returns
Promise<Uint8Array<ArrayBufferLike> | undefined>
The value, or undefined when it is not kept.
Inherited from
register()
register(
name,group):Promise<void>
Register a group of a queue: its durable queue declared and bound, keeping the messages added from then on.
Parameters
name
string
The queue.
group
string
The group.
Returns
Promise<void>
Overrides
reply()
reply(
address,reply):Promise<void>
Publish a reply to the address of its request.
Parameters
address
string
The address of the request.
reply
The reply.
Returns
Promise<void>
Overrides
replyAddress()
replyAddress():
string
The direct replies of RabbitMQ.
Returns
string
'amq.rabbitmq.reply-to' (the server gives each request its address).
Overrides
saveResult()?
optionalsaveResult(key,value,ttl):Promise<void>
Keep a value (capability results).
Parameters
key
string
The key.
value
Uint8Array
The value.
ttl
number
How long it is kept, in milliseconds.
Returns
Promise<void>
Resolved once kept.
Inherited from
send()
send(
message):Promise<boolean>
Publish a message to the exchange, routed to every group of its queue (to its group only, when it has one); to the delay queue of its delay first, when it is delayed. Resolved once the server confirms it.
Parameters
message
The message.
Returns
Promise<boolean>
Always true (RabbitMQ does not drop the duplicates).