AsyncQueue is a TypeScript class that provides a simple asynchronous queue implementation. It implements the `Symbol.asyncIterator` interface, allowing it to be used in `for-await-of` loops. This enables one or multiple consumers to process items in the q
npm install @ai-zen/async-queueSymbol.asyncIterator interface, allowing it to be used in for-await-of loops. This enables one or multiple consumers to process items in the queue asynchronously.
npm install @ai-zen/async-queue
`
Usage
$3
`typescript
import AsyncQueue from "@ai-zen/async-queue";
`
$3
`typescript
const queue = new AsyncQueue([1, 2, 3]);
`
$3
`typescript
queue.push(value);
// Or
queue.push(value1, value2, value3, ...);
// Or
queue.push(...values);
`
The push method adds one or more values to the queue, and increases its size.
$3
`typescript
queue.done();
`
The done method marks the queue as finished. This indicates that no more items will be added to the queue.
$3
`typescript
for await (const value of queue) {
// Consume the value
}
`
The for await...of loop can be used to iterate over the items in the queue. This loop is asynchronous and will wait for each value to be available before consuming it.
$3
`typescript
const size = queue.size;
`
The size property returns the current size of the queue.
Examples
$3
You can add data asynchronously while consuming data asynchronously.
`typescript
const queue = new AsyncQueue();
(async () => {
for (const value of [1, 2, 3, 4, 5]) {
await sleep(1); // Simulate asynchronous operations
queue.push(value);
}
queue.done();
})();
for await (const value of queue) {
await sleep(1); // Simulate asynchronous operations
console.log(value);
}
console.log("Done!");
`
Output:
`
1
2
3
4
5
Done!
`
$3
This is useful for limiting the number of concurrent operations, such as network requests.
`typescript
// Assume Task and TaskResult are types defined elsewhere in your code
const queue = new AsyncQueue(tasks);
queue.done();
const results: TaskResult[] = [];
// Start 10 concurrent consumers using the competing-consumers pattern
await Promise.all(
Array.from({ length: 10 }).map(async () => {
for await (const task of queue) {
try {
const result = await download(task); // Perform the task
results.push(result);
} catch (error) {
console.error("Task failed:", error);
}
}
})
);
`
$3
The library provides a very simple backpressure control mechanism that can prevent the queue length from exceeding a threshold. Just call and wait for the backpressure method before push. If the queue reaches the threshold, the backpressure method will wait until the queue length decreases to have space.
`typescript
const queue = new AsyncQueue();
const result: number[] = [];
(async () => {
for (const value of Array.from({ length: 100 }).map((_, i) => i)) {
await queue.backpressure(10); // Wait for the queue to have space, the maximum size is 10
queue.push(value);
}
queue.done();
})();
for await (const value of queue) {
await sleep(1000); // Simulate asynchronous operations
result.push(value);
}
``