Class: KafkaQueueDriver
The Kafka driver of the queues: each queue is a topic, each group of a queue a consumer group, registered with its offsets set at the end of the topic (it receives the messages added from then on). The messages of a partition are processed in their order, one at a time, concurrency partitions at once; a message is committed once acknowledged, retried in place after its delay (its partition waiting), left uncommitted when it is abandoned (fetched again by the next owner of its partition). The client sends the heartbeats while the messages are processed. It keeps no result, has no reply channel (the queue reads the results it keeps), no delays, no priorities and no deduplication. It needs the package @platformatic/kafka (loaded when it connects) and Kafka 3.5 or later. The Queue creates it from KafkaQueueOptions; create it directly to give it to another component.
Example
import {KafkaQueueDriver} from '@lakutata/queue'
const driver: KafkaQueueDriver = new KafkaQueueDriver({brokers: ['127.0.0.1:9092'], prefix: 'billing'})
await driver.connect()
console.log(await driver.groups('invoice.send'))
await driver.close()Extends
Constructors
Constructor
new KafkaQueueDriver(
options?):KafkaQueueDriver
Parameters
options?
Omit<KafkaQueueOptions, "type">
The brokers, the credentials, the prefix and the settings of the topics.
Returns
KafkaQueueDriver
Overrides
Properties
capabilities
readonlycapabilities:QueueCapabilities
The retries in place and the ordering by partition: the queue keeps the results.
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
Methods
close()
close():
Promise<void>
Close the consumers (leaving their groups), the producer and the admin client.
Returns
Promise<void>
Overrides
connect()
connect():
Promise<void>
Load the client package, connect the admin client and the producer, and check that the brokers speak the protocol versions the client needs.
Returns
Promise<void>
Throws
When the brokers cannot be reached.
Throws
When @platformatic/kafka is not installed.
Throws
When the brokers are older than the client supports (Kafka 3.5).
Overrides
consume()
consume(
options,handler):Promise<QueueSubscription>
Register the group, then consume the topic in its consumer group: the messages of a partition one at a time, concurrency partitions at once.
Parameters
options
The queue, the group and the concurrency.
handler
(delivery) => Promise<void>
Processes a delivery.
Returns
Promise<QueueSubscription>
The subscription: stop() processes no message any more (the ones being processed go on, the ones fetched are fetched again by the next owner of their partition).
Overrides
groups()
groups(
name):Promise<string[]>
The groups registered for a queue: the consumer groups of its prefix and name.
Parameters
name
string
The queue.
Returns
Promise<string[]>
Their names.
Overrides
onReply()?
optionalonReply(handler):void
Receive the replies sent to this instance (capability reply).
Parameters
handler
(reply) => void
Called for each reply.
Returns
void
Inherited from
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 consumer group, without offsets yet, gets the end of the topic as its offsets (it receives the messages added from then on).
Parameters
name
string
The queue.
group
string
The group.
Returns
Promise<void>
Overrides
reply()?
optionalreply(address,reply):Promise<void>
Send a reply to an instance (capability reply).
Parameters
address
string
Its address.
reply
The reply.
Returns
Promise<void>
Resolved once sent.
Inherited from
replyAddress()?
optionalreplyAddress():string
The address the replies to this instance are sent to (capability reply).
Returns
string
The address.
Inherited from
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>
Produce a message to the topic of its queue (created when missing), partitioned by its key.
Parameters
message
The message.
Returns
Promise<boolean>
Always true (Kafka does not drop the duplicates).