Source: promise-utils.js

"use strict";

/**
 * Creates a non-clearable timer that resolves the promise once elapses.
 * @param {number} ms
 * @returns {Promise<void>}
 */
function delay(ms) {
    return new Promise((r) => setTimeout(r, ms || 0));
}

/**
 * Creates a Promise that gets resolved or rejected based on an event.
 * @param {object} emitter
 * @param {string} eventName
 * @returns {Promise}
 */
function fromEvent(emitter, eventName) {
    return new Promise((resolve, reject) =>
        emitter.once(eventName, (err, result) => {
            if (err) {
                reject(err);
            } else {
                resolve(result);
            }
        }),
    );
}

/**
 * Creates a Promise from a callback based function.
 * @param {Function} fn
 * @returns {Promise}
 */
function fromCallback(fn) {
    return new Promise((resolve, reject) =>
        fn((err, result) => {
            if (err) {
                reject(err);
            } else {
                resolve(result);
            }
        }),
    );
}

/**
 * Gets a function that has the signature of a callback that invokes the appropriate promise handler parameters.
 * @param {Function} resolve
 * @param {Function} reject
 * @returns {Function}
 */
function getCallback(resolve, reject) {
    return function (err, result) {
        if (err) {
            reject(err);
        } else {
            resolve(result);
        }
    };
}

async function invokeSequentially(info, length, fn) {
    let index;
    while ((index = info.counter++) < length) {
        await fn(index);
    }
}

/**
 * Invokes the new query plan of the load balancing policy and returns a Promise.
 * @param {LoadBalancingPolicy} lbp The load balancing policy.
 * @param {String} keyspace Name of currently logged keyspace at `Client` level.
 * @param {ExecutionOptions|null} executionOptions The information related to the execution of the request.
 * @returns {Promise<Iterator>}
 */
function newQueryPlan(lbp, keyspace, executionOptions) {
    return new Promise((resolve, reject) => {
        lbp.newQueryPlan(keyspace, executionOptions, (err, iterator) => {
            if (err) {
                reject(err);
            } else {
                resolve(iterator);
            }
        });
    });
}

/**
 * Method that handles optional callbacks (dual promise and callback support).
 * When callback is undefined it returns the promise.
 * When using a callback, it will use it as handlers of the continuation of the promise.
 * @param {Promise} promise
 * @param {Function?} callback
 * @returns {Promise|undefined}
 */
function optionalCallback(promise, callback) {
    if (!callback) {
        return promise;
    }

    toCallback(promise, callback);
}

/**
 * Invokes the provided function multiple times, considering the concurrency level limit.
 * @param {Number} count
 * @param {Number} limit
 * @param {Function} fn
 * @returns {Promise}
 */
function times(count, limit, fn) {
    if (limit > count) {
        limit = count;
    }

    const promises = new Array(limit);

    const info = {
        counter: 0,
    };

    for (let i = 0; i < limit; i++) {
        promises[i] = invokeSequentially(info, count, fn);
    }

    return Promise.all(promises);
}

/**
 * Deals with unexpected rejections in order to avoid the unhandled promise rejection warning or failure.
 * @param {Promise} promise
 * @returns {undefined}
 */
function toBackground(promise) {
    promise.catch(() => {});
}

/**
 * Invokes the callback once outside the promise chain the promise is resolved or rejected.
 * @param {Promise} promise
 * @param {Function?} callback
 * @returns {undefined}
 */
function toCallback(promise, callback) {
    promise.then(
        (result) => process.nextTick(() => callback(null, result)),
        // Avoid marking the promise as rejected
        (err) => process.nextTick(() => callback(err)),
    );
}

module.exports = {
    delay,
    fromCallback,
    fromEvent,
    getCallback,
    newQueryPlan,
    optionalCallback,
    times,
    toBackground,
    toCallback,
};