Class: NatsQueueDriver
The NATS JetStream driver of the queues: each queue is a stream (its messages kept for maxAge), each group a durable pull consumer of it created when the group registers (the messages added from then on are kept for it), the results a key-value bucket, the replies the inbox of each instance. A delivery is acknowledged once the server confirms it, delivered again after a delay with a negative acknowledgement, extended while it is processed. The delays are scheduled messages on NATS 2.12 and later, negative acknowledgements until they are due before. It needs the package nats (loaded when it connects) and a NATS server 2.9.4 or later with JetStream. The Queue creates it from NatsQueueOptions; create it directly to give it to another component.
Example
import {NatsQueueDriver} from '@lakutata/queue'
const driver: NatsQueueDriver = new NatsQueueDriver({servers: 'nats://127.0.0.1:4222', prefix: 'billing'})
await driver.connect()
console.log(driver.version, await driver.groups('invoice.send'))
await driver.close()Extends
Constructors
Constructor
new NatsQueueDriver(
options?):NatsQueueDriver
Parameters
options?
Omit<NatsQueueOptions, "type">
The servers, the credentials, the prefix and the settings of the streams.
Returns
NatsQueueDriver
Overrides
Properties
capabilities
readonlycapabilities:QueueCapabilities
Every capability but the priorities and the ordering by key.
Overrides
lease
readonlylease:number
The ackWait of the consumers, in milliseconds: the queue extends the deliveries every third of it.
Overrides
version?
optionalversion?:string
The version of the server, once connected.
Methods
close()
close():
Promise<void>
Stop the pulls and close the connection (the decisions sent before are flushed).
Returns
Promise<void>
Overrides
connect()
connect():
Promise<void>
Load the client package, connect, check the server (its version, JetStream) and subscribe to the replies.
Returns
Promise<void>
Throws
When nats is not installed.
Throws
When the server is older than NATS 2.9.4 or runs without JetStream.
Overrides
consume()
consume(
options,handler):Promise<QueueSubscription>
Register the group, then pull its messages: one pull at a time for each message processed at once.
Parameters
options
The queue, the group and the concurrency.
handler
(delivery) => Promise<void>
Processes a delivery.
Returns
Promise<QueueSubscription>
The subscription: stop() ends the pulls (the messages being processed go on; a message pulled after it is given back).
Overrides
groups()
groups(
name):Promise<string[]>
The groups registered for a queue: the consumers of its stream.
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 or expired.
Overrides
register()
register(
name,group):Promise<void>
Register a group of a queue: its durable consumer, created when missing, keeps the messages added from then on. The ackWait of a consumer is the one it was created with.
Parameters
name
string
The queue.
group
string
The group.
Returns
Promise<void>
Overrides
reply()
reply(
address,reply):Promise<void>
Publish a reply to the inbox of the instance waiting for it.
Parameters
address
string
The inbox subject.
reply
The reply.
Returns
Promise<void>
Overrides
replyAddress()
replyAddress():
string
The inbox of the replies to this instance.
Returns
string
The inbox subject.
Overrides
saveResult()
saveResult(
key,value,ttl):Promise<void>
Keep a result in the bucket, with the time it expires (the bucket keeps its values for the longest time asked).
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>
Publish a message to the stream of its queue: scheduled when it is delayed (NATS 2.12), else delivered and given back by the consumers until it is due.
Parameters
message
The message.
Returns
Promise<boolean>
False when the server dropped it as a duplicate.