Async Batch Processor
High-volume event streams (analytics, logs, telemetry) are usually best forwarded in batches: batching lowers per-call overhead, plays nicely with bulk endpoints, and lets you compress traffic. The trick is choosing when to flush. This snippet builds a processor that flushes whenever the buffer hits a max size OR a max wait elapses, then layers in flush-on-demand and graceful shutdown so nothing is lost during process exit.
1,193 views
6
function makeBatcherV1({ maxSize = 100, maxWaitMs = 250, flush }) {
let buffer = [];
let timer = null;
function schedule() {
if (timer || buffer.length === 0) return;
timer = setTimeout(drain, maxWaitMs);
}
async function drain() {
if (timer) clearTimeout(timer);
timer = null;
if (buffer.length === 0) return;
const items = buffer;
buffer = [];
await flush(items);
}
function add(item) {
buffer.push(item);
if (buffer.length >= maxSize) drain();
else schedule();
}
return { add };
}
const batches = [];
const b1 = makeBatcherV1({ maxSize: 3, maxWaitMs: 30, flush: async (items) => batches.push(items) });
b1.add(1); b1.add(2); b1.add(3); // size flush
b1.add(4);
setTimeout(() => console.log(batches), 50); // wait flushA batcher needs two flush triggers: a size cap that handles bursty traffic and a wall-clock cap that keeps stragglers from sitting forever. The size branch flushes immediately; the wait branch arms a setTimeout whose callback drains the buffer. Replacing buffer = [] with a fresh array before awaiting flush is the critical detail: items added during the in-flight flush land in the new buffer instead of mixing with the in-flight batch. Without that swap, you get duplicate sends or lost items if flush rejects.
2 more snippets in this entry are available for premium members.
Upgrade to Premium