Blocking Promise-driven queues with ECMAScript's async/await and generators

Question

I implemented an AsyncQueue, whose dequeue() operation returns a Promise. The AsyncLimitedQueue additionally enforces a limit on the number of entries, so that queue() returns a Promise as well.

• Full code + unit tests: https://github.com/ComFreek/async-playground
• Generated documentation: https://comfreek.github.io/async-playground

Both classes use Semaphore, which I also built using Promises. For the sake of brevity, I would like to restrain this question to the queue implementations.

Usage example

Excerpt from https://github.com/ComFreek/async-playground/blob/master/examples/queue-stdio-lines.ts:

async function* readInput() {
 // null signals the end of input
 const queue: IAsyncQueue<string|null> = new AsyncQueue();
 // readline is an NPM package simplifying reading lines from stdio 
 const rl = readline.createInterface({ /* ... */ });
 rl.on('line', (line: string) => queue.queue(line));
 rl.on('close', () => queue.queue(null));
 yield* queue;
}
for await (const line of readInput()) {
 if (line === null) {
 break;
 }
 console.log(line);
}

Questions

• Have I followed best practices?

• Are the method contracts clean? Especially, have I employed asynchronous operations and promises where appropriate?

• Having both interfaces IAsyncQueue and IAsyncLimitedQueue, I intentionally did not create a common superinterface. Was this a right decision?

  The reason is that IAsyncQueue#queue and similar operations are synchronous and therefore non-blocking contrary to IAsyncLimitedQueue#queue, which is asynchronous.
  This difference is also manifested in their types, void vs Promise<void>. Making AsyncQueue#queue returning a Promise as well would violate the principle of least astonishment, namely that it is synchronous despite its return type.

Code

Online: repo, docs

AsyncQueue.ts

import { ISemaphore, Semaphore } from '../semaphore/index';
/**
 * Asynchronous FIFO queue with a Promise-driven dequeue operation.
 *
 * All element values are allowed, especially falsy ones, e.g.
 * false, 0, undefined, null, [], {} are all valid elements which
 * can be queued and dequeued.
 *
 * The {@link AsyncIterable} interface iterates the queue's (future) contents
 * ad infinitum. Users are advised to signal the end by manual insertion of a
 * special value (a so-called poison pill):
 *
 * ```
 * const queue = new AsyncQueue<string|null>();
 * file.on('data', (data) => queue.queue(data));
 * file.on('close', () => queue.queue(null));
 *
 * for await (const data of queue) {
 * if (data === null) {
 * break;
 * }
 * // Otherwise, process data
 * }
 * ```
 */
export interface IAsyncQueue<T> extends AsyncIterable<T> {
 /**
 * Queue an element immediately.
 */
 queue(data: T): void;
 /**
 * Queue all elements of an iterable, e.g. an array or a generator function.
 *
 * @example `queue.queueAll(['myArray', 'of', 'strings'])`
 *
 * @example If one has a generator function f:
 * `function *f(): Iterable<string> { ... }`
 * then you can call `queue.queueAll(f())`.
 */
 queueAll(iterable: Iterable<T>): void;
 /**
 * Queue all elements of an asynchronous iterable, e.g. an asynchronous
 * generator functions.
 *
 * @example Using an asynchronous generator function:
 * ```
 * async function *f(): AsyncIterable<string> {
 * yield* ['Array', 'of', 'strings'];
 * }
 *
 * const previousSize = queue.size();
 * queue.queueAllAsync(f());
 * // ^ We do not await the queueing!
 * // Therefore: queue.size() === previousSize here!
 * // This is indeed guaranteed by JS' execution model. There is
 * // no way queueAllAsync could have queried an element from f()
 * // asynchronously using a promise before this code gives up
 * // the "CPU power" by await or yield.
 *
 * await queue.dequeue(); // 'Array'
 * await queue.dequeue(); // 'of'
 * await queue.dequeue(); // 'strings'
 *
 * // queue.size() === 0 and queue.dequeue() would block
 * // ad infinitum
 *
 * await queue.queueAllAsync(f());
 * // We now await the queueing!
 * // Therefore: queue.size() === 3 here!
 * ```
 *
 * @example AsyncQueue instances are also asynchronous iterables,
 * meaning that you can stack multiple queues together:
 * ```
 * const backgroundQueue: IAsyncQueue<string> = new AsyncQueue();
 * const foregroundQueue: IAsyncQueue<string> = new AsyncQueue();
 *
 * setTimeout(() => backgroundQueue.queue('Hello World!'), 100);
 *
 * foregroundQueue.queueAllAsync(backgroundQueue);
 * const retrievedString = await foregroundQueue.dequeue();
 *
 * // retrievedString === 'Hello World!'
 * ```
 */
 queueAllAsync(iterable: AsyncIterable<T>): Promise<void>;
 /**
 * Dequeue an element, waiting for data to be available if necessary.
 *
 * @returns A promise which is fulfilled when an element (as queued by
 * queue()) becomes available.
 * If multiple dequeus() are issued sequentially, it is
 * implementation-defined whether they are fulfilled in the same
 * order or not. However, the data is still retrieved in FIFO
 * fashion, meaning the first fulfilled promise gets the first
 * element, the second fulfilled the second one and so forth.
 */
 dequeue(): Promise<T>;
 /**
 * Dequeue an element if available or throw an exception otherwise.
 *
 * @returns The first element of the queue.
 * @throws An exception if the queue is empty at the time of the call.
 */
 poll(): T;
 /**
 * Return the current size at the moment of the call.
 *
 * Even though code like
 * ```
 * if (queue.size() >= 1) {
 * const element = queue.poll();
 * }
 * ```
 * is technically not wrong (due to JS' execution model), users are
 * advised to avoid this pattern. Instead, users are encouraged to
 *
 * - in cases where waiting for a promise is impossible, to use
 * {@link poll} and catch the exception,
 * - or to use {@link dequeue} with JS' `await` or
 * `queue.dequeue().then(...)`.
 */
 size(): number;
}
export class NoElementError extends Error {
}
export class AsyncQueue<T> implements IAsyncQueue<T> {
 private buffer: T[] = [];
 private elementSem: ISemaphore = new Semaphore(0);
 public queue(data: T): void {
 this.buffer.push(data);
 this.elementSem.free();
 }
 public queueAll(iterable: Iterable<T>): void {
 for (const element of iterable) {
 this.queue(element);
 }
 }
 public async queueAllAsync(iterable: AsyncIterable<T>): Promise<void> {
 for await (const element of iterable) {
 this.queue(element);
 }
 }
 public async dequeue(): Promise<T> {
 await this.elementSem.take();
 try {
 return this.poll();
 }
 catch (err) {
 if (err instanceof NoElementError) {
 throw new Error('AsyncQueue dequeue: poll() threw an exception \
 even though dequeue() waited for its element semaphore to be available via take().');
 }
 else {
 throw err;
 }
 }
 }
 public poll(): T {
 if (this.buffer.length >= 1) {
 const dequeuedElement = this.buffer.shift();
 // Force-cast the element since we know that the buffer contains
 // at least one element and JS' execution model prohibits other
 // interleaving fibers to modify the buffer (=> no race condition).
 //
 // Also, we cannot check for shift() returning undefined as the queue
 // might well contain "undefined" as such.
 return (dequeuedElement as T);
 }
 else {
 throw new NoElementError();
 }
 }
 public size(): number {
 return this.buffer.length;
 }
 public async *[Symbol.asyncIterator](): AsyncIterableIterator<T> {
 while (true) {
 yield this.dequeue();
 }
 }
}

AsyncLimitedQueue:

import { IAsyncQueue, AsyncQueue } from './AsyncQueue';
import { ISemaphore, Semaphore } from '../semaphore/index';
/**
 * Asynchronous entrance-limited FIFO queue with a Promise-driven dequeue operation.
 *
 * Contrary to {@link IAsyncQueue}, the queue operation is Promise-driven as well,
 * e.g. implementations might delay entrance into the queue, e.g. to enforce a
 * limit on the number of elements stored in the queue at the same time, cf.
 * {@link AsyncLimitedQueue}.
 * Other types of entrance limitations are conceivable as well, such as a
 * restriction on the sum of contained elements in case of a number queue.
 *
 * All element values are allowed, especially falsy ones, e.g.
 * false, 0, undefined, null, [], {} are all valid elements which
 * can be queued and dequeued.
 *
 * {@link queue IAsyncLimitedQueue#queue} operations are possibly delayed and
 * executed in implementation-dependenent order.
 *
 * @example Issueing multiple {@link queue} operations without awaiting the
 * previous ones may result in implementation-defined insertion order.
 * ```
 * queue.queue(1);
 * queue.queue(2);
 *
 * await queue.dequeue(); // can be 1 or 2
 * await queue.dequeue(); // can be 1 or 2 as well (the remaining number)
 * ```
 *
 * @example If you would like to retain the order, await the {@link queue}
 * operations, use {@link queueAll IAsyncLimitedQueue#queueAll} or
 * {@link queueAllAsync IAsyncLimitedQueue#queueAllAsync}.
 * ```
 * await queue.queue(1);
 * await queue.queue(2);
 * ```
 * ```
 * queue.queueAll([1, 2]);
 * ```
 *
 * The {@link AsyncIterable} interface iterates the queue's (future) contents
 * ad infinitum. Users are advised to signal the end by manual insertion of a
 * special value (a so-called poison pill), see {@link IAsyncQueue}.
 */
export interface IAsyncLimitedQueue<T> extends AsyncIterable<T> {
 /**
 * Queue an element, waiting for entrance if necessary.
 *
 * @example
 * ```
 * queue.queue(42).then(() => {
 * // 42 is now stored within the queue
 * });
 * ```
 */
 queue(data: T): Promise<void>;
 /**
 * Queue all elements of an iterable, e.g. an array or a generator function.
 * @see IAsyncQueue#queueAll
 */
 queueAll(iterable: Iterable<T>): Promise<void>;
 /**
 * Queue all elements of an asynchronous iterable, e.g. an asynchronous
 * generator functions.
 *
 * @see IAsyncQueue#queueAllAsync
 */
 queueAllAsync(iterable: AsyncIterable<T>): Promise<void>;
 /**
 * Offer an element, only queueing it if entrance is available at the time
 * of the call.
 *
 * @returns True if the element could be inserted right away. False
 * otherwise.
 */
 offer(data: T): boolean;
 /**
 * Offer all elements of an iterable for in-order insertion.
 *
 * @param iterable An iterable whose first (limit - queue.size()) elements
 * will be inserted. Iterables which iterate an infinite
 * number of elements can also be passed and will *not*
 * result in an endless loop.
 *
 * @returns The number of elements, which could be inserted right away.
 * Possibly 0 when the queue was full at the time of the call.
 */
 offerAll(iterable: Iterable<T>): number;
 /**
 * Offer all elements of an asynchronous iterable for in-order insertion.
 *
 * @param iterable An iterable whose elements will be {@link offer}ed
 * in-order for this queue.
 * The method will stop querying and offering further
 * elements upon the first {@link offer} call, which
 * returns `false`.
 * <br>
 * Contrary to {@link offerAll}, iterables iterating an
 * infinite number of elements might prevent the Promise,
 * which {@link offerAllAsync} returns, from ever resolving.
 * <br>
 * This depends on {@link dequeue} operations which could
 * get scheduled by the JS VM while elements from the passed
 * asynchronous iterator are accessed.
 *
 * @returns A promise resolving to the number of elements, which could be
 * inserted (offered successfully) consecutively without waiting.
 * Possibly 0 when the queue was full at the time of the call.
 * Fulfillment of this promise is not guaranteed in case of infinite
 * iterables.
 */
 offerAllAsync(iterable: AsyncIterable<T>): Promise<number>;
 /**
 * Dequeue an element if available or throw an exception otherwise.
 *
 * @returns The first element of the queue.
 * @throws An exception if the queue is empty at the time of the call.
 */
 poll(): T;
 /**
 * Dequeue an element, waiting for data to be available if necessary.
 *
 * @returns A promise which is fulfilled when an element (as queued by
 * queue()) becomes available.
 * If multiple dequeus() are issued sequentially, it is
 * implementation-defined whether they are fulfilled in the same
 * order or not. However, the data is still retrieved in FIFO
 * fashion, meaning the first fulfilled promise gets the first
 * element, the second fulfilled the second one and so forth.
 */
 dequeue(): Promise<T>;
 /**
 * Return the current size at the moment of the call.
 *
 * Even though code like
 * ```
 * if (queue.size() >= 1) {
 * const element = queue.poll();
 * }
 * ```
 * is technically not wrong (due to JS' execution model), users are
 * advised to avoid this pattern. Instead, users are encouraged to
 *
 * - in cases where waiting for a promise is impossible, to use
 * {@link poll} and catch the exception,
 * - or to use {@link dequeue} with JS' `await` or
 * `queue.dequeue().then(...)`.
 */
 size(): number;
}
/**
 * Asynchronous element-limited FIFO queue with a Promise-driven dequeue operation.
 *
 * {@link AsyncLimitedQueue#queue} operations are delayed (in unspecified order)
 * until space becomes available through dequeue operations.
 */
export class AsyncLimitedQueue<T> implements IAsyncLimitedQueue<T> {
 private limitSem: ISemaphore;
 /**
 * Initialize the queue.
 * @param limit A integer >= 1 specifying the number of elements after which
 * queue() effectively blocks (i.e. the promise returned by it
 * does not get "immediately" fulfilled for some informal value
 * of immediately).
 * @param storageQueue An asynchronous (non-limiting) queue backing the data.
 * It defaults to a AsyncQueue.
 *
 * @throws An exception in case the limit is not an integer or is <= 0.
 */
 public constructor(limit: number, private storageQueue: IAsyncQueue<T> = new AsyncQueue()) {
 if (!Number.isInteger(limit) || limit <= 0) {
 throw new Error('AsyncLimitedQueue: Illegal limit (non-integer or\
 <= 0) on queued elements. It must be an integer >= 1.');
 }
 this.limitSem = new Semaphore(limit);
 }
 public async queue(data: T): Promise<void> {
 await this.limitSem.take();
 this.storageQueue.queue(data);
 }
 public async queueAll(iterable: Iterable<T>): Promise<void> {
 for (const element of iterable) {
 await this.queue(element);
 }
 }
 public async queueAllAsync(iterable: AsyncIterable<T>): Promise<void> {
 for await (const element of iterable) {
 await this.queue(element);
 }
 }
 public offer(data: T): boolean {
 if (this.limitSem.tryTake()) {
 this.storageQueue.queue(data);
 return true;
 }
 else {
 return false;
 }
 }
 public offerAll(iterable: Iterable<T>): number {
 let insertedElements = 0;
 for (const element of iterable) {
 if (!this.offer(element)) {
 return insertedElements;
 }
 insertedElements++;
 }
 return insertedElements;
 }
 public async offerAllAsync(iterable: AsyncIterable<T>): Promise<number> {
 let insertedElements = 0;
 for await (const element of iterable) {
 if (!this.offer(element)) {
 return insertedElements;
 }
 insertedElements++;
 }
 return insertedElements;
 }
 public async dequeue(): Promise<T> {
 return this.storageQueue.dequeue().then(element => {
 this.limitSem.free();
 return element;
 });
 }
 public poll(): T {
 return this.storageQueue.poll();
 }
 public async *[Symbol.asyncIterator](): AsyncIterableIterator<T> {
 while (true) {
 yield this.dequeue();
 }
 }
 public size(): number {
 return this.storageQueue.size();
 }
}