server-libs/src/TcpNet.js

"use strict";
var __importStar = (this && this.__importStar) || function (mod) {
    if (mod && mod.__esModule) return mod;
    var result = {};
    if (mod != null) for (var k in mod) if (Object.hasOwnProperty.call(mod, k)) result[k] = mod[k];
    result["default"] = mod;
    return result;
};
Object.defineProperty(exports, "__esModule", { value: true });
const net = __importStar(require("net"));
const events_1 = require("events");
const timers_1 = require("timers");
const debug_1 = require("debug");
const version_1 = require("./version");
let Debug = debug_1.debug('server-libs');
class TcpUnit extends events_1.EventEmitter {
    constructor(sockadd, sock, host = '', port = 0, serverId = '') {
        super();
        this.sockaddr = '';
        this.host = '';
        this.port = 0;
        this.retryCount = 0;
        this.serverId = '';
        this.recv_pool = [];
        this.recv_len = 0;
        // 这里host 是 ::ffff: 开头的要剔除掉
        this.sockaddr = sockadd;
        this.sock = sock;
        this.host = host.replace('::ffff:', '');
        this.port = port;
        this.serverId = serverId;
        if (version_1.V_UnderV8 && this.off == undefined) {
            this.off = this.removeListener;
        }
    }
    on(event, listener) {
        super.on(event, listener);
        return this;
    }
    reciveData(data) {
        this.recv_pool.push(data);
        this.recv_len += data.length;
        this.doParse();
    }
    resetData(data) {
        this.recv_pool.unshift(data);
        this.recv_len += data.length;
    }
    process(cmd, info) {
        switch (cmd) {
            case 0:
            case 1:
                Debug(info.toString());
                break;
        }
        // 这里如果存在主节点，那么提交给主节点处理，也可以自己处理
        if (!this.serverId || TcpServer.isDispersedMode(this.serverId)) {
            this.emit('data', cmd, info);
        }
        else {
            TcpServer.serPools[this.serverId].emit('data', this.sockaddr, cmd, info);
        }
    }
    doParse() {
        let pool_buff;
        let currlen = 0;
        if (this.recv_pool.length == 1) {
            pool_buff = this.recv_pool.shift();
        }
        else if (this.recv_pool.length > 1) {
            pool_buff = Buffer.alloc(this.recv_len);
            while (this.recv_pool.length) {
                let cache_buff = this.recv_pool.shift();
                if (!cache_buff)
                    break;
                cache_buff.copy(pool_buff, currlen, 0, cache_buff.length);
                currlen += cache_buff.length;
            }
        }
        else {
            return;
        }
        if (pool_buff.length < 2) {
            // 数据不足，需要放到下一个里面合成一个数据块
            this.recv_pool.unshift(pool_buff);
        }
        else {
            let len = pool_buff.readInt16BE(0);
            if (pool_buff.length < len) {
                // 数据不足，需要放到下一个里面合成一个数据块
                this.recv_pool.unshift(pool_buff);
            }
            else {
                // 处理数据
                this.process(pool_buff.readInt16BE(2), pool_buff.slice(4, len));
                if (len < pool_buff.length) {
                    // 数据不足，需要放到下一个里面合成一个数据块
                    this.recv_pool.unshift(pool_buff.slice(len, pool_buff.length));
                }
                // 再处理一下
                this.doParse();
            }
        }
    }
    sendMsg(cmd, msg) {
        if (typeof msg == 'string') {
            msg = Buffer.from(msg);
        }
        let datalen = msg.length + 4;
        let dataBuff = Buffer.alloc(datalen);
        dataBuff.writeInt16BE(datalen, 0);
        dataBuff.writeInt16BE(cmd, 2);
        msg.copy(dataBuff, 4, 0, msg.length);
        return this.sock ? this.sock.write(dataBuff) : false;
    }
    close() {
        this.sock && this.sock.end();
        this.emit('close', this.sockaddr, false);
    }
}
exports.TcpUnit = TcpUnit;
/**tcp 连接的管理，负责tcp连接的响应和处理 */
class TcpServer extends events_1.EventEmitter {
    constructor(port) {
        super();
        this.netPools = {};
        this.isDispersedMode = true;
        this.sockaddr = 'localhost:' + port;
        let tcp_server = net.createServer();
        tcp_server.listen(port);
        tcp_server.on("connection", this.onconnection.bind(this));
        tcp_server.on('error', (error) => { this.emit('error', error); });
        TcpServer.serPools[this.sockaddr] = this;
    }
    // 创建一个服务器
    static create(port) {
        return new TcpServer(port);
    }
    // 获取服务器节点
    static getServer(sockaddr) {
        return this.serPools[sockaddr];
    }
    static isDispersedMode(sockaddr) {
        if (this.serPools.hasOwnProperty(sockaddr)) {
            return this.serPools[sockaddr].isDispersedMode;
        }
        return true;
    }
    // 获取链接节点
    getUnit(sockaddr) {
        return this.netPools[sockaddr];
    }
    on(event, listener) {
        if (event == 'data')
            this.isDispersedMode = false;
        super.on(event, listener);
        return this;
    }
    onconnection(sock) {
        let sockaddr = sock.remoteAddress + ':' + sock.remotePort;
        sock.on('data', this.onData.bind(this, sockaddr));
        sock.on('close', this.onClose.bind(this, sockaddr));
        sock.on('error', this.onError.bind(this, sockaddr));
        this.netPools[sockaddr] = new TcpUnit(sockaddr, sock, sock.remoteAddress, sock.remotePort, this.sockaddr);
        this.emit("connection", sockaddr);
        // this.pushMsg(sockadd, 1, "your host is" + sock.remoteAddress);
        // this.pushMsg(sockadd, 2, "your port is" + sock.remotePort);
        // 这里发起一个connect 测试一下conncet转listen的成果
    }
    pushMsg(sockaddr, cmd, msg) {
        let sockUnit = this.netPools[sockaddr];
        if (sockUnit) {
            return sockUnit.sendMsg(cmd, msg);
        }
        return false;
    }
    onData(sockaddr, data) {
        let sockUnit = this.netPools[sockaddr];
        if (sockUnit) {
            sockUnit.reciveData(data);
        }
    }
    onClose(sockaddr) {
        if (this.netPools.hasOwnProperty(sockaddr)) {
            delete this.netPools[sockaddr];
            this.emit("close", sockaddr);
        }
    }
    onError(sockaddr) {
        if (this.netPools.hasOwnProperty(sockaddr)) {
            delete this.netPools[sockaddr];
            this.emit("close", sockaddr);
        }
    }
    close() {
        // 这里要干掉所有的链接
        let pools = Object.keys(this.netPools);
        for (let i = 0; i < pools.length; i++) {
            let net = this.netPools[pools[i]];
            if (net)
                net.close();
        }
    }
}
exports.TcpServer = TcpServer;
TcpServer.serPools = {};
// 链接socket
class TcpSocket {
    static create(host, port, count) {
        let sockaddr = host + ':' + port;
        if (count == null || count == undefined || count == 0)
            count = 1;
        let sockUnit = new TcpUnit(sockaddr, undefined, host, port);
        return sockUnit;
        // 连接成功
        // Debug("succ:" + host + ':' + port);
        // this.sendMsg(sock, 1, "欢迎实现connect&listen相同端口:" + port);
    }
    static _createSocket(sockUnit) {
        let sock = net.createConnection(sockUnit.port, sockUnit.host);
        sock.once("connect", () => {
            sockUnit.retryCount = 0;
            sockUnit.emit("connection");
        });
        sock.on("close", (e) => { sockUnit.emit('close', sockUnit.sockaddr, e); });
        sock.on("data", (data) => { sockUnit.reciveData(data); });
        sock.on("error", (e) => {
            sockUnit.emit('error', sockUnit.sockaddr, e);
        });
        sockUnit.sock = sock;
    }
    static connect(sockUnit) {
        if (sockUnit.sock) {
            sockUnit.sock.removeAllListeners();
            sockUnit.removeAllListeners();
        }
        let waitTime = 10 * 1000;
        if (sockUnit.retryCount < 5) {
            waitTime = 1000;
        }
        else if (sockUnit.retryCount < 10) {
            waitTime = 2 * 1000;
        }
        else if (sockUnit.retryCount < 20) {
            waitTime = 5 * 1000;
        }
        sockUnit.retryCount++;
        // 重连机制这里添加一个时间间隔
        timers_1.setTimeout(function () { TcpSocket._createSocket(sockUnit); }, waitTime);
        return new Promise(function (resolve, reject) {
            function onError(sockaddr, e) {
                sockUnit.off('connection', onConnection);
                reject([sockaddr, e]);
            }
            function onConnection(addr) {
                sockUnit.off('error', onError);
                resolve([addr]);
            }
            sockUnit.once('connection', onConnection);
            sockUnit.once('error', onError);
        });
    }
    static reConnect(sockUnit) {
        if (sockUnit.sock) {
            // sockUnit.sock.removeAllListeners('connect');
            // sockUnit.sock.removeAllListeners('close');
            // sockUnit.sock.removeAllListeners('data');
            // sockUnit.sock.removeAllListeners('error');
            sockUnit.sock.removeAllListeners();
        }
        let waitTime = 10 * 1000;
        if (sockUnit.retryCount < 5) {
            waitTime = 1000;
        }
        else if (sockUnit.retryCount < 10) {
            waitTime = 2 * 1000;
        }
        else if (sockUnit.retryCount < 20) {
            waitTime = 5 * 1000;
        }
        sockUnit.retryCount++;
        // 重连机制这里添加一个时间间隔
        timers_1.setTimeout(function () { TcpSocket._createSocket(sockUnit); }, waitTime);
    }
}
exports.TcpSocket = TcpSocket;