Class: RedisQueueDriver
The Redis driver of the queues, on BullMQ: each queue and group is a BullMQ queue (its jobs removed once processed), the groups of a queue are a Redis set (a message is added for each), the results are keys expiring with their time to keep, the replies are published on a channel of each instance. A job is delivered again after a delay with moveToDelayed, its lock renewed by BullMQ while it is processed. It needs the packages bullmq and ioredis (loaded when it connects) and a Redis server 5.0 or later. The Queue creates it from RedisQueueOptions; create it directly to give it to another component.
Example
import {RedisQueueDriver} from '@lakutata/queue'
const driver: RedisQueueDriver = new RedisQueueDriver({url: 'redis://127.0.0.1:6379', prefix: 'billing'})
await driver.connect()
console.log(await driver.groups('invoice.send'))
await driver.close()Extends
Constructors
Constructor
new RedisQueueDriver(
options?):RedisQueueDriver
Parameters
options?
Omit<RedisQueueOptions, "type">
The server and the prefix of the keys.
Returns
RedisQueueDriver
Overrides
Properties
capabilities
readonlycapabilities:QueueCapabilities
Every capability but the ordering by key.
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>
Close the workers (the jobs being processed complete first), the BullMQ queues and the connections.
Returns
Promise<void>
Overrides
connect()
connect():
Promise<void>
Load the client packages, connect, check the version of the server and subscribe to the replies.
Returns
Promise<void>
Throws
When the server cannot be reached.
Throws
When bullmq or ioredis is not installed.
Throws
When the server is older than Redis 5.0.
Overrides
consume()
consume(
options,handler):Promise<QueueSubscription>
Register the group of the queue, then process its BullMQ queue with a worker.
Parameters
options
The queue, the group and the concurrency.
handler
(delivery) => Promise<void>
Processes a delivery.
Returns
Promise<QueueSubscription>
The subscription: stop() takes no job any more, its worker closed once the jobs being processed end.
Overrides
groups()
groups(
name):Promise<string[]>
The groups which consumed a queue.
Parameters
name
string
The queue.
Returns
Promise<string[]>
Their names.
Overrides
onReply()
onReply(
handler):void
Receive the replies to this instance.
Parameters
handler
(reply) => void
Receives a reply.
Returns
void
Overrides
readResult()
readResult(
key):Promise<Uint8Array<ArrayBufferLike> |undefined>
Read a result kept.
Parameters
key
string
The key of the result.
Returns
Promise<Uint8Array<ArrayBufferLike> | undefined>
The encoded result, undefined when it is not kept.
Overrides
register()
register(
name,group):Promise<void>
Register a group of a queue: a message is added for it 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 on the channel of the instance waiting for it.
Parameters
address
string
The channel.
reply
The reply.
Returns
Promise<void>
Overrides
replyAddress()
replyAddress():
string
The channel of the replies to this instance.
Returns
string
The channel.
Overrides
saveResult()
saveResult(
key,value,ttl):Promise<void>
Keep a result, expiring after its time to keep.
Parameters
key
string
The key of the result.
value
Uint8Array
The encoded result.
ttl
number
The time to keep it, in milliseconds.
Returns
Promise<void>
Overrides
send()
send(
message):Promise<boolean>
Add the message to the BullMQ queue of each group of its queue (of its group only, when it has one); none when no group consumed the queue yet.
Parameters
message
The message.
Returns
Promise<boolean>
Always true (BullMQ drops the duplicates without telling).