中文
devexamples

Async Cargo Adaptive Batch Processing

Buffer a stream of items into adaptive batches with async cargo, flushing on size or timing thresholds so slow items never block the queue.

JavascriptResilience

Code

snippet
import cargo from 'async/cargo';

const flush = async (items) => {
  console.log('flushing', items.length, 'events');
  await new Promise((r) => setTimeout(r, 30));
};

const queue = cargo(flush, 3);

queue.push('event-1');
queue.push('event-2');
queue.push('event-3');
queue.push('event-4');

queue.push(['event-5', 'event-6'], (err) => {
  console.log('callback after group drained, err:', err ?? null);
});

const drained = await queue.drain();
console.log('all batches processed:', drained === undefined);

queue.push('event-7');
await queue.idle();

Related Examples