import { EventEmitter } from "events";
export interface Job<T> {
id: string;
payload: T;
attempts: number;
}
export class JobQueue<T> extends EventEmitter {
private items: Job<T>[] = [];
private seq = 0;
enqueue(payload: T, attempts = 0): Job<T> {
const job: Job<T> = { id: `job_${++this.seq}`, payload, attempts };
this.items.push(job);
this.emit("enqueued");
return job;
}
dequeue(): Job<T> | undefined {
return this.items.shift();
}
get size(): number {
return this.items.length;
}
waitForJob(signal: () => boolean): Promise<void> {
if (this.size > 0 || !signal()) return Promise.resolve();
return new Promise((resolve) => {
const onEvent = () => {
this.off("enqueued", onEvent);
this.off("wake", onEvent);
resolve();
};
this.once("enqueued", onEvent);
this.once("wake", onEvent);
});
}
wake(): void {
this.emit("wake");
}
}
import { Job, JobQueue } from "./queue";
export type Handler<T> = (payload: T, job: Job<T>) => Promise<void>;
export interface WorkerOptions<T> {
concurrency: number;
maxRetries: number;
handler: Handler<T>;
onFailed?: (job: Job<T>, err: unknown) => void;
}
export class Worker<T> {
private running = false;
constructor(
private readonly queue: JobQueue<T>,
private readonly opts: WorkerOptions<T>
) {}
async start(): Promise<void> {
this.running = true;
const loops = Array.from({ length: this.opts.concurrency }, () =>
this.runLoop()
);
await Promise.all(loops);
}
stop(): void {
this.running = false;
this.queue.wake(); // unpark any idle loops
}
private async runLoop(): Promise<void> {
while (this.running) {
await this.queue.waitForJob(() => this.running);
const job = this.queue.dequeue();
if (!job) continue;
try {
await this.opts.handler(job.payload, job);
} catch (err) {
if (job.attempts < this.opts.maxRetries) {
this.queue.enqueue(job.payload, job.attempts + 1);
} else {
this.opts.onFailed?.(job, err);
}
}
}
}
}
import { JobQueue } from "./queue";
import { Worker } from "./worker";
interface EmailJob {
to: string;
subject: string;
}
async function main(): Promise<void> {
const queue = new JobQueue<EmailJob>();
const worker = new Worker<EmailJob>(queue, {
concurrency: 4,
maxRetries: 3,
handler: async (payload, job) => {
if (payload.to.includes("invalid")) throw new Error("bad address");
await new Promise((r) => setTimeout(r, 50));
console.log(`[${job.id}] sent "${payload.subject}" to ${payload.to}`);
},
onFailed: (job, err) => {
console.error(`dropping ${job.id} after retries:`, err);
},
});
for (let i = 0; i < 20; i++) {
queue.enqueue({ to: `user${i}@example.com`, subject: `Digest ${i}` });
}
queue.enqueue({ to: "invalid@example.com", subject: "Broken" });
const run = worker.start();
setTimeout(() => worker.stop(), 2000);
await run;
}
main().catch((e) => {
console.error(e);
process.exit(1);
});
This snippet builds a small, fully typed job queue with a worker pool that drains it while respecting a concurrency limit. The generic type parameter is the key idea: instead of a queue that stores any payloads, JobQueue<T> in queue.ts carries the payload shape through every method, so enqueue, the internal Job<T> record, and the handler in worker.ts all agree at compile time. This catches an entire class of bugs where a producer pushes one shape and a consumer expects another.
In queue.ts the queue is a plain FIFO backed by an array, but each entry is wrapped in a Job<T> that tracks id, attempts, and payload. The queue extends EventEmitter so the worker can react to enqueued events without polling; waitForJob returns a promise that resolves either immediately if work is pending or on the next enqueued signal. That push model is what lets an idle worker sleep at zero CPU cost instead of busy-looping.
The real work happens in Worker in worker.ts. It spins up concurrency independent runLoop coroutines that each pull one job, run the user handler, and loop again. Because there are exactly N loops, at most N jobs run at once — the array of loops is the concurrency limit, so no semaphore or counter is needed. Promise.all over the loops means start resolves only when every loop has exited.
Retries are handled by re-enqueuing: on a thrown handler, attempts is compared against maxRetries, and the job is either pushed back with an incremented count or routed to onFailed. This keeps failures durable within the run rather than swallowing them.
The stop method flips a running flag and wakes any parked loops so they can observe the shutdown and drain cleanly instead of hanging on waitForJob.
main.ts wires it together: a typed EmailJob payload, an async handler, and four concurrent workers. The trade-off to note is that this queue is in-memory and single-process — restarts lose jobs — so it fits background work within one Node instance, not cross-service durability, where a real broker like Redis or SQS belongs.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
import axios, { AxiosError } from 'axios'
import { v4 as uuidv4 } from 'uuid'
const api = axios.create({
baseURL: import.meta.env.VITE_API_URL || 'http://localhost:3000/api/v1',
timeout: 15000,Axios API client with interceptors
Share this code
Here's the card — post it anywhere.
