@stoplight/graphite/scheduler/scheduler.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const tslib_1 = require("tslib");
const lifecycle_1 = require("@stoplight/lifecycle");
const immer_1 = require("immer");
const lodash_1 = require("lodash");
const mobx_1 = require("mobx");
const neo_async_1 = require("neo-async");
const errorReporter_1 = require("../errorReporter");
const scheduler_1 = require("../scheduler");
const types_1 = require("../types");
function createScheduler(props) {
    return new Scheduler(props);
}
exports.createScheduler = createScheduler;
class Scheduler {
    constructor(props) {
        this._queued = {};
        this._handlers = handlerMap();
        this._running = {};
        this.mightBeStuck = false;
        this.queue = (task) => {
            if (!this._handlers[task.op].length)
                return Promise.resolve();
            const taskId = computeTaskId(task);
            let queued;
            if (taskId) {
                queued = this._queued[taskId];
            }
            if (!queued) {
                queued = new Promise((resolve, reject) => {
                    this._queue.push(task, priorityMap[task.op] || 99, err => {
                        if (taskId) {
                            delete this._queued[taskId];
                        }
                        if (err) {
                            errorReporter_1.errorReporter.reportError({
                                code: types_1.GraphiteErrorCode.Generic,
                                message: `Error running task ${task.op} ${err}`,
                                trace: task.trace,
                                nodeId: task.nodeId,
                            });
                            reject(err);
                        }
                        resolve();
                    });
                });
                if (taskId) {
                    this._queued[taskId] = queued;
                }
            }
            return queued;
        };
        this.queueAll = (tasks) => {
            return tasks.map(task => this.queue(task));
        };
        this.registerHandler = (op, handler) => {
            this._handlers[op].push(handler);
            return lifecycle_1.createDisposable(() => {
                const idx = this._handlers[op].indexOf(handler);
                if (idx >= 0) {
                    this._handlers[op].splice(idx, 1);
                }
            });
        };
        this.drain = () => tslib_1.__awaiter(this, void 0, void 0, function* () {
            yield this.drainQueue();
            yield mobx_1.when(() => !this.isBusy);
        });
        this._run = (task) => {
            const node = this._graph.getNodeById(task.nodeId);
            if (!node) {
                throw new Error(`Cannot find node with id ${task.nodeId} for task ${scheduler_1.GraphTaskOp[task.op]}.`);
            }
            const handler = findHandler(this._handlers[task.op], node);
            if (!handler)
                return;
            return handler.run(node, {
                task,
                getNodeById: this._graph.getNodeById,
                getNodeByUri: this._graph.getNodeByUri,
                getNodesByType: this._graph.getNodesByType,
                moveNode: this._graph.moveNode,
                applyPatch: patch => {
                    return this._graph.applyPatch({
                        operations: patch.operations,
                        trace: Object.assign({}, patch.trace, { sourceOp: task.op }),
                    });
                },
                removeNode: (id, trace = {}) => {
                    return this._graph.removeNode(id, Object.assign({}, trace, { sourceOp: task.op }));
                },
                addNode: (props, trace = {}) => {
                    return this._graph.addNode(props, Object.assign({}, trace, { sourceOp: task.op }));
                },
                setSourceNodeProp: (id, prop, value, trace = {}) => {
                    return this._graph.setSourceNodeProp(id, prop, value, Object.assign({}, trace, { sourceOp: task.op }));
                },
                patchSourceNodeProp: (id, prop, parsed, trace = {}) => {
                    return this._graph.patchSourceNodeProp(id, prop, parsed, Object.assign({}, trace, { sourceOp: task.op }));
                },
                reportError: (nodeId, error, trace = {}) => {
                    return this._graph.reportError(nodeId, error, immer_1.default(trace, source => {
                        source.sourceOp = task.op;
                    }));
                },
                setSourceNodeDiagnostics: (id, diagnostics, trace = {}) => {
                    return this._graph.setSourceNodeDiagnostics(id, handler.id, diagnostics, Object.assign({}, trace, { sourceOp: task.op }));
                },
                runTask: (t) => {
                    return this.run(t);
                },
                resolver: this._resolver,
            });
        };
        this.drainQueue = () => {
            return new Promise((resolve, reject) => {
                if (this.queueLength()) {
                    this._queue.drain = resolve;
                    this._queue.error = reject;
                }
                else {
                    resolve();
                }
            });
        };
        this._runQueueTask = (task, cb) => {
            const taskId = computeTaskId(task);
            this._running[taskId] = true;
            try {
                const result = this._run(task);
                if (result instanceof Promise) {
                    result
                        .then(r => {
                        this._handleRunResult(r);
                        delete this._running[taskId];
                        cb();
                    })
                        .catch(cb)
                        .then(() => delete this._running[taskId]);
                }
                else {
                    this._handleRunResult(result);
                    delete this._running[taskId];
                    setTimeout(cb, 0);
                }
            }
            catch (e) {
                delete this._running[taskId];
                cb(e);
            }
        };
        this._handleRunResult = (result) => {
            if (result) {
                this.queueAll(result);
            }
        };
        this._graph = props.graph;
        this._queue = neo_async_1.priorityQueue(this._runQueueTask, 1);
        this._resolver = props.resolver;
        mobx_1.reaction(() => this.queuedCount, mobx_1.action(() => {
            this.mightBeStuck = false;
            if (this._mightBeStuckTimeout)
                clearTimeout(this._mightBeStuckTimeout);
            this._mightBeStuckTimeout = setTimeout(mobx_1.action(() => {
                this.mightBeStuck = this.queuedCount !== 0;
                this._mightBeStuckTimeout = void 0;
            }), 5000);
        }));
    }
    get isBusy() {
        return this.queuedCount !== 0;
    }
    get queuedCount() {
        return lodash_1.size(this._queued);
    }
    snapshot() {
        const tasks = [];
        let task = lodash_1.get(this._queue, '_tasks.head');
        while (task !== null && typeof task === 'object') {
            tasks.push(task.data);
            task = task.next;
        }
        const runningKeys = Object.keys(this._running);
        const running = [];
        for (const key of runningKeys) {
            const [op, id] = key.split(':');
            let node;
            try {
                node = this._graph.getNodeById(id);
            }
            catch (e) {
            }
            running.push({
                id,
                op: Number(op),
                type: node && node.type,
                spec: node && node.spec,
                path: node && node.path,
                language: node && node.language,
            });
        }
        return {
            running,
            queued: Object.keys(this._queued),
            tasks,
        };
    }
    queueLength() {
        return this._queue.length();
    }
    run(task) {
        const taskId = computeTaskId(task);
        const queued = this._queued[taskId];
        if (queued && this._running[taskId]) {
            return queued;
        }
        this._queue.remove((t) => taskId === computeTaskId(t));
        const taskPromise = new Promise((resolve, reject) => this._runQueueTask(task, err => {
            if (taskId) {
                delete this._queued[taskId];
            }
            if (err) {
                console.error(`Error running task ${task.op}`, err);
                reject(err);
            }
            resolve();
        }));
        this._queued[taskId] = taskPromise;
        return taskPromise;
    }
}
tslib_1.__decorate([
    mobx_1.observable,
    tslib_1.__metadata("design:type", Object)
], Scheduler.prototype, "_queued", void 0);
tslib_1.__decorate([
    mobx_1.observable,
    tslib_1.__metadata("design:type", Object)
], Scheduler.prototype, "mightBeStuck", void 0);
tslib_1.__decorate([
    mobx_1.computed,
    tslib_1.__metadata("design:type", Object),
    tslib_1.__metadata("design:paramtypes", [])
], Scheduler.prototype, "isBusy", null);
tslib_1.__decorate([
    mobx_1.computed,
    tslib_1.__metadata("design:type", Object),
    tslib_1.__metadata("design:paramtypes", [])
], Scheduler.prototype, "queuedCount", null);
function computeTaskId(task) {
    return `${task.op}:${task.nodeId}`;
}
const priorityMap = {
    [scheduler_1.GraphTaskOp.SerializeSourceNode]: 1,
    [scheduler_1.GraphTaskOp.DeserializeSourceNode]: 1,
    [scheduler_1.GraphTaskOp.DiffRawToParsed]: 1,
    [scheduler_1.GraphTaskOp.ComputeSourceMap]: 2,
    [scheduler_1.GraphTaskOp.ReadSourceNode]: 3,
    [scheduler_1.GraphTaskOp.WriteSourceNode]: 3,
    [scheduler_1.GraphTaskOp.DeleteSourceNode]: 3,
    [scheduler_1.GraphTaskOp.MoveSourceNode]: 3,
    [scheduler_1.GraphTaskOp.ResolveSourceNode]: 4,
    [scheduler_1.GraphTaskOp.TransformParsed]: 5,
    [scheduler_1.GraphTaskOp.ValidateSourceNode]: 6,
};
const handlerMap = () => ({
    [scheduler_1.GraphTaskOp.SerializeSourceNode]: [],
    [scheduler_1.GraphTaskOp.DeserializeSourceNode]: [],
    [scheduler_1.GraphTaskOp.DiffRawToParsed]: [],
    [scheduler_1.GraphTaskOp.ComputeSourceMap]: [],
    [scheduler_1.GraphTaskOp.ReadSourceNode]: [],
    [scheduler_1.GraphTaskOp.WriteSourceNode]: [],
    [scheduler_1.GraphTaskOp.DeleteSourceNode]: [],
    [scheduler_1.GraphTaskOp.TransformParsed]: [],
    [scheduler_1.GraphTaskOp.MoveSourceNode]: [],
    [scheduler_1.GraphTaskOp.ResolveSourceNode]: [],
    [scheduler_1.GraphTaskOp.ValidateSourceNode]: [],
});
const findHandler = (handlers, node) => {
    return handlers.find(h => h.selector(node));
};
//# sourceMappingURL=scheduler.js.map