import { ObjectId } from "bson" ; import chalk from "chalk" ; import * as path from "path" ; import { TSBuffer } from 'tsbuffer' ; import { ApiService , Counter , Flow , getCustomObjectIdTypes , MsgHandlerManager , MsgService , ParsedServerInput , ServiceMap , ServiceMapUtil , TransportDataUtil } from 'tsrpc-base-client' ; import { ApiReturn , ApiServiceDef , BaseServiceType , Logger , ServerInputData , ServiceProto , TsrpcError , TsrpcErrorType } from 'tsrpc-proto' ; import { ApiCallInner } from "../inner/ApiCallInner" ; import { InnerConnection } from "../inner/InnerConnection" ; import { TerminalColorLogger } from '../models/TerminalColorLogger' ; import { ApiCall } from './ApiCall' ; import { BaseConnection } from './BaseConnection' ; import { MsgCall } from './MsgCall' ; typeParam export abstract class BaseServer < ServiceType extends BaseServiceType = BaseServiceType > { throws abstract start ( ) ; /**
     * Stop server immediately, not waiting for the requests ending.
     */ abstract stop ( ) ; protected _status : ServerStatus = ServerStatus . Closed ; get status ( ) : ServerStatus { return this . _status ; } readonly proto : ServiceProto < ServiceType > ; readonly options : BaseServerOptions < ServiceType > ; readonly tsbuffer : TSBuffer ; readonly serviceMap : ServiceMap ; readonly logger : Logger ; protected _connIdCounter = new Counter ( 1 ) ; /**
     * Flow is a specific concept created by TSRPC family.
     * All pre-flow can interrupt latter behaviours.
     * All post-flow can NOT interrupt latter behaviours.
     */ readonly flows = { /** After the connection is created */ postConnectFlow : new Flow < BaseConnection < ServiceType > > ( ) , /** After the connection is disconnected */ postDisconnectFlow : new Flow < { conn : BaseConnection < ServiceType > ; reason ? : string ; } > ( ) , /**
         * Before processing the received data, usually be used to encryption / decryption.
         * Return `null | undefined` would ignore the buffer.
         */ preRecvDataFlow : new Flow < { conn : BaseConnection < ServiceType > ; data : string | Uint8Array | object ; } > ( ) , /**
         * Before send out data to network, usually be used to encryption / decryption.
         * Return `null | undefined` would not send the buffer.
         */ preSendDataFlow : new Flow < { conn : BaseConnection < ServiceType > ; data : string | Uint8Array | object ; call ? : ApiCall ; } > ( ) , deprecated preRecvBufferFlow : new Flow < { conn : BaseConnection < ServiceType > ; buf : Uint8Array ; } > ( ) , deprecated preSendBufferFlow : new Flow < { conn : BaseConnection < ServiceType > ; buf : Uint8Array ; call ? : ApiCall ; } > ( ) , /**
         * Before a API request is send.
         * Return `null | undefined` would cancel the request.
         */ preApiCallFlow : new Flow < ApiCall > ( ) , /**
         * Before return the `ApiReturn` to the client.
         * It may be used to change the return value, or return `null | undefined` to abort the request.
         */ preApiReturnFlow : new Flow < { call : ApiCall ; return : ApiReturn < any > ; } > ( ) , /**
         * After the `ApiReturn` is send.
         * return `null | undefined` would NOT interrupt latter behaviours.
         */ postApiReturnFlow : new Flow < { call : ApiCall ; return : ApiReturn < any > ; } > ( ) , /**
         * After the api handler is executed.
         * return `null | undefined` would NOT interrupt latter behaviours.
         */ postApiCallFlow : new Flow < ApiCall > ( ) , /**
         * Before handle a `MsgCall`
         */ preMsgCallFlow : new Flow < MsgCall > ( ) , /**
         * After handlers of a `MsgCall` are executed.
         * return `null | undefined` would NOT interrupt latter behaviours.
         */ postMsgCallFlow : new Flow < MsgCall > ( ) , /**
         * Before send out a message.
         * return `null | undefined` would NOT interrupt latter behaviours.
         */ preSendMsgFlow : new Flow < { conn : BaseConnection < ServiceType > ; service : MsgService ; msg : any ; } > ( ) , /**
         * After send out a message.
         * return `null | undefined` would NOT interrupt latter behaviours.
         */ postSendMsgFlow : new Flow < { conn : BaseConnection < ServiceType > ; service : MsgService ; msg : any ; } > ( ) , } as const ; private _apiHandlers : { [ apiName ] : ApiHandler < any > | undefined ; } = { } ; private _msgHandlers : MsgHandlerManager = new MsgHandlerManager ( ) ; private static _isUncaughtExceptionProcessed = false ; param logger returns static processUncaughtException ( logger ) { if ( this . _isUncaughtExceptionProcessed ) { return ; } this . _isUncaughtExceptionProcessed = true ; process . on ( 'uncaughtException' , e => { logger . error ( '[uncaughtException]' , e ) ; } ) ; process . on ( 'unhandledRejection' , e => { logger . error ( '[unhandledRejection]' , e ) ; } ) ; } constructor ( proto , options ) { this . proto = proto ; this . options = options ; if ( this . options . json ) { this . options . jsonEnabled = true ; } this . tsbuffer = new TSBuffer ( { ... proto . types , ... getCustomObjectIdTypes ( ObjectId ) } , { strictNullChecks : this . options . strictNullChecks } ) ; this . serviceMap = ServiceMapUtil . getServiceMap ( proto ) ; this . logger = this . options . logger ; BaseServer . processUncaughtException ( this . logger ) ; this . _setDefaultFlowOnError ( ) ; } protected _setDefaultFlowOnError ( ) { this . flows . preApiCallFlow . onError = ( e , call ) => { if ( e instanceof TsrpcError ) { call . error ( e ) ; } else { this . onInternalServerError ( e , call ) ; } } ; this . flows . postApiCallFlow . onError = ( e , call ) => { if ( ! call . return ) { if ( e instanceof TsrpcError ) { call . error ( e ) ; } else { this . onInternalServerError ( e , call ) ; } } else { call . logger . error ( 'postApiCallFlow Error:' , e ) ; } } ; this . flows . preApiReturnFlow . onError = ( e , last ) => { last . call [ '_return' ] = undefined ; if ( e instanceof TsrpcError ) { last . call . error ( e ) ; } else { this . onInternalServerError ( e , last . call ) ; } } ; this . flows . postApiReturnFlow . onError = ( e , last ) => { if ( ! last . call . return ) { if ( e instanceof TsrpcError ) { last . call . error ( e ) ; } else { this . onInternalServerError ( e , last . call ) ; } } } ; } protected _pendingApiCallNum = 0 ; /**
     * Process the buffer, after the `preRecvBufferFlow`.
     */ async _onRecvData ( conn , data , serviceName ? ) { if ( ! ( conn instanceof InnerConnection ) && this . status !== ServerStatus . Opened ) { return ; } if ( this . options . debugBuf ) { if ( typeof data === 'string' ) { conn . logger ?. debug ( `[RecvText] length=${ data . length }` , data ) ; } else if ( data instanceof Uint8Array ) { conn . logger ?. debug ( `[RecvBuf] length=${ data . length }` , data ) ; } else { conn . logger ?. debug ( '[RecvJSON]' , data ) ; } } if ( typeof data === 'string' && ! this . options . jsonEnabled ) { this . onInputDataError ( 'JSON mode is not enabled, please use binary instead.' , conn , data ) ; return ; } let pre = await this . flows . preRecvDataFlow . exec ( { conn : conn , data : data } , conn . logger ) ; if ( ! pre ) { return ; } data = pre . data ; if ( data instanceof Uint8Array ) { let preBuf = await this . flows . preRecvBufferFlow . exec ( { conn : conn , buf : data } , conn . logger ) ; if ( ! preBuf ) { return ; } data = preBuf . buf ; } let opInput = this . _parseServerInput ( this . tsbuffer , this . serviceMap , data , serviceName ) ; if ( ! opInput . isSucc ) { this . onInputDataError ( opInput . errMsg , conn , data ) ; return ; } let call = conn . makeCall ( opInput . result ) ; if ( call . type === 'api' ) { await this . _handleApiCall ( call ) ; } else { await this . _onMsgCall ( call ) ; } } protected async _handleApiCall ( call ) { ++ this . _pendingApiCallNum ; await this . _onApiCall ( call ) ; if ( -- this . _pendingApiCallNum === 0 ) { this . _gracefulStop ?. rs ( ) ; } } protected async _onApiCall ( call ) { let timeoutTimer = this . options . apiTimeout ? setTimeout ( ( ) => { if ( ! call . return ) { call . error ( 'Server Timeout' , { code : 'SERVER_TIMEOUT' , type : TsrpcErrorType . ServerError } ) ; } timeoutTimer = undefined ; } , this . options . apiTimeout ) : undefined ; let preFlow = await this . flows . preApiCallFlow . exec ( call , call . logger ) ; if ( ! preFlow ) { if ( timeoutTimer ) { clearTimeout ( timeoutTimer ) ; timeoutTimer = undefined ; } return ; } call = preFlow ; call . logger . log ( '[ApiReq]' , this . options . logReqBody ? call . req : '' ) ; let { handler } = await this . getApiHandler ( call . service , this . _delayImplementApiPath , call . logger ) ; if ( handler ) { try { await handler ( call ) ; } catch ( e ) { if ( e instanceof TsrpcError ) { call . error ( e ) ; } else { this . onInternalServerError ( e , call ) ; } } } else { call . error ( `Unhandled API: ${ call . service . name }` , { code : 'UNHANDLED_API' , type : TsrpcErrorType . ServerError } ) ; } await this . flows . postApiCallFlow . exec ( call , call . logger ) ; if ( timeoutTimer ) { clearTimeout ( timeoutTimer ) ; timeoutTimer = undefined ; } } protected async _onMsgCall ( call ) { if ( call . conn . type === 'SHORT' ) { call . conn . close ( ) ; } let preFlow = await this . flows . preMsgCallFlow . exec ( call , call . logger ) ; if ( ! preFlow ) { return ; } call = preFlow ; this . options . logMsg && call . logger . log ( '[RecvMsg]' , call . msg ) ; let promises = [ ... ( call . conn [ '_msgHandlers' ] ?. forEachHandler ( call . service . name , call . logger , call ) ?? [ ] ) , this . _msgHandlers . forEachHandler ( call . service . name , call . logger , call ) ] ; if ( ! promises . length ) { this . logger . debug ( '[UNHANDLED_MSG]' , call . service . name ) ; } else { await Promise . all ( promises ) ; } await this . flows . postMsgCallFlow . exec ( call , call . logger ) ; } param apiName param handler implementApi < Api extends keyof ServiceType [ 'api' ] , Call extends ApiCall < ServiceType [ 'api' ] [ Api ] [ 'req' ] , ServiceType [ 'api' ] [ Api ] [ 'res' ] > > ( apiName , handler ) { if ( this . _apiHandlers [ apiName as string ] ) { throw new Error ( 'Already exist handler for API: ' + apiName ) ; } this . _apiHandlers [ apiName as string ] = handler ; this . logger . log ( `API implemented succ: [${ apiName }]` ) ; } ; /** 用于延迟注册 API */ protected _delayImplementApiPath ? : string ; param apiPath returns async autoImplementApi ( apiPath , delay ? ) { let apiServices = Object . values ( this . serviceMap . apiName2Service ) as ApiServiceDef [ ] ; let output = { succ : [ ] , fail : [ ] } ; if ( delay ) { this . _delayImplementApiPath = apiPath ; return output ; } for ( let svc of apiServices ) { let { handler , errMsg } = await this . getApiHandler ( svc , apiPath , this . logger ) ; if ( ! handler ) { output . fail . push ( svc . name ) ; let logMsg = chalk . red ( `Implement API ${ chalk . cyan . underline ( `${ svc . name }` ) } failed:` ) ; logMsg += '\n  |- ' + errMsg ; this . logger . error ( logMsg ) ; continue ; } this . implementApi ( svc . name , handler ) ; output . succ . push ( svc . name ) ; } if ( output . fail . length ) { this . logger . error ( chalk . red ( `${ output . fail . length } API implemented failed: ` + output . fail . map ( v => chalk . cyan . underline ( v ) ) . join ( ' ' ) ) ) ; } return output ; } async getApiHandler ( svc , apiPath ? , logger ? ) { if ( this . _apiHandlers [ svc . name ] ) { return { handler : this . _apiHandlers [ svc . name ] ! } ; } if ( ! apiPath ) { return { errMsg : `Api not implemented: ${ svc . name }` } ; } let match = svc . name . match ( /^(.+\/)*(.+)$/ ) ; if ( ! match ) { logger ?. error ( 'Invalid apiName: ' + svc . name ) ; return { errMsg : `Invalid api name: ${ svc . name }` } ; } let handlerPath = match [ 1 ] || '' ; let handlerName = match [ 2 ] ; let modulePath = path . resolve ( apiPath , handlerPath , 'Api' + handlerName ) ; try { let handlerModule = await import ( modulePath ) ; let handler = handlerModule . default ?? handlerModule [ 'Api' + handlerName ] ; if ( handler ) { return { handler : handler } ; } else { return { errMsg : `Cannot find export { ${ 'Api' + handlerName } } at: ${ modulePath }` } ; } } catch ( e ) { if ( e . code === 'MODULE_NOT_FOUND' ) { return { errMsg : `Module not found: ${ modulePath }` } ; } else { return { errMsg : e . message } ; } } } param msgName param handler listenMsg < Msg extends keyof ServiceType [ 'msg' ] , Call extends MsgCall < ServiceType [ 'msg' ] [ Msg ] > > ( msgName , handler ) { this . _msgHandlers . addHandler ( msgName as string , handler ) ; return handler ; } ; /**
     * Remove a message handler
     */ unlistenMsg < Msg extends keyof ServiceType [ 'msg' ] , Call extends MsgCall < ServiceType [ 'msg' ] [ Msg ] > > ( msgName , handler ) { this . _msgHandlers . removeHandler ( msgName as string , handler ) ; } ; /**
     * Remove all handlers from a message
     */ unlistenMsgAll < Msg extends keyof ServiceType [ 'msg' ] , Call extends MsgCall < ServiceType [ 'msg' ] [ Msg ] > > ( msgName ) { this . _msgHandlers . removeAllHandlers ( msgName as string ) ; } ; /**
     * Event when the server cannot parse input buffer to api/msg call.
     * By default, it will return "Input Data Error" .
     */ async onInputDataError ( errMsg , conn , data ) { if ( this . options . debugBuf ) { if ( typeof data === 'string' ) { conn . logger . error ( `[InputDataError] ${ errMsg } length = ${ data . length }` , data ) ; } else if ( data instanceof Uint8Array ) { conn . logger . error ( `[InputBufferError] ${ errMsg } length = ${ data . length }` , data . subarray ( 0 , 16 ) ) ; } else { conn . logger . error ( `[InputJsonError] ${ errMsg } ` , data ) ; } } const message = data instanceof Uint8Array ? `Invalid request buffer, please check the version of service proto.` : errMsg ; if ( conn . type === 'SHORT' ) { let opEncode = ApiCall . encodeApiReturn ( this . tsbuffer , { type : 'api' , name : '?' , id : 0 , reqSchemaId : '?' , resSchemaId : '?' } , { isSucc : false , err : new TsrpcError ( { message : message , type : TsrpcErrorType . ServerError , code : 'INPUT_DATA_ERR' } ) } , conn . dataType ) ; if ( opEncode . isSucc ) { let opSend = await conn . sendData ( opEncode . output ) ; if ( opSend . isSucc ) { return ; } } } conn . close ( message ) ; } /**
     * Event when a uncaught error (except `TsrpcError`) is throwed.
     * By default, it will return a `TsrpcError` with message "Internal server error".
     * If `returnInnerError` is `true`, the original error would be returned as `innerErr` property.
     */ onInternalServerError ( err , call ) { call . logger . error ( err ) ; call . error ( 'Internal Server Error' , { code : 'INTERNAL_ERR' , type : TsrpcErrorType . ServerError , innerErr : call . conn . server . options . returnInnerError ? err . message : undefined } ) ; } protected _gracefulStop ? : { rs : ( ) => void ; } ; param maxWaitTime async gracefulStop ( maxWaitTime ? ) { if ( this . _status !== ServerStatus . Opened ) { throw new Error ( `Cannot gracefulStop when server status is '${ this . _status }'.` ) ; } this . logger . log ( '[GracefulStop] Start graceful stop, waiting all ApiCall finished...' ) ; this . _status = ServerStatus . Closing ; let promiseWaitApi = new Promise < void > ( rs => { this . _gracefulStop = { rs : rs } ; } ) ; return new Promise < void > ( rs => { let maxWaitTimer ; if ( maxWaitTime ) { maxWaitTimer = setTimeout ( ( ) => { maxWaitTimer = undefined ; if ( this . _gracefulStop ) { this . _gracefulStop = undefined ; this . logger . log ( 'Graceful stop timeout, stop the server directly.' ) ; this . stop ( ) . then ( ( ) => { rs ( ) ; } ) ; } } , maxWaitTime ) ; } promiseWaitApi . then ( ( ) => { this . logger . log ( 'All ApiCall finished, continue stop server.' ) ; if ( maxWaitTimer ) { clearTimeout ( maxWaitTimer ) ; maxWaitTimer = undefined ; } if ( this . _gracefulStop ) { this . _gracefulStop = undefined ; this . stop ( ) . then ( ( ) => { rs ( ) ; } ) ; } } ) ; } ) ; } param apiName param req param options callApi < T extends keyof ServiceType [ 'api' ] > ( apiName , req ) { return new Promise ( rs => { let service = this . serviceMap . apiName2Service [ apiName as string ] ; if ( ! service ) { let errMsg = `Cannot find service: ${ apiName }` ; this . logger . warn ( `[callApi]` , errMsg ) ; rs ( { isSucc : false , err : new TsrpcError ( errMsg , { type : TsrpcErrorType . ServerError , code : 'ERR_API_NAME' } ) } ) ; return ; } let conn = new InnerConnection ( { dataType : 'json' , server : this , id : '' + this . _connIdCounter . getNext ( ) , ip : '' , return : { type : 'raw' , rs : rs } } ) ; let call = new ApiCallInner ( { conn : conn , req : req , service : service } ) ; this . _handleApiCall ( call ) ; } ) ; } param apiName param jsonReq returns param apiName param req returns async inputJSON ( apiName , req ) { if ( apiName . startsWith ( '/' ) ) { apiName = apiName . slice ( 1 ) ; } if ( ! this . serviceMap . apiName2Service [ apiName ] ) { return { isSucc : false , err : new TsrpcError ( `Invalid service name: ${ apiName }` , { type : TsrpcErrorType . ServerError , code : 'INPUT_DATA_ERR' } ) } ; } return new Promise ( rs => { let conn = new InnerConnection ( { dataType : 'json' , server : this , id : '' + this . _connIdCounter . getNext ( ) , ip : '' , return : { type : 'json' , rs : rs } } ) ; this . _onRecvData ( conn , req , apiName ) ; } ) ; } param buf returns inputBuffer ( buf ) { return new Promise ( rs => { let conn = new InnerConnection ( { dataType : 'buffer' , server : this , id : '' + this . _connIdCounter . getNext ( ) , ip : '' , return : { type : 'buffer' , rs : rs } } ) ; this . _onRecvData ( conn , buf ) ; } ) ; } protected _parseServerInput ( tsbuffer , serviceMap , data , serviceName ? ) { if ( data instanceof Uint8Array ) { let opServerInputData = TransportDataUtil . tsbuffer . decode ( data , 'ServerInputData' ) ; if ( ! opServerInputData . isSucc ) { return opServerInputData ; } let serverInput = opServerInputData . value as ServerInputData ; let service = serviceMap . id2Service [ serverInput . serviceId ] ; if ( ! service ) { return { isSucc : false , errMsg : `Cannot find service ID: ${ serverInput . serviceId }` } ; } if ( service . type === 'api' ) { let opReq = tsbuffer . decode ( serverInput . buffer , service . reqSchemaId ) ; return opReq . isSucc ? { isSucc : true , result : { type : 'api' , service : service , req : opReq . value , sn : serverInput . sn } } : opReq ; } else { let opMsg = tsbuffer . decode ( serverInput . buffer , service . msgSchemaId ) ; return opMsg . isSucc ? { isSucc : true , result : { type : 'msg' , service : service , msg : opMsg . value } } : opMsg ; } } else { let json ; if ( typeof data === 'string' ) { try { json = JSON . parse ( data ) ; } catch ( e ) { return { isSucc : false , errMsg : `Input is not a valid JSON string: ${ e . message }` } ; } } else { json = data ; } let body ; let sn ; let service ; if ( serviceName == undefined ) { if ( ! Array . isArray ( json ) ) { return { isSucc : false , errMsg : `Invalid request format` } ; } serviceName = json [ 0 ] as string ; body = json [ 1 ] ; sn = json [ 2 ] ; } else { body = json ; } service = serviceMap . apiName2Service [ serviceName ] ?? serviceMap . msgName2Service [ serviceName ] ; if ( ! service ) { return { isSucc : false , errMsg : `Invalid service name: ${ serviceName }` } ; } if ( service . type === 'api' ) { let op = tsbuffer . decodeJSON ( body , service . reqSchemaId ) ; if ( ! op . isSucc ) { return op ; } return { isSucc : true , result : { type : 'api' , service : service , sn : sn , req : op . value } } ; } else { let op = tsbuffer . decodeJSON ( body , service . msgSchemaId ) ; if ( ! op . isSucc ) { return op ; } return { isSucc : true , result : { type : 'msg' , service : service , msg : op . value } } ; } } } } export interface BaseServerOptions < ServiceType extends BaseServiceType > { remarks defaultValue json : boolean ; deprecated jsonEnabled ? : boolean ; defaultValue strictNullChecks : boolean ; defaultValue apiTimeout : number | undefined ; defaultValue logger : Logger ; defaultValue logReqBody : boolean ; defaultValue logResBody : boolean ; defaultValue logMsg : boolean ; /**
     * If `true`, all sent and received raw buffer would be print into the log.
     * It may be useful when you do something for buffer encryption/decryption, and want to debug them.
     */ debugBuf ? : boolean ; defaultValue returnInnerError : boolean ; } export const defaultBaseServerOptions = { json : false , strictNullChecks : false , apiTimeout : 30000 , logger : new TerminalColorLogger , logReqBody : true , logResBody : true , logMsg : true , returnInnerError : process . env [ 'NODE_ENV' ] !== 'production' } ; export type ApiHandler < Call extends ApiCall = ApiCall > = ( call : Call ) => void | Promise < void > ; export type MsgHandler < Call extends MsgCall = MsgCall > = ( call : Call ) => void | Promise < void > ; export enum ServerStatus { Opening = 'OPENING' , Opened = 'OPENED' , Closing = 'CLOSING' , Closed = 'CLOSED' }