Breadline

Breadline is a robust, type-safe asynchronous task queue for Node.js. It helps you manage concurrency, enforce rate limits, and schedule tasks with priorities, all while retaining full control over execution flow.

Features

• Concurrency Control: Limit the number of tasks running in parallel.
• Rate Limiting: Enforce strict execution limits over time windows (e.g., 10 reqs / 1 sec).
• Priority Support: Schedule urgent tasks to run before others.
• Pluggable Queues: Use the default Priority Queue or switch to Binary Max-Heap for performance.
• AbortSignal Support: Cancel queued or running tasks using standard AbortController.
• Event-Driven: Hook into lifecycle events like empty, idle, or rateLimited.
• Zero Dependencies: (Almost) zero — only eventemitter3 for efficient event handling.
• TypeScript: Written in TypeScript with full type definitions.

Installation

npm install breadline-ts

Usage

Basic Usage (Concurrency Control)

Limit concurrent execution to prevent overwhelming resources.

import { Breadline } from "breadline";

// Create a queue allowing 2 concurrent tasks
const queue = new Breadline({ concurrency: 2 });

const task = (id: number) => async () => {
    console.log(`Start ${id}`);
    await new Promise(r => setTimeout(r, 1000));
    console.log(`End ${id}`);
    return id;
};

// Add tasks
queue.add(task(1));
queue.add(task(2));
queue.add(task(3)); // Will wait until 1 or 2 finishes

Rate Limiting

Ensure you don't exceed API rate limits (e.g., 5 requests per second).

const queue = new Breadline({
    interval: 1000, // 1 second window
    intervalCap: 5  // Max 5 tasks per window
});

for (let i = 0; i < 20; i++) {
    queue.add(async () => {
        await fetch("https://api.example.com/data");
    });
}

Prioritization

Process important tasks first, even if they were added later.

const queue = new Breadline({ concurrency: 1 });

queue.add(async () => console.log("Low priority"), { priority: 0 });
queue.add(async () => console.log("High priority"), { priority: 10 });
queue.add(async () => console.log("Medium priority"), { priority: 5 });

// Output:
// High priority
// Medium priority
// Low priority

Custom Queues

You can swap the internal queue implementation for better performance or custom behavior.

import { Breadline, BinaryMaxHeap } from "breadline";

// Use BinaryMaxHeap for potentially faster priority handling in very large queues
const queue = new Breadline({
    queue: BinaryMaxHeap
});

Cancellation (AbortSignal)

Cancel tasks that are waiting in the queue or currently running (if supported by the task).

const controller = new AbortController();
const queue = new Breadline();

queue.add(
    async ({ signal }) => {
        const response = await fetch("https://example.com", { signal });
        return response.json();
    },
    { signal: controller.signal }
).catch(err => console.log("Task aborted:", err));

// Cancel the task
controller.abort();

API Reference

new Breadline(options?)

Creates a new queue instance.

Option	Type	Default	Description
concurrency	number	Infinity	Max concurrent tasks.
interval	number	1	Time window in milliseconds for rate limiting.
intervalCap	number	Infinity	Max tasks allowed per interval.
immediate	boolean	true	If true, tasks start immediately. If false, call start().
queue	class	PriorityQueue	Custom queue implementation (e.g. BinaryMaxHeap).

Methods

• add(task, options?): Adds a task to the queue. Returns a Promise that resolves with the task result.
  • task: ({ signal }) => Promise<T>
  • options: { priority?: number, signal?: AbortSignal, id?: string }
• addMany(tasks, options?): Adds multiple tasks.
• pause(): Pauses processing of new tasks.
• start(): Resumes processing.
• clear(): Removes all queued tasks.
• prioritize(id, priority): Updates the priority of a waiting task.
• onEmpty(): Returns a Promise that resolves when the queue becomes empty.
• onIdle(): Returns a Promise that resolves when the queue is empty AND all running tasks have finished.

Events

The queue emits the following events:

• "add": A task was added.
• "active": A task started executing.
• "done": A task completed successfully.
• "error": A task failed.
• "empty": The queue is empty (but tasks may be running).
• "idle": The queue is empty and no tasks are running.
• "rateLimited": Rate limit has been reached.
• "rateLimitCleared": Rate limit has reset.

License

ISC