@terascope/job-components/dist/operations/parallel-slicer.js

"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
const utils_1 = require("@terascope/utils");
const slicer_core_1 = __importDefault(require("./core/slicer-core"));
/**
 * A varient of a "Slicer" for running a parallel stream of slicers.
 * @see SlicerCore
 */
class ParallelSlicer extends slicer_core_1.default {
    constructor() {
        super(...arguments);
        this._slicers = [];
    }
    /**
     * Register the different Slicer instances
     * @see SlicerCore#initialize
    */
    async initialize(recoveryData) {
        await super.initialize(recoveryData);
        const { slicers = 1 } = this.executionConfig;
        const promises = utils_1.times(slicers, async (id) => {
            const fn = await this.newSlicer();
            if (!utils_1.isFunction(fn))
                return;
            this._slicers.push({
                done: false,
                fn,
                id,
                processing: false,
                order: 0,
            });
        });
        await Promise.all(promises);
    }
    /**
     * Cleanup the slicers functions
     * @see SlicerCore#shutdown
    */
    async shutdown() {
        this._slicers.length = 0;
        return super.shutdown();
    }
    slicers() {
        return this._slicers.length;
    }
    async handle() {
        if (this.isFinished)
            return true;
        const promises = this._slicers
            .filter((slicer) => !slicer.processing)
            .map((slicer) => this.processSlicer(slicer));
        await Promise.race(promises);
        return this.isFinished;
    }
    get isFinished() {
        return this._slicers.every((slicer) => slicer.done);
    }
    async processSlicer(slicer) {
        if (slicer.done || slicer.processing)
            return;
        slicer.processing = true;
        let result;
        try {
            result = await slicer.fn();
        }
        finally {
            slicer.processing = false;
        }
        if (result == null && this.canComplete()) {
            this.logger.info(`slicer ${slicer.id} has completed its range`);
            slicer.done = true;
            this.events.emit('slicer:done', slicer.id);
        }
        else if (result != null) {
            if (Array.isArray(result)) {
                this.events.emit('slicer:subslice');
                result.forEach((item) => {
                    slicer.order += 1;
                    this.createSlice(item, slicer.order, slicer.id);
                });
            }
            else {
                slicer.order += 1;
                this.createSlice(result, slicer.order, slicer.id);
            }
        }
    }
}
exports.default = ParallelSlicer;
//# sourceMappingURL=data:application/json;base64,eyJ2ZXJzaW9uIjozLCJmaWxlIjoicGFyYWxsZWwtc2xpY2VyLmpzIiwic291cmNlUm9vdCI6IiIsInNvdXJjZXMiOlsiLi4vLi4vc3JjL29wZXJhdGlvbnMvcGFyYWxsZWwtc2xpY2VyLnRzIl0sIm5hbWVzIjpbXSwibWFwcGluZ3MiOiI7Ozs7O0FBQUEsNENBQXFEO0FBRXJELHFFQUE0QztBQUU1Qzs7O0dBR0c7QUFFSCxNQUE4QixjQUE2QixTQUFRLHFCQUFhO0lBQWhGOztRQUNjLGFBQVEsR0FBZ0IsRUFBRSxDQUFDO0lBMEZ6QyxDQUFDO0lBeEZHOzs7TUFHRTtJQUNGLEtBQUssQ0FBQyxVQUFVLENBQUMsWUFBc0I7UUFDbkMsTUFBTSxLQUFLLENBQUMsVUFBVSxDQUFDLFlBQVksQ0FBQyxDQUFDO1FBQ3JDLE1BQU0sRUFBRSxPQUFPLEdBQUcsQ0FBQyxFQUFFLEdBQUcsSUFBSSxDQUFDLGVBQWUsQ0FBQztRQUU3QyxNQUFNLFFBQVEsR0FBRyxhQUFLLENBQUMsT0FBTyxFQUFFLEtBQUssRUFBRSxFQUFFLEVBQUUsRUFBRTtZQUN6QyxNQUFNLEVBQUUsR0FBRyxNQUFNLElBQUksQ0FBQyxTQUFTLEVBQUUsQ0FBQztZQUNsQyxJQUFJLENBQUMsa0JBQVUsQ0FBQyxFQUFFLENBQUM7Z0JBQUUsT0FBTztZQUU1QixJQUFJLENBQUMsUUFBUSxDQUFDLElBQUksQ0FBQztnQkFDZixJQUFJLEVBQUUsS0FBSztnQkFDWCxFQUFFO2dCQUNGLEVBQUU7Z0JBQ0YsVUFBVSxFQUFFLEtBQUs7Z0JBQ2pCLEtBQUssRUFBRSxDQUFDO2FBQ1gsQ0FBQyxDQUFDO1FBQ1AsQ0FBQyxDQUFDLENBQUM7UUFFSCxNQUFNLE9BQU8sQ0FBQyxHQUFHLENBQUMsUUFBUSxDQUFDLENBQUM7SUFDaEMsQ0FBQztJQUVEOzs7TUFHRTtJQUNGLEtBQUssQ0FBQyxRQUFRO1FBQ1YsSUFBSSxDQUFDLFFBQVEsQ0FBQyxNQUFNLEdBQUcsQ0FBQyxDQUFDO1FBQ3pCLE9BQU8sS0FBSyxDQUFDLFFBQVEsRUFBRSxDQUFDO0lBQzVCLENBQUM7SUFRRCxPQUFPO1FBQ0gsT0FBTyxJQUFJLENBQUMsUUFBUSxDQUFDLE1BQU0sQ0FBQztJQUNoQyxDQUFDO0lBRUQsS0FBSyxDQUFDLE1BQU07UUFDUixJQUFJLElBQUksQ0FBQyxVQUFVO1lBQUUsT0FBTyxJQUFJLENBQUM7UUFFakMsTUFBTSxRQUFRLEdBQUcsSUFBSSxDQUFDLFFBQVE7YUFDekIsTUFBTSxDQUFDLENBQUMsTUFBTSxFQUFFLEVBQUUsQ0FBQyxDQUFDLE1BQU0sQ0FBQyxVQUFVLENBQUM7YUFDdEMsR0FBRyxDQUFDLENBQUMsTUFBTSxFQUFFLEVBQUUsQ0FBQyxJQUFJLENBQUMsYUFBYSxDQUFDLE1BQU0sQ0FBQyxDQUFDLENBQUM7UUFFakQsTUFBTSxPQUFPLENBQUMsSUFBSSxDQUFDLFFBQVEsQ0FBQyxDQUFDO1FBQzdCLE9BQU8sSUFBSSxDQUFDLFVBQVUsQ0FBQztJQUMzQixDQUFDO0lBRUQsSUFBSSxVQUFVO1FBQ1YsT0FBTyxJQUFJLENBQUMsUUFBUSxDQUFDLEtBQUssQ0FBQyxDQUFDLE1BQU0sRUFBRSxFQUFFLENBQUMsTUFBTSxDQUFDLElBQUksQ0FBQyxDQUFDO0lBQ3hELENBQUM7SUFFTyxLQUFLLENBQUMsYUFBYSxDQUFDLE1BQWlCO1FBQ3pDLElBQUksTUFBTSxDQUFDLElBQUksSUFBSSxNQUFNLENBQUMsVUFBVTtZQUFFLE9BQU87UUFFN0MsTUFBTSxDQUFDLFVBQVUsR0FBRyxJQUFJLENBQUM7UUFDekIsSUFBSSxNQUFvQixDQUFDO1FBRXpCLElBQUk7WUFDQSxNQUFNLEdBQUcsTUFBTSxNQUFNLENBQUMsRUFBRSxFQUFFLENBQUM7U0FDOUI7Z0JBQVM7WUFDTixNQUFNLENBQUMsVUFBVSxHQUFHLEtBQUssQ0FBQztTQUM3QjtRQUVELElBQUksTUFBTSxJQUFJLElBQUksSUFBSSxJQUFJLENBQUMsV0FBVyxFQUFFLEVBQUU7WUFDdEMsSUFBSSxDQUFDLE1BQU0sQ0FBQyxJQUFJLENBQUMsVUFBVSxNQUFNLENBQUMsRUFBRSwwQkFBMEIsQ0FBQyxDQUFDO1lBQ2hFLE1BQU0sQ0FBQyxJQUFJLEdBQUcsSUFBSSxDQUFDO1lBRW5CLElBQUksQ0FBQyxNQUFNLENBQUMsSUFBSSxDQUFDLGFBQWEsRUFBRSxNQUFNLENBQUMsRUFBRSxDQUFDLENBQUM7U0FDOUM7YUFBTSxJQUFJLE1BQU0sSUFBSSxJQUFJLEVBQUU7WUFDdkIsSUFBSSxLQUFLLENBQUMsT0FBTyxDQUFDLE1BQU0sQ0FBQyxFQUFFO2dCQUN2QixJQUFJLENBQUMsTUFBTSxDQUFDLElBQUksQ0FBQyxpQkFBaUIsQ0FBQyxDQUFDO2dCQUNwQyxNQUFNLENBQUMsT0FBTyxDQUFDLENBQUMsSUFBSSxFQUFFLEVBQUU7b0JBQ3BCLE1BQU0sQ0FBQyxLQUFLLElBQUksQ0FBQyxDQUFDO29CQUNsQixJQUFJLENBQUMsV0FBVyxDQUFDLElBQUksRUFBRSxNQUFNLENBQUMsS0FBSyxFQUFFLE1BQU0sQ0FBQyxFQUFFLENBQUMsQ0FBQztnQkFDcEQsQ0FBQyxDQUFDLENBQUM7YUFDTjtpQkFBTTtnQkFDSCxNQUFNLENBQUMsS0FBSyxJQUFJLENBQUMsQ0FBQztnQkFDbEIsSUFBSSxDQUFDLFdBQVcsQ0FBQyxNQUFNLEVBQUUsTUFBTSxDQUFDLEtBQUssRUFBRSxNQUFNLENBQUMsRUFBRSxDQUFDLENBQUM7YUFDckQ7U0FDSjtJQUNMLENBQUM7Q0FDSjtBQTNGRCxpQ0EyRkMifQ==