alar/dist/rpc/server.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const net = require("net");
const path = require("path");
const fs = require("fs-extra");
const structured_clone_1 = require("@hyurl/structured-clone");
const advanced_collections_1 = require("advanced-collections");
const check_iterable_1 = require("check-iterable");
const isSocketResetError = require("is-socket-reset-error");
const channel_1 = require("./channel");
const util_1 = require("../util");
const values = require("lodash/values");
const isOwnKey_1 = require("@hyurl/utils/isOwnKey");
const authorized = Symbol("authorized");
class RpcServer extends channel_1.RpcChannel {
    constructor(options, host) {
        super(options, host);
        this.server = null;
        this.registry = util_1.dict();
        this.clients = new advanced_collections_1.BiMap();
        this.suspendedTasks = new Map();
        this.proxyRoot = null;
        this.enableLifeCycle = false;
        this.id = this.id || this.dsn;
    }
    async open(enableLifeCycle = true) {
        if (enableLifeCycle) {
            this.enableLifeCycle = true;
            for (let mod of values(this.registry)) {
                await util_1.tryLifeCycleFunction(mod, "init", this.errorHandler);
            }
        }
        if (this.path) {
            await fs.ensureDir(path.dirname(this.path));
            if (await fs.pathExists(this.path)) {
                await fs.unlink(this.path);
            }
        }
        return new Promise((resolve, reject) => {
            let server = this.server = net.createServer();
            let listener = () => {
                server.on("error", (err) => {
                    if (this.errorHandler) {
                        this.errorHandler.call(this, err);
                    }
                    else {
                        console.error(err);
                        process.exit(1);
                    }
                });
                resolve(this);
            };
            server.once("error", reject)
                .on("connection", this.handleConnection.bind(this));
            if (this.path) {
                server.listen(util_1.absPath(this.path, true), listener);
            }
            else if (this.host) {
                server.listen(this.port, this.host, listener);
            }
            else {
                server.listen(this.port, listener);
            }
        });
    }
    async close() {
        await new Promise(resolve => {
            if (this.server) {
                let timer = setTimeout(() => {
                    for (let [, socket] of this.clients) {
                        socket.destroy();
                    }
                }, 1000);
                this.server.unref();
                this.server.close(() => {
                    clearTimeout(timer);
                    resolve();
                });
            }
            else {
                resolve();
            }
        });
        if (this.enableLifeCycle) {
            await Promise.all(values(this.registry).map(mod => {
                return util_1.tryLifeCycleFunction(mod, "destroy").catch(err => {
                    this.errorHandler && this.errorHandler(err);
                });
            }));
        }
        if (this.proxyRoot) {
            this.proxyRoot["server"] = null;
            this.proxyRoot = null;
        }
        return this;
    }
    register(mod) {
        this.registry[mod.name] = mod;
        return this;
    }
    publish(topic, data, clients) {
        let sent = false;
        let socket;
        let targets = clients || this.clients.keys();
        for (let id of targets) {
            if (socket = this.clients.get(id)) {
                this.dispatch(socket, channel_1.RpcEvents.BROADCAST, topic, data);
                sent = true;
            }
        }
        return sent;
    }
    getClients() {
        let clients = [];
        for (let [id] of this.clients) {
            clients.push(id);
        }
        return clients;
    }
    dispatch(socket, event, ...data) {
        if (!socket.destroyed && socket.writable) {
            if (event === channel_1.RpcEvents.THROW) {
                data = structured_clone_1.compose(data);
            }
            socket.write([event, ...data]);
        }
    }
    handleConnection(socket) {
        let addr = `${socket.remoteAddress || ""}:${socket.remotePort || ""}`;
        let destroyWithHandshakeError = () => {
            socket.destroy(new Error(`Handshake required (client: ${addr})`));
        };
        let autoDestroy = setTimeout(destroyWithHandshakeError, 5000);
        this.bsp.wrap(socket).on("error", err => {
            if (!isSocketResetError(err) && this.errorHandler) {
                this.errorHandler(err);
            }
        }).on("close", () => {
            let tasks = this.suspendedTasks.get(socket);
            if (tasks) {
                this.suspendedTasks.delete(socket);
                this.clients.deleteValue(socket);
                for (let task of tasks.values()) {
                    task.return();
                }
            }
        }).on("data", async (msg) => {
            if (this.secret && !socket[authorized]) {
                if (this.secret === msg) {
                    socket[authorized] = true;
                    return;
                }
                else {
                    return socket.destroy(new Error(`Connection unauthorized (client: ${addr})`));
                }
            }
            if (this.codec === "BSON" && typeof msg === "object") {
                msg = Array.from(Object.assign(msg, {
                    length: Object.keys(msg).length
                }));
            }
            if (!Array.isArray(msg))
                return;
            let [event, taskId, modName, method, ...args] = msg;
            if (!this.suspendedTasks.has(socket) &&
                event !== channel_1.RpcEvents.HANDSHAKE) {
                return destroyWithHandshakeError();
            }
            switch (event) {
                case channel_1.RpcEvents.HANDSHAKE: {
                    let clientId = String(taskId);
                    clearTimeout(autoDestroy);
                    this.clients.set(clientId, socket);
                    this.suspendedTasks.set(socket, new Map());
                    this.dispatch(socket, channel_1.RpcEvents.CONNECT, clientId, this.id);
                    break;
                }
                case channel_1.RpcEvents.PING: {
                    this.dispatch(socket, channel_1.RpcEvents.PONG);
                    break;
                }
                case channel_1.RpcEvents.INVOKE: {
                    let data;
                    let tasks = this.suspendedTasks.get(socket);
                    try {
                        let ins = this.registry[modName]();
                        if (isOwnKey_1.default(ins, util_1.readyState) && ins[util_1.readyState] !== 2) {
                            util_1.throwUnavailableError(modName);
                        }
                        let task = ins[method].apply(ins, args);
                        if (task && check_iterable_1.isIteratorLike(task)) {
                            tasks.set(taskId, task);
                            event = channel_1.RpcEvents.INVOKE;
                        }
                        else {
                            data = await task;
                            event = channel_1.RpcEvents.RETURN;
                        }
                    }
                    catch (err) {
                        event = channel_1.RpcEvents.THROW;
                        data = err;
                    }
                    this.dispatch(socket, event, taskId, data);
                    break;
                }
                case channel_1.RpcEvents.YIELD:
                case channel_1.RpcEvents.RETURN:
                case channel_1.RpcEvents.THROW: {
                    let data, input;
                    let tasks = this.suspendedTasks.get(socket);
                    let task = tasks.get(taskId);
                    try {
                        if (!task) {
                            let callee = `${modName}(<route>).${method}()`;
                            throw new ReferenceError(`${callee} failed (${taskId})`);
                        }
                        else {
                            input = args[0];
                        }
                        if (event === channel_1.RpcEvents.YIELD) {
                            data = await task.next(input);
                        }
                        else if (event === channel_1.RpcEvents.RETURN) {
                            data = await task.return(input);
                        }
                        else {
                            await task.throw(input);
                        }
                        data.done && tasks.delete(taskId);
                    }
                    catch (err) {
                        event = channel_1.RpcEvents.THROW;
                        data = err;
                        task && tasks.delete(taskId);
                    }
                    this.dispatch(socket, event, taskId, data);
                    break;
                }
            }
        });
    }
}
exports.RpcServer = RpcServer;
//# sourceMappingURL=server.js.map