@jupyterlab/services/lib/kernel/future.js

"use strict";
// Copyright (c) Jupyter Development Team.
// Distributed under the terms of the Modified BSD License.
var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
    if (k2 === undefined) k2 = k;
    var desc = Object.getOwnPropertyDescriptor(m, k);
    if (!desc || ("get" in desc ? !m.__esModule : desc.writable || desc.configurable)) {
      desc = { enumerable: true, get: function() { return m[k]; } };
    }
    Object.defineProperty(o, k2, desc);
}) : (function(o, m, k, k2) {
    if (k2 === undefined) k2 = k;
    o[k2] = m[k];
}));
var __setModuleDefault = (this && this.__setModuleDefault) || (Object.create ? (function(o, v) {
    Object.defineProperty(o, "default", { enumerable: true, value: v });
}) : function(o, v) {
    o["default"] = v;
});
var __importStar = (this && this.__importStar) || function (mod) {
    if (mod && mod.__esModule) return mod;
    var result = {};
    if (mod != null) for (var k in mod) if (k !== "default" && Object.prototype.hasOwnProperty.call(mod, k)) __createBinding(result, mod, k);
    __setModuleDefault(result, mod);
    return result;
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.KernelShellFutureHandler = exports.KernelControlFutureHandler = exports.KernelFutureHandler = void 0;
const coreutils_1 = require("@lumino/coreutils");
const disposable_1 = require("@lumino/disposable");
const KernelMessage = __importStar(require("./messages"));
/**
 * Implementation of a kernel future.
 *
 * If a reply is expected, the Future is considered done when both a `reply`
 * message and an `idle` iopub status message have been received.  Otherwise, it
 * is considered done when the `idle` status is received.
 *
 */
class KernelFutureHandler extends disposable_1.DisposableDelegate {
    /**
     * Construct a new KernelFutureHandler.
     */
    constructor(cb, msg, expectReply, disposeOnDone, kernel) {
        super(cb);
        this._status = 0;
        this._stdin = Private.noOp;
        this._iopub = Private.noOp;
        this._reply = Private.noOp;
        this._done = new coreutils_1.PromiseDelegate();
        this._hooks = new Private.HookList();
        this._disposeOnDone = true;
        this._msg = msg;
        if (!expectReply) {
            this._setFlag(Private.KernelFutureFlag.GotReply);
        }
        this._disposeOnDone = disposeOnDone;
        this._kernel = kernel;
    }
    /**
     * Get the original outgoing message.
     */
    get msg() {
        return this._msg;
    }
    /**
     * A promise that resolves when the future is done.
     */
    get done() {
        return this._done.promise;
    }
    /**
     * Get the reply handler.
     */
    get onReply() {
        return this._reply;
    }
    /**
     * Set the reply handler.
     */
    set onReply(cb) {
        this._reply = cb;
    }
    /**
     * Get the iopub handler.
     */
    get onIOPub() {
        return this._iopub;
    }
    /**
     * Set the iopub handler.
     */
    set onIOPub(cb) {
        this._iopub = cb;
    }
    /**
     * Get the stdin handler.
     */
    get onStdin() {
        return this._stdin;
    }
    /**
     * Set the stdin handler.
     */
    set onStdin(cb) {
        this._stdin = cb;
    }
    /**
     * Register hook for IOPub messages.
     *
     * @param hook - The callback invoked for an IOPub message.
     *
     * #### Notes
     * The IOPub hook system allows you to preempt the handlers for IOPub
     * messages handled by the future.
     *
     * The most recently registered hook is run first. A hook can return a
     * boolean or a promise to a boolean, in which case all kernel message
     * processing pauses until the promise is fulfilled. If a hook return value
     * resolves to false, any later hooks will not run and the function will
     * return a promise resolving to false. If a hook throws an error, the error
     * is logged to the console and the next hook is run. If a hook is
     * registered during the hook processing, it will not run until the next
     * message. If a hook is removed during the hook processing, it will be
     * deactivated immediately.
     */
    registerMessageHook(hook) {
        if (this.isDisposed) {
            throw new Error('Kernel future is disposed');
        }
        this._hooks.add(hook);
    }
    /**
     * Remove a hook for IOPub messages.
     *
     * @param hook - The hook to remove.
     *
     * #### Notes
     * If a hook is removed during the hook processing, it will be deactivated immediately.
     */
    removeMessageHook(hook) {
        if (this.isDisposed) {
            return;
        }
        this._hooks.remove(hook);
    }
    /**
     * Send an `input_reply` message.
     */
    sendInputReply(content, parent_header) {
        this._kernel.sendInputReply(content, parent_header);
    }
    /**
     * Dispose and unregister the future.
     */
    dispose() {
        this._stdin = Private.noOp;
        this._iopub = Private.noOp;
        this._reply = Private.noOp;
        this._hooks = null;
        if (!this._testFlag(Private.KernelFutureFlag.IsDone)) {
            // TODO: Uncomment the following logging code, and check for any tests that trigger it.
            // let status = [];
            // if (!this._testFlag(Private.KernelFutureFlag.GotIdle)) {
            //   status.push('idle');
            // }
            // if (!this._testFlag(Private.KernelFutureFlag.GotReply)) {
            //   status.push('reply');
            // }
            // console.warn(
            //   `*************** DISPOSED BEFORE DONE: K${this._kernel.id.slice(
            //     0,
            //     6
            //   )} M${this._msg.header.msg_id.slice(0, 6)} missing ${status.join(' ')}`
            // );
            // Reject the `done` promise, but catch its error here in case no one else
            // is waiting for the promise to resolve. This prevents the error from
            // being displayed in the console, but does not prevent it from being
            // caught by a client who is waiting for it.
            // Note: any `.then` and `.finally` attached to the `done` promise
            // will cause the error to be thrown as uncaught anyways.
            this._done.promise.catch(() => {
                /* no-op */
            });
            this._done.reject(new Error(`Canceled future for ${this.msg.header.msg_type} message before replies were done`));
        }
        super.dispose();
    }
    /**
     * Handle an incoming kernel message.
     */
    async handleMsg(msg) {
        switch (msg.channel) {
            case 'control':
            case 'shell':
                if (msg.channel === this.msg.channel &&
                    msg.parent_header.msg_id === this.msg.header.msg_id) {
                    await this._handleReply(msg);
                }
                break;
            case 'stdin':
                await this._handleStdin(msg);
                break;
            case 'iopub':
                await this._handleIOPub(msg);
                break;
            default:
                break;
        }
    }
    async _handleReply(msg) {
        const reply = this._reply;
        if (reply) {
            // tslint:disable-next-line:await-promise
            await reply(msg);
        }
        this._replyMsg = msg;
        this._setFlag(Private.KernelFutureFlag.GotReply);
        if (this._testFlag(Private.KernelFutureFlag.GotIdle)) {
            this._handleDone();
        }
    }
    async _handleStdin(msg) {
        this._kernel.hasPendingInput = true;
        const stdin = this._stdin;
        if (stdin) {
            // tslint:disable-next-line:await-promise
            await stdin(msg);
        }
    }
    async _handleIOPub(msg) {
        const process = await this._hooks.process(msg);
        const iopub = this._iopub;
        if (process && iopub) {
            // tslint:disable-next-line:await-promise
            await iopub(msg);
        }
        if (KernelMessage.isStatusMsg(msg) &&
            msg.content.execution_state === 'idle') {
            this._setFlag(Private.KernelFutureFlag.GotIdle);
            if (this._testFlag(Private.KernelFutureFlag.GotReply)) {
                this._handleDone();
            }
        }
    }
    _handleDone() {
        if (this._testFlag(Private.KernelFutureFlag.IsDone)) {
            return;
        }
        this._setFlag(Private.KernelFutureFlag.IsDone);
        this._done.resolve(this._replyMsg);
        if (this._disposeOnDone) {
            this.dispose();
        }
    }
    /**
     * Test whether the given future flag is set.
     */
    _testFlag(flag) {
        // tslint:disable-next-line
        return (this._status & flag) !== 0;
    }
    /**
     * Set the given future flag.
     */
    _setFlag(flag) {
        // tslint:disable-next-line
        this._status |= flag;
    }
}
exports.KernelFutureHandler = KernelFutureHandler;
class KernelControlFutureHandler extends KernelFutureHandler {
}
exports.KernelControlFutureHandler = KernelControlFutureHandler;
class KernelShellFutureHandler extends KernelFutureHandler {
}
exports.KernelShellFutureHandler = KernelShellFutureHandler;
var Private;
(function (Private) {
    /**
     * A no-op function.
     */
    Private.noOp = () => {
        /* no-op */
    };
    /**
     * Defer a computation.
     *
     * #### NOTES
     * We can't just use requestAnimationFrame since it is not available in node.
     * This implementation is from Phosphor:
     * https://github.com/phosphorjs/phosphor/blob/e88e4321289bb1198f3098e7bda40736501f2ed8/tests/test-messaging/src/index.spec.ts#L63
     */
    const defer = (() => {
        const ok = typeof requestAnimationFrame === 'function';
        return ok ? requestAnimationFrame : setImmediate;
    })();
    class HookList {
        constructor() {
            this._hooks = [];
        }
        /**
         * Register a hook.
         *
         * @param hook - The callback to register.
         */
        add(hook) {
            this.remove(hook);
            this._hooks.push(hook);
        }
        /**
         * Remove a hook, if it exists in the hook list.
         *
         * @param hook - The callback to remove.
         */
        remove(hook) {
            const index = this._hooks.indexOf(hook);
            if (index >= 0) {
                this._hooks[index] = null;
                this._scheduleCompact();
            }
        }
        /**
         * Process a message through the hooks.
         *
         * @returns a promise resolving to false if any hook resolved as false,
         * otherwise true
         *
         * #### Notes
         * The most recently registered hook is run first. A hook can return a
         * boolean or a promise to a boolean, in which case processing pauses until
         * the promise is fulfilled. If a hook return value resolves to false, any
         * later hooks will not run and the function will return a promise resolving
         * to false. If a hook throws an error, the error is logged to the console
         * and the next hook is run. If a hook is registered during the hook
         * processing, it will not run until the next message. If a hook is removed
         * during the hook processing, it will be deactivated immediately.
         */
        async process(msg) {
            // Wait until we can start a new process run.
            await this._processing;
            // Start the next process run.
            const processing = new coreutils_1.PromiseDelegate();
            this._processing = processing.promise;
            let continueHandling;
            // Call the end hook (most recently-added) first. Starting at the end also
            // guarantees that hooks added during the processing will not be run in
            // this process run.
            for (let i = this._hooks.length - 1; i >= 0; i--) {
                const hook = this._hooks[i];
                // If the hook has been removed, continue to the next one.
                if (hook === null) {
                    continue;
                }
                // Execute the hook and log any errors.
                try {
                    // tslint:disable-next-line:await-promise
                    continueHandling = await hook(msg);
                }
                catch (err) {
                    continueHandling = true;
                    console.error(err);
                }
                // If the hook resolved to false, stop processing and return.
                if (continueHandling === false) {
                    processing.resolve(undefined);
                    return false;
                }
            }
            // All hooks returned true (or errored out), so return true.
            processing.resolve(undefined);
            return true;
        }
        /**
         * Schedule a cleanup of the list, removing any hooks that have been nulled out.
         */
        _scheduleCompact() {
            if (!this._compactScheduled) {
                this._compactScheduled = true;
                // Schedule a compaction in between processing runs. We do the
                // scheduling in an animation frame to rate-limit our compactions. If we
                // need to compact more frequently, we can change this to directly
                // schedule the compaction.
                defer(() => {
                    this._processing = this._processing.then(() => {
                        this._compactScheduled = false;
                        this._compact();
                    });
                });
            }
        }
        /**
         * Compact the list, removing any nulls.
         */
        _compact() {
            let numNulls = 0;
            for (let i = 0, len = this._hooks.length; i < len; i++) {
                const hook = this._hooks[i];
                if (this._hooks[i] === null) {
                    numNulls++;
                }
                else {
                    this._hooks[i - numNulls] = hook;
                }
            }
            this._hooks.length -= numNulls;
        }
    }
    Private.HookList = HookList;
    /**
     * Bit flags for the kernel future state.
     */
    let KernelFutureFlag;
    (function (KernelFutureFlag) {
        KernelFutureFlag[KernelFutureFlag["GotReply"] = 1] = "GotReply";
        KernelFutureFlag[KernelFutureFlag["GotIdle"] = 2] = "GotIdle";
        KernelFutureFlag[KernelFutureFlag["IsDone"] = 4] = "IsDone";
        KernelFutureFlag[KernelFutureFlag["DisposeOnDone"] = 8] = "DisposeOnDone";
    })(KernelFutureFlag = Private.KernelFutureFlag || (Private.KernelFutureFlag = {}));
})(Private || (Private = {}));
//# sourceMappingURL=future.js.map