alar/dist/rpc/client.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const net = require("net");
const sequid_1 = require("sequid");
const backoff_1 = require("backoff");
const structured_clone_1 = require("@hyurl/structured-clone");
const isSocketResetError = require("is-socket-reset-error");
const thenable_generator_1 = require("thenable-generator");
const channel_1 = require("./channel");
const util_1 = require("../util");
const last = require("lodash/last");
const isOwnKey_1 = require("@hyurl/utils/isOwnKey");
class RpcClient extends channel_1.RpcChannel {
    constructor(options, host) {
        super(options, host);
        this.state = "initiated";
        this.socket = null;
        this.registry = util_1.dict();
        this.taskId = sequid_1.default(0, true);
        this.tasks = new Map();
        this.topics = new Map();
        this.finishConnect = null;
        this.rejectConnect = null;
        this.lastActiveTime = Date.now();
        this.selfDestruction = null;
        this.pingTimer = null;
        this.reconnect = null;
        this.id = this.id || Math.random().toString(16).slice(2);
        this.timeout = this.timeout || 5000;
        this.pingInterval = this.pingInterval || 5000;
        this.serverId = this.serverId || this.dsn;
    }
    get connecting() {
        return this.state === "connecting";
    }
    get connected() {
        return this.state === "connected";
    }
    get closed() {
        return this.state === "closed";
    }
    ;
    open() {
        return new Promise((resolve, reject) => {
            let { serverId } = this;
            if (this.socket && this.socket.connecting) {
                throw new Error(`Channel to ${serverId} is already open`);
            }
            else if (this.closed) {
                throw new Error(`Cannot reconnect to ${serverId} after closing the channel`);
            }
            this.state === "connecting";
            this.finishConnect = () => {
                this.state = "connected";
                this.resume();
                if (!this.pingTimer && !this.reconnect) {
                    this.setPingAndReconnectTimer(serverId);
                }
                resolve(this);
            };
            this.rejectConnect = () => {
                reject(new Error(`Unable to connect ${serverId}`));
            };
            let timer = setTimeout(this.rejectConnect, this.timeout);
            let connectListener = () => {
                clearTimeout(timer);
                this.socket.removeListener("error", errorListener);
                this.prepareChannel();
                if (this.secret) {
                    this.socket.write(this.secret, () => {
                        this.send(channel_1.RpcEvents.HANDSHAKE, this.id);
                    });
                }
                else {
                    this.send(channel_1.RpcEvents.HANDSHAKE, this.id);
                }
            };
            let errorListener = (err) => {
                clearTimeout(timer);
                this.socket.removeListener("connect", connectListener);
                reject(err);
            };
            if (this.path) {
                this.socket = net.createConnection(this.path, connectListener);
            }
            else {
                this.socket = net.createConnection(this.port, this.host, connectListener);
            }
            this.socket = this.bsp.wrap(this.socket);
            this.socket.once("error", errorListener);
        });
    }
    close() {
        return new Promise(resolve => {
            clearInterval(this.pingTimer);
            clearTimeout(this.selfDestruction);
            this.state = "closed";
            this.reconnect.reset();
            this.pause();
            if (this.socket) {
                this.socket.unref();
                this.socket.end();
                resolve(this);
            }
            else {
                resolve(this);
            }
        });
    }
    register(mod) {
        if (!this.registry[mod.name]) {
            this.registry[mod.name] = mod;
            let singletons = mod["remoteSingletons"];
            singletons[this.serverId] = util_1.createRemoteInstance(mod, (prop) => this.createFunction(mod, prop));
            singletons[this.serverId][util_1.readyState] = this.connected ? 2 : 0;
        }
        return this;
    }
    setPingAndReconnectTimer(serverId) {
        this.pingTimer = setInterval(() => {
            let duration = Date.now() - this.lastActiveTime;
            if (duration >= this.pingInterval) {
                this.selfDestruction = setTimeout(this.socket.destroy.bind(this.socket), this.timeout);
                this.send(channel_1.RpcEvents.PING, this.id);
            }
        }, 5000);
        this.reconnect = backoff_1.exponential({
            maxDelay: 5000
        }).on("ready", async (num) => {
            try {
                await this.open();
            }
            catch (e) { }
            if (this.connected) {
                this.reconnect.reset();
                this.resume();
            }
            else if (num === 365) {
                await this.close();
                console.error(`Connection to ${serverId} lost permanently`);
            }
            else {
                this.reconnect.backoff();
            }
        });
    }
    flushReadyState(state) {
        for (let name in this.registry) {
            let mod = this.registry[name];
            let singletons = mod["remoteSingletons"];
            singletons[this.serverId][util_1.readyState] = state;
        }
    }
    pause() {
        this.flushReadyState(0);
    }
    resume() {
        this.flushReadyState(2);
    }
    subscribe(topic, handle) {
        let handlers = this.topics.get(topic);
        handlers || this.topics.set(topic, handlers = new Set());
        handlers.add(handle);
        return this;
    }
    unsubscribe(topic, handle) {
        if (!handle) {
            return this.topics.delete(topic);
        }
        else {
            let handlers = this.topics.get(topic);
            if (handlers) {
                return handlers.delete(handle);
            }
            else {
                return false;
            }
        }
    }
    send(...data) {
        if (this.socket && !this.socket.destroyed && this.socket.writable) {
            if (last(data) === undefined) {
                data.pop();
            }
            this.socket.write(data);
        }
    }
    createFunction(mod, method) {
        let self = this;
        return function (...args) {
            let root = mod[util_1.proxyRoot];
            if (root && root["server"] && root["server"].id === self.serverId) {
                let ins = mod.instance();
                if (isOwnKey_1.default(ins, util_1.readyState) && ins[util_1.readyState] !== 2 &&
                    !mod.fallbackToLocal()) {
                    util_1.throwUnavailableError(mod.name);
                }
                else {
                    return new thenable_generator_1.ThenableAsyncGenerator(ins[method](...args));
                }
            }
            if (!self.connected) {
                if (mod.fallbackToLocal()) {
                    return new thenable_generator_1.ThenableAsyncGenerator(mod.instance()[method](...args));
                }
                else {
                    util_1.throwUnavailableError(mod.name);
                }
            }
            return new thenable_generator_1.ThenableAsyncGenerator(new ThenableIteratorProxy(self, mod.name, method, ...args));
        };
    }
    prepareChannel() {
        this.socket.on("error", err => {
            if (!isSocketResetError(err)) {
                if (this.errorHandler) {
                    this.errorHandler(err);
                }
                else {
                    console.error(err);
                }
            }
        }).on("close", () => {
            if (this.connecting) {
                this.rejectConnect && this.rejectConnect();
            }
            else if (!this.closed) {
                this.state = "connecting";
                this.pause();
                this.reconnect && this.reconnect.backoff();
            }
        }).on("data", async (msg) => {
            this.lastActiveTime = Date.now();
            if (this.selfDestruction) {
                clearTimeout(this.selfDestruction);
                this.selfDestruction = null;
            }
            if (this.codec === "BSON") {
                msg = Array.from(Object.assign(msg, {
                    length: Object.keys(msg).length
                }));
            }
            let [event, taskId, data] = msg;
            let task;
            switch (event) {
                case channel_1.RpcEvents.CONNECT: {
                    if (data !== this.serverId) {
                        for (let name in this.registry) {
                            let mod = this.registry[name];
                            let singletons = mod["remoteSingletons"];
                            if (singletons[this.serverId]) {
                                singletons[data] = singletons[this.serverId];
                                delete singletons[this.serverId];
                            }
                        }
                    }
                    this.serverId = data;
                    this.finishConnect();
                    break;
                }
                case channel_1.RpcEvents.BROADCAST: {
                    let handlers = this.topics.get(taskId);
                    if (handlers) {
                        handlers.forEach(async (handle) => {
                            try {
                                await handle(data);
                            }
                            catch (err) {
                                this.errorHandler && this.errorHandler(err);
                            }
                        });
                    }
                    break;
                }
                case channel_1.RpcEvents.INVOKE:
                case channel_1.RpcEvents.YIELD:
                case channel_1.RpcEvents.RETURN: {
                    if (task = this.tasks.get(taskId)) {
                        task.resolve(data);
                    }
                    break;
                }
                case channel_1.RpcEvents.THROW: {
                    if (task = this.tasks.get(taskId)) {
                        (this.codec !== "CLONE") && (data = structured_clone_1.decompose(data));
                        task.reject(data);
                    }
                    break;
                }
            }
        });
        return this;
    }
}
exports.RpcClient = RpcClient;
class ThenableIteratorProxy {
    constructor(client, modName, method, ...args) {
        this.client = client;
        this.modName = modName;
        this.method = method;
        this.taskId = this.client["taskId"].next().value;
        this.queue = [];
        this.status = "uninitiated";
        this.args = args;
        this.result = this.invokeTask(channel_1.RpcEvents.INVOKE, ...this.args);
    }
    next(value) {
        return this.invokeTask(channel_1.RpcEvents.YIELD, value);
    }
    return(value) {
        return this.invokeTask(channel_1.RpcEvents.RETURN, value);
    }
    throw(err) {
        return this.invokeTask(channel_1.RpcEvents.THROW, err);
    }
    then(resolver, rejecter) {
        return Promise.resolve(this.result).then((res) => {
            this.status = "closed";
            this.result = res;
            this.client["tasks"].delete(this.taskId);
            return res;
        }).then(resolver, rejecter);
    }
    close() {
        this.status = "closed";
        for (let task of this.queue) {
            switch (task.event) {
                case channel_1.RpcEvents.INVOKE:
                    task.resolve(void 0);
                    break;
                case channel_1.RpcEvents.YIELD:
                    task.resolve({ value: void 0, done: true });
                    break;
                case channel_1.RpcEvents.RETURN:
                    task.resolve({ value: task.data, done: true });
                    break;
                case channel_1.RpcEvents.THROW:
                    task.reject(task.data);
                    break;
            }
        }
        this.queue = [];
    }
    createTimeout() {
        return setTimeout(() => {
            if (this.queue.length > 0) {
                let task = this.queue.shift();
                let callee = `${this.modName}(<route>).${this.method}()`;
                let duration = util_1.humanizeDuration(this.client.timeout);
                task.reject(new Error(`${callee} timeout after ${duration}`));
            }
            this.close();
        }, this.client.timeout);
    }
    prepareTask(event, data) {
        let task = this.client["tasks"].get(this.taskId);
        if (!task) {
            this.client["tasks"].set(this.taskId, task = {
                resolve: (data) => {
                    if (this.status === "suspended") {
                        if (this.queue.length > 0) {
                            this.queue.shift().resolve(data);
                        }
                    }
                },
                reject: (err) => {
                    if (this.status === "suspended") {
                        if (this.queue.length > 0) {
                            this.queue.shift().reject(err);
                        }
                        this.close();
                    }
                }
            });
        }
        return new Promise((resolve, reject) => {
            let timer = this.createTimeout();
            this.queue.push({
                event,
                data,
                resolve: (data) => {
                    clearTimeout(timer);
                    resolve(data);
                },
                reject: (err) => {
                    clearTimeout(timer);
                    reject(err);
                }
            });
        });
    }
    async invokeTask(event, ...args) {
        if (this.status === "closed") {
            switch (event) {
                case channel_1.RpcEvents.INVOKE:
                    return Promise.resolve(this.result);
                case channel_1.RpcEvents.YIELD:
                    return Promise.resolve({ value: undefined, done: true });
                case channel_1.RpcEvents.RETURN:
                    return Promise.resolve({ value: args[0], done: true });
                case channel_1.RpcEvents.THROW:
                    return Promise.reject(args[0]);
            }
        }
        else {
            if (this.status === "uninitiated" && event !== channel_1.RpcEvents.INVOKE) {
                this.client["send"](event, this.taskId, this.modName, this.method, [...this.args], ...args);
            }
            else {
                this.client["send"](event, this.taskId, this.modName, this.method, ...args);
            }
            this.status = "suspended";
            try {
                let res = await this.prepareTask(event, args[0]);
                if (event !== channel_1.RpcEvents.INVOKE) {
                    ("value" in res) || (res.value = void 0);
                    if (res.done) {
                        this.status = "closed";
                        this.result = res.value;
                        this.client["tasks"].delete(this.taskId);
                    }
                }
                return res;
            }
            catch (err) {
                this.status = "closed";
                this.client["tasks"].delete(this.taskId);
                throw err;
            }
        }
    }
}
//# sourceMappingURL=client.js.map