promise-batcher/index.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const Debug = require("debug");
const defer = require("p-defer");
const debug = Debug("promise-batcher");
function isNull(val) {
    return val === undefined || val === null;
}
class BatcherToken {
}
exports.BatcherToken = BatcherToken;
/**
 * If this token is returned in the results from a batchingFunction, the corresponding requests will be placed back
 * into the the head of the queue.
 */
exports.BATCHER_RETRY_TOKEN = new BatcherToken();
// tslint:disable-next-line:max-classes-per-file
class Batcher {
    constructor(options) {
        this._maxBatchSize = Infinity;
        this._queuingDelay = 1;
        this._inputQueue = [];
        this._outputQueue = [];
        this._waiting = false;
        this._activePromiseCount = 0;
        this._immediateCount = 0;
        this._batchingFunction = options.batchingFunction;
        this._delayFunction = options.delayFunction;
        if (Array.isArray(options.queuingThresholds)) {
            if (!options.queuingThresholds.length) {
                throw new Error("options.batchThresholds must contain at least one number");
            }
            options.queuingThresholds.forEach((n) => {
                if (n < 1) {
                    throw new Error("options.batchThresholds must only contain numbers greater than 0");
                }
            });
            this._queuingThresholds = options.queuingThresholds.slice();
        }
        else {
            this._queuingThresholds = [1];
        }
        if (!isNull(options.maxBatchSize)) {
            if (options.maxBatchSize < 1) {
                throw new Error("options.batchSize must be greater than 0");
            }
            this._maxBatchSize = options.maxBatchSize;
        }
        if (!isNull(options.queuingDelay)) {
            if (options.queuingDelay < 0) {
                throw new Error("options.queuingDelay must be greater than or equal to 0");
            }
            this._queuingDelay = options.queuingDelay;
        }
    }
    /**
     * Returns a promise which resolves or rejects with the individual result returned from the batching function.
     */
    getResult(input) {
        const index = this._inputQueue.length;
        debug("Queuing request at index %O", index);
        this._inputQueue[index] = input;
        const deferred = defer();
        this._outputQueue[index] = deferred;
        this._trigger();
        return deferred.promise;
    }
    /**
     * Triggers a batch to run, bypassing the queuingDelay while respecting other imposed delays.
     */
    send() {
        debug("Send triggered.");
        // no inputs?
        // delayed?
        this._immediateCount = this._inputQueue.length;
        this._trigger();
    }
    /**
     * Triggers a batch to run, adhering to the maxBatchSize, queueingThresholds, and queuingDelay
     */
    _trigger() {
        // If the batch is set to run immediately, there is nothing more to be done
        if (this._waiting && !this._waitTimeout) {
            return;
        }
        // Always obey the queuing threshold
        const thresholdIndex = Math.min(this._activePromiseCount, this._queuingThresholds.length - 1);
        if (this._inputQueue.length < this._queuingThresholds[thresholdIndex]) {
            return;
        }
        // If the queue has reached the maximum batch size, start it immediately
        if (this._inputQueue.length >= this._maxBatchSize || this._immediateCount) {
            debug("Running immediately.");
            if (this._waitTimeout) {
                clearTimeout(this._waitTimeout);
                this._waitTimeout = undefined;
            }
            this._waiting = true;
            this._run();
            return;
        }
        if (this._waiting) {
            return;
        }
        // Run the batch, but with a delay
        this._waiting = true;
        debug("Running in %Oms (thresholdIndex %O).", this._queuingDelay, thresholdIndex);
        // Tests showed that nextTick would commonly run before promises could resolve.
        // SetImmediate would run later than setTimeout as well.
        this._waitTimeout = setTimeout(() => {
            this._waitTimeout = undefined;
            this._run();
        }, this._queuingDelay);
    }
    /**
     * Runs the batch, while respecting delays imposed by the supplied delayFunction
     */
    _run() {
        if (this._delayFunction) {
            let result;
            try {
                result = this._delayFunction();
            }
            catch (err) {
                result = Promise.reject(err);
            }
            if (!isNull(result)) {
                const resultPromise = result instanceof Promise ? result : Promise.resolve(result);
                resultPromise.then(() => {
                    this._runImmediately();
                }).catch((err) => {
                    debug("Caught error in delayFunction. Rejecting promises.");
                    this._inputQueue.length = 0;
                    const promises = this._outputQueue.splice(0, this._outputQueue.length);
                    promises.forEach((promise) => {
                        promise.reject(err);
                    });
                    this._waiting = false;
                });
                return;
            }
            debug("Bypassing batch delay.");
        }
        this._runImmediately();
    }
    /**
     * Runs the batch immediately without further delay
     */
    _runImmediately() {
        const inputs = this._inputQueue.splice(0, this._maxBatchSize);
        const outputPromises = this._outputQueue.splice(0, this._maxBatchSize);
        if (this._immediateCount) {
            this._immediateCount = Math.max(0, this._immediateCount - inputs.length);
        }
        debug("Running batch of %O", inputs.length);
        let batchPromise;
        try {
            batchPromise = this._batchingFunction.call(this, inputs);
            if (!(batchPromise instanceof Promise)) {
                batchPromise = Promise.resolve(batchPromise);
            }
        }
        catch (err) {
            batchPromise = Promise.reject(err);
        }
        this._waiting = false;
        this._activePromiseCount++;
        batchPromise.then((outputs) => {
            if (!Array.isArray(outputs)) {
                throw new Error("Invalid type returned from batching function.");
            }
            debug("Promise resolved.");
            if (outputs.length !== outputPromises.length) {
                throw new Error("Batching function output length does not equal the input length.");
            }
            const retryInputs = [];
            const retryPromises = [];
            outputPromises.forEach((promise, index) => {
                const output = outputs[index];
                if (output === exports.BATCHER_RETRY_TOKEN) {
                    retryInputs.push(inputs[index]);
                    retryPromises.push(promise);
                }
                else if (output instanceof Error) {
                    promise.reject(output);
                }
                else {
                    promise.resolve(output);
                }
            });
            if (retryPromises.length) {
                debug("Adding %O requests to the queue to retry.", retryPromises.length);
                if (this._immediateCount) {
                    this._immediateCount += retryPromises.length;
                }
                this._inputQueue.unshift(...retryInputs);
                this._outputQueue.unshift(...retryPromises);
            }
        }).catch((err) => {
            outputPromises.forEach((promise) => {
                promise.reject(err);
            });
        }).then(() => {
            this._activePromiseCount--;
            // Since we may be operating at a lower queuing threshold now, we should try run again
            this._trigger();
        });
        // The batch has started. Trigger another batch if appropriate.
        this._trigger();
    }
}
exports.Batcher = Batcher;