import Application from "..//application" ; import tcpServer from "../components/tcpServer" ; import { SocketProxy , loggerLevel , loggerType , } from "../util/interfaceDefine" ; import * as define from "../util/define" ; import * as rpcService from "./rpcService" ; let serverToken = "" ; let maxLen = 0 ; export function start ( app , cb ) { let rpcConfig = app . someconfig . rpc || { } ; maxLen = rpcConfig . maxLen || define . some_config . SocketBufferMaxLen ; let noDelay = rpcConfig . noDelay === false ? false : true ; tcpServer ( app . serverInfo . port , noDelay , startCb , newClientCb ) ; function startCb ( ) { let str = `listening at [${ app . serverInfo . host }:${ app . serverInfo . port }]  ${ app . serverId }` ; console . log ( str ) ; app . logger ( loggerType . frame , loggerLevel . info , str ) ; cb ( ) ; } function newClientCb ( socket ) { new RpcServerSocket ( app , socket ) ; } let tokenConfig = app . someconfig . recognizeToken || { } ; serverToken = tokenConfig . serverToken || define . some_config . Server_Token ; } class RpcServerSocket { private app : Application ; private socket : SocketProxy ; private id : string = "" ; private registered : boolean = false ; private registerTimer : NodeJS . Timeout = null as any ; private heartbeatTimer : NodeJS . Timeout = null as any ; private sendCache : boolean = false ; private sendArr : Buffer [ ] = [ ] ; private sendTimer : NodeJS . Timer = null as any ; constructor ( app , socket ) { this . app = app ; this . socket = socket ; socket . once ( "data" , this . onRegisterData . bind ( this ) ) ; socket . on ( "close" , this . onClose . bind ( this ) ) ; this . registerTimer = setTimeout ( function ( ) { app . logger ( loggerType . frame , loggerLevel . error , `rpcServer -> register timeout, close the rpc socket: ${ socket . remoteAddress }` ) ; socket . close ( ) ; } , 5000 ) ; } private onRegisterData ( data ) { try { let type = data . readUInt8 ( 0 ) ; if ( type === define . Rpc_Msg . register ) { this . registerHandle ( data ) ; } else { this . app . logger ( loggerType . frame , loggerLevel . error , `rpcServer -> illegal rpc register, close the rpc socket: ${ this . socket . remoteAddress }` ) ; this . socket . close ( ) ; } } catch ( e ) { this . socket . close ( ) ; this . app . logger ( loggerType . frame , loggerLevel . error , e . stack ) ; } } param data private onData ( data ) { try { let type = data . readUInt8 ( 0 ) ; if ( type === define . Rpc_Msg . clientMsgIn ) { this . app . backendServer . handleMsg ( this . id , data ) ; } else if ( type === define . Rpc_Msg . clientMsgOut ) { this . app . frontendServer . sendMsgByUids ( data ) ; } else if ( type === define . Rpc_Msg . rpcMsg ) { rpcService . handleMsg ( this . id , data ) ; } else if ( type === define . Rpc_Msg . rpcMsgAwait ) { rpcService . handleMsgAwait ( this . id , data ) ; } else if ( type === define . Rpc_Msg . applySession ) { this . app . frontendServer . applySession ( data ) ; } else if ( type === define . Rpc_Msg . heartbeat ) { this . heartbeatHandle ( ) ; this . heartbeatResponse ( ) ; } else { this . app . logger ( loggerType . frame , loggerLevel . error , `rpcServer -> illegal data type, close rpc client named: ${ this . id }` ) ; this . socket . close ( ) ; } } catch ( e ) { this . app . logger ( loggerType . msg , loggerLevel . error , e . stack ) ; } } /**
     * The socket connection is closed
     */ private onClose ( ) { clearTimeout ( this . registerTimer ) ; clearTimeout ( this . heartbeatTimer ) ; this . heartbeatTimer = null as any ; clearInterval ( this . sendTimer ) ; this . sendArr = [ ] ; if ( this . registered ) { this . app . rpcPool . removeSocket ( this . id ) ; } this . app . logger ( loggerType . frame , loggerLevel . error , `rpcServer -> a rpc client disconnected: ${ this . id }, ${ this . socket . remoteAddress }` ) ; } /**
     * register
     */ private registerHandle ( msg ) { clearTimeout ( this . registerTimer ) ; let data ; try { data = JSON . parse ( msg . slice ( 1 ) . toString ( ) ) ; } catch ( err ) { this . app . logger ( loggerType . frame , loggerLevel . error , `rpcServer -> JSON parse error，close the rpc socket: ${ this . socket . remoteAddress }` ) ; this . socket . close ( ) ; return ; } if ( data . serverToken !== serverToken ) { this . app . logger ( loggerType . frame , loggerLevel . error , `rpcServer -> illegal serverToken, close the rpc socket: ${ this . socket . remoteAddress }` ) ; this . socket . close ( ) ; return ; } if ( this . app . rpcPool . getSocket ( data . id ) ) { this . app . logger ( loggerType . frame , loggerLevel . error , `rpcServer -> already has a rpc client named: ${ data . id }, close it, ${ this . socket . remoteAddress }` ) ; this . socket . close ( ) ; return ; } if ( this . app . serverId <= data . id ) { this . socket . close ( ) ; return ; } this . registered = true ; this . socket . maxLen = maxLen ; this . socket . on ( "data" , this . onData . bind ( this ) ) ; this . id = data . id ; this . app . rpcPool . addSocket ( this . id , this ) ; this . app . logger ( loggerType . frame , loggerLevel . info , `rpcServer -> get new rpc client named: ${ this . id }` ) ; let rpcConfig = this . app . someconfig . rpc || { } ; let interval = 0 ; if ( rpcConfig . interval ) { if ( typeof rpcConfig . interval === "number" ) { interval = rpcConfig . interval ; } else { interval = rpcConfig . interval [ data . serverType ] || rpcConfig . interval . default || 0 ; } } if ( interval >= 10 ) { this . sendCache = true ; this . sendTimer = setInterval ( this . sendInterval . bind ( this ) , interval ) ; } let buffer = Buffer . allocUnsafe ( 5 ) ; buffer . writeUInt32BE ( 1 , 0 ) ; buffer . writeUInt8 ( define . Rpc_Msg . register , 4 ) ; this . socket . send ( buffer ) ; this . heartbeatHandle ( ) ; } /**
     * Heartbeat
     */ private heartbeatHandle ( ) { if ( this . heartbeatTimer ) { this . heartbeatTimer . refresh ( ) ; return ; } let rpcConfig = this . app . someconfig . rpc || { } ; let heartbeat = rpcConfig . heartbeat || define . some_config . Time . Rpc_Heart_Beat_Time ; if ( heartbeat < 5 ) { heartbeat = 5 ; } this . heartbeatTimer = setTimeout ( ( ) => { this . app . logger ( loggerType . frame , loggerLevel . warn , `rpcServer -> heartBeat time out, close it: ${ this . id }` ) ; this . socket . close ( ) ; } , heartbeat * 1000 * 2 ) ; } /**
     * Heartbeat response
     */ private heartbeatResponse ( ) { let buffer = Buffer . allocUnsafe ( 5 ) ; buffer . writeUInt32BE ( 1 , 0 ) ; buffer . writeUInt8 ( define . Rpc_Msg . heartbeat , 4 ) ; this . socket . send ( buffer ) ; } send ( data ) { if ( this . sendCache ) { this . sendArr . push ( data ) ; } else { this . socket . send ( data ) ; } } private sendInterval ( ) { if ( this . sendArr . length > 0 ) { this . socket . send ( Buffer . concat ( this . sendArr ) ) ; this . sendArr . length = 0 ; } } }