"use strict";
// Copyright (c) Jupyter Development Team.
// Distributed under the terms of the Modified BSD License.
var __await = (this && this.__await) || function (v) { return this instanceof __await ? (this.v = v, this) : new __await(v); }
var __asyncGenerator = (this && this.__asyncGenerator) || function (thisArg, _arguments, generator) {
    if (!Symbol.asyncIterator) throw new TypeError("Symbol.asyncIterator is not defined.");
    var g = generator.apply(thisArg, _arguments || []), i, q = [];
    return i = {}, verb("next"), verb("throw"), verb("return"), i[Symbol.asyncIterator] = function () { return this; }, i;
    function verb(n) { if (g[n]) i[n] = function (v) { return new Promise(function (a, b) { q.push([n, v, a, b]) > 1 || resume(n, v); }); }; }
    function resume(n, v) { try { step(g[n](v)); } catch (e) { settle(q[0][3], e); } }
    function step(r) { r.value instanceof __await ? Promise.resolve(r.value.v).then(fulfill, reject) : settle(q[0][2], r); }
    function fulfill(value) { resume("next", value); }
    function reject(value) { resume("throw", value); }
    function settle(f, v) { if (f(v), q.shift(), q.length) resume(q[0][0], q[0][1]); }
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.EventManager = void 0;
const coreutils_1 = require("@jupyterlab/coreutils");
const coreutils_2 = require("@lumino/coreutils");
const polling_1 = require("@lumino/polling");
const signaling_1 = require("@lumino/signaling");
const serverconnection_1 = require("../serverconnection");
/**
 * The url for the jupyter-server events service.
 */
const SERVICE_EVENTS_URL = 'api/events';
/**
 * The events API service manager.
 */
class EventManager {
    /**
     * Create a new event manager.
     */
    constructor(options = {}) {
        var _a;
        // TODO: Remove this check for the `jupyter_server` version.
        // It is only necessary in JupyterLab < 4.
        this._isDisabled = 2 > coreutils_1.PageConfig.getNotebookVersion()[0];
        this._isDisposed = false;
        this._socket = null;
        this.serverSettings = (_a = options.serverSettings) !== null && _a !== void 0 ? _a : serverconnection_1.ServerConnection.makeSettings();
        this._poll = new polling_1.Poll({ factory: () => this._subscribe() });
        this._stream = new Private.Stream(this);
        // TODO: Remove this logic in JupyterLab 4
        if (this._isDisabled) {
            this._stream.stop();
        }
        else {
            void this._poll.start();
        }
    }
    /**
     * Whether the event manager is disposed.
     */
    get isDisposed() {
        return this._isDisposed;
    }
    /**
     * An event stream that emits and yields each new event.
     */
    get stream() {
        return this._stream;
    }
    /**
     * Dispose the event manager.
     */
    dispose() {
        if (this.isDisposed) {
            return;
        }
        this._isDisposed = true;
        // Clean up poll.
        this._poll.dispose();
        // Clean up socket.
        const socket = this._socket;
        if (socket) {
            this._socket = null;
            socket.onopen = () => undefined;
            socket.onerror = () => undefined;
            socket.onmessage = () => undefined;
            socket.onclose = () => undefined;
            socket.close();
        }
        // Clean up stream.
        signaling_1.Signal.clearData(this);
        this._stream.stop();
    }
    /**
     * Post an event request to be emitted by the event bus.
     */
    async emit(event) {
        const { serverSettings } = this;
        const { baseUrl, token } = serverSettings;
        const { makeRequest, ResponseError } = serverconnection_1.ServerConnection;
        const url = coreutils_1.URLExt.join(baseUrl, SERVICE_EVENTS_URL) +
            (token ? `?token=${token}` : '');
        const init = { body: JSON.stringify(event), method: 'POST' };
        const response = await makeRequest(url, init, serverSettings);
        if (response.status !== 204) {
            throw new ResponseError(response);
        }
    }
    /**
     * Subscribe to event bus emissions.
     */
    _subscribe() {
        return new Promise((_, reject) => {
            if (this.isDisposed || this._isDisabled) {
                return;
            }
            const { token, WebSocket, wsUrl } = this.serverSettings;
            const url = coreutils_1.URLExt.join(wsUrl, SERVICE_EVENTS_URL, 'subscribe') +
                (token ? `?token=${encodeURIComponent(token)}` : '');
            const socket = (this._socket = new WebSocket(url));
            const stream = this._stream;
            // Cause the poll to tick a rejection and back off if the socket closes.
            socket.onclose = () => reject(new Error('EventManager socket closed'));
            socket.onmessage = msg => msg.data && stream.emit(JSON.parse(msg.data));
        });
    }
}
exports.EventManager = EventManager;
/**
 * A namespace for private module data.
 */
var Private;
(function (Private) {
    /**
     * A stream with the characteristics of a signal and an async iterable.
     */
    class Stream extends signaling_1.Signal {
        constructor() {
            super(...arguments);
            this._pending = new coreutils_2.PromiseDelegate();
        }
        /**
         * Return an async iterator that yields every emission.
         */
        [Symbol.asyncIterator]() {
            return __asyncGenerator(this, arguments, function* _a() {
                let pending = this._pending;
                while (true) {
                    try {
                        const { args, next } = yield __await(pending.promise);
                        pending = next;
                        yield yield __await(args);
                    }
                    catch (_) {
                        return yield __await(void 0); // Any promise rejection stops the iterator.
                    }
                }
            });
        }
        /**
         * Emit the signal, invoke the connected slots, and yield the emission.
         *
         * @param args - The args to pass to the connected slots.
         */
        emit(args) {
            const pending = this._pending;
            this._pending = new coreutils_2.PromiseDelegate();
            pending.resolve({ args, next: this._pending });
            super.emit(args);
        }
        /**
         * Stop the stream's async iteration.
         */
        stop() {
            this._pending.promise.catch(() => undefined);
            this._pending.reject('stop');
        }
    }
    Private.Stream = Stream;
})(Private || (Private = {}));
//# sourceMappingURL=index.js.map