Async Queue

An async queue is like a line at a bank. Tasks wait in line and get processed one by one (or a few at a time). This helps you control how many things happen simultaneously and ensures everything gets done in an orderly way.

Basic Usage

The simplest way to use a queue is with the default sequential processing:

import { createAsyncQueue } from 'commandkit/async-queue';

// Create a sequential queue (default)
const queue = createAsyncQueue();

// Add tasks to the queue
const result1 = await queue.add(async () => {
  return await processTask1();
});

const result2 = await queue.add(async () => {
  return await processTask2();
});

Custom Configuration

You can create a queue that processes multiple tasks at the same time:

import { createAsyncQueue } from 'commandkit/async-queue';

// Allow 3 concurrent tasks
const queue = createAsyncQueue({
  concurrency: 3,
  onDrain: () => {
    console.log('All tasks completed');
  },
});

// Tasks will run with up to 3 concurrent executions
const promises = [
  queue.add(async () => await task1()),
  queue.add(async () => await task2()),
  queue.add(async () => await task3()),
  queue.add(async () => await task4()),
];

const results = await Promise.all(promises);

Advanced Usage

Queue Control

You can pause and resume the queue, and check its status:

import { createAsyncQueue } from 'commandkit/async-queue';

const queue = createAsyncQueue({ concurrency: 2 });

// Pause the queue
queue.pause();

// Add tasks (they won't execute until resumed)
const promise1 = queue.add(async () => await task1());
const promise2 = queue.add(async () => await task2());

// Resume the queue
queue.resume();

// Check queue status
console.log(`Running: ${queue.getRunning()}`);
console.log(`Pending: ${queue.getPending()}`);
console.log(`Paused: ${queue.isPaused()}`);

Cancelling Operations

You can cancel queue operations if they take too long or if you need to stop them:

import { createAsyncQueue } from 'commandkit/async-queue';

// Create a timeout that cancels after 5 seconds
const signal = AbortSignal.timeout(5000);

const queue = createAsyncQueue({
  concurrency: 2,
  signal,
});

// Add tasks with individual timeouts
const task1Promise = queue.add(
  async () => await longRunningTask1(),
  AbortSignal.timeout(3000),
);

const task2Promise = queue.add(
  async () => await longRunningTask2(),
  AbortSignal.timeout(3000),
);

try {
  await Promise.all([task1Promise, task2Promise]);
} catch (error) {
  if (error.message.includes('aborted')) {
    console.log('Queue was aborted');
  }
}

Batch Processing

Queues are great for processing large batches of items efficiently:

import { createAsyncQueue } from 'commandkit/async-queue';

const queue = createAsyncQueue({
  concurrency: 5,
  onDrain: () => console.log('Batch processing complete'),
});

// Process a large batch of items
const items = Array.from({ length: 100 }, (_, i) => i);

const processItem = async (item: number) => {
  await new Promise((resolve) => setTimeout(resolve, 100));
  return `Processed item ${item}`;
};

// Add all items to queue
const promises = items.map((item) => queue.add(() => processItem(item)));

// Wait for all to complete
const results = await Promise.all(promises);

Error Handling

Queues handle errors gracefully, so one failed task doesn't stop the others:

import { createAsyncQueue } from 'commandkit/async-queue';

const queue = createAsyncQueue({ concurrency: 2 });

// Handle individual task errors
const results = await Promise.allSettled([
  queue.add(async () => {
    throw new Error('Task failed');
  }),
  queue.add(async () => {
    return 'Task succeeded';
  }),
  queue.add(async () => {
    return 'Another success';
  }),
]);

results.forEach((result, index) => {
  if (result.status === 'fulfilled') {
    console.log(`Task ${index}: ${result.value}`);
  } else {
    console.log(`Task ${index}: ${result.reason.message}`);
  }
});

Default Settings

• Concurrency: 1 (sequential processing)
• Storage: In-memory
• Abort Support: Yes

Common Use Cases

• API Rate Limiting: Control how many API calls are made at once
• File Processing Pipelines: Process files one by one or in small batches
• Database Operation Batching: Execute database operations in controlled batches
• Image Processing Queues: Process images without overwhelming the system
• Email Sending Queues: Send emails in an orderly way
• Background Job Processing: Handle background tasks efficiently

Tips for Beginners

1. Start with Sequential: Use the default concurrency of 1 for simple cases
2. Set Reasonable Limits: Don't set concurrency too high (might overwhelm resources) or too low (might be too slow)
3. Use onDrain Callback: Get notified when all tasks are complete
4. Handle Errors: Use Promise.allSettled() to handle individual task failures
5. Monitor Queue State: Check running and pending counts for debugging
6. Use Timeouts: Set timeouts to prevent tasks from hanging forever
7. Think About Your Resources: Set concurrency based on what your system can handle