Ack Range
Acknowledge a contiguous range of KubeMQ queue messages by sequence in Node.js to confirm many deliveries in a single call.
Overview
A single streaming delivery often bundles several messages together, but "successfully processed" rarely applies to all of them uniformly — one handler might fail while its siblings succeed. Acknowledging one message at a time works, but costs a round-trip per message; acknowledging the whole batch blindly risks confirming work that didn't actually finish. Range acknowledgment splits the difference: settle exactly the messages you know succeeded, in one call.
handle.ackRange(sequences) takes an array of broker-assigned sequence numbers — read off each message in the onMessages() callback — and acknowledges all of them together in a single server round-trip, instead of calling an individual ack for each one. Any sequence you leave out of the array stays unsettled and is redelivered once the session's visibility window expires.
Gotchas: the sequences you pass to ackRange must be numbers your handler actually received in that delivery — passing an unknown or already-settled sequence is a wasted (or erroring) call, not a no-op you can rely on. Messages you never include in any ackRange call aren't implicitly skipped forever; they come back for redelivery once the timeout elapses. And handle.close() ends the streaming session — call it only after you've settled everything you intend to, since messages still in flight when the handle closes are left for the next consumer.
Prerequisites
- KubeMQ server running on
localhost:50000 - Node.js SDK installed (
npm install kubemq-js)
Code
import { KubeMQClient, createQueueMessage } from 'kubemq-js';
async function main() {
const client = await KubeMQClient.create({
address: 'localhost:50000',
clientId: 'js-queues-stream-ack-range-client',
});
try {
for (let i = 0; i < 3; i++) {
await client.sendQueueMessage(
createQueueMessage({ channel: 'js-queues-stream.ack-range', body: `msg-${i}` }),
);
}
const handle = client.streamQueueMessages({
channel: 'js-queues-stream.ack-range',
maxMessages: 3,
});
handle.onMessages((msgs) => {
const sequences = msgs.map((m) => m.sequence);
console.log('Received sequences:', sequences);
handle.ackRange(sequences);
console.log('Acknowledged range:', sequences);
handle.close();
});
handle.onError((err) => {
console.error('Error:', err.message);
});
await new Promise((r) => setTimeout(r, 2000));
} finally {
await client.close();
}
}
main().catch(console.error);
How It Works
streamQueueMessages()opens a streaming downstream session;onMessages()fires once the batch of up tomaxMessagesis delivered.handle.ackRange(sequences)acknowledges all messages whosesequencenumbers appear in the array in a single server round-trip, more efficient than per-message acks.- The sequence numbers come from
m.sequenceon each receivedQueueStreamMessage— the server assigns these monotonically per queue. handle.close()is called inside theonMessagescallback to end the session after all messages are acknowledged; the outersetTimeoutprovides a safety timeout in case the callback never fires.
Related
Was this page helpful?