import * as http from "http" ; import { EncodeOutput , TransportDataUtil } from "tsrpc-base-client" ; import { BaseServiceType , ServiceProto } from 'tsrpc-proto' ; import * as WebSocket from 'ws' ; import { Server as WebSocketServer } from 'ws' ; import { HttpUtil } from '../../models/HttpUtil' ; import { BaseServer , BaseServerOptions , defaultBaseServerOptions , ServerStatus } from '../base/BaseServer' ; import { WsConnection } from './WsConnection' ; typeParam export class WsServer < ServiceType extends BaseServiceType = any > extends BaseServer < ServiceType > { readonly options ! : WsServerOptions < ServiceType > ; readonly connections : WsConnection < ServiceType > [ ] = [ ] ; private readonly _id2Conn : { [ connId ] : WsConnection < ServiceType > | undefined ; } = { } ; constructor ( proto , options ? ) { super ( proto , { ... defaultWsServerOptions , ... options } ) ; } private _wsServer ? : WebSocketServer ; /**
     * {@inheritDoc BaseServer.start}
     */ start ( ) { if ( this . _wsServer ) { throw new Error ( 'Server already started' ) ; } this . _status = ServerStatus . Opening ; return new Promise ( ( rs , rj ) => { this . logger . log ( 'Starting WebSocket server...' ) ; this . _wsServer = new WebSocketServer ( { port : this . options . port } , ( ) => { this . logger . log ( `Server started at ${ this . options . port }...` ) ; this . _status = ServerStatus . Opened ; rs ( ) ; } ) ; this . _wsServer . on ( 'connection' , this . _onClientConnect ) ; this . _wsServer . on ( 'error' , e => { this . logger . error ( '[ServerError]' , e ) ; rj ( e ) ; } ) ; } ) ; } /**
     * {@inheritDoc BaseServer.stop}
     */ async stop ( ) { if ( ! this . _wsServer ) { throw new Error ( 'Server has not been started' ) ; } if ( this . _status === ServerStatus . Closed ) { throw new Error ( 'Server is closed already' ) ; } this . _status = ServerStatus . Closing ; return new Promise < void > ( async ( rs , rj ) => { await Promise . all ( this . connections . map ( v => v . close ( 'Server Stop' ) ) ) ; this . _wsServer ! . close ( err => { err ? rj ( err ) : rs ( ) ; } ) ; } ) . then ( ( ) => { this . logger . log ( 'Server stopped' ) ; this . _status = ServerStatus . Closed ; this . _wsServer = undefined ; } ) ; } private _onClientConnect = ( ws , httpReq ) => { if ( this . _status !== ServerStatus . Opened ) { ws . close ( 1012 ) ; return ; } let isDataTypeConfirmed = true ; let dataType ; let protocols = httpReq . headers [ 'sec-websocket-protocol' ] ?. split ( ',' ) . map ( v => v . trim ( ) ) . filter ( v => ! ! v ) ; if ( protocols ?. includes ( 'text' ) ) { dataType = 'text' ; } else if ( protocols ?. includes ( 'buffer' ) ) { dataType = 'buffer' ; } else { dataType = this . options . jsonEnabled ? 'text' : 'buffer' ; isDataTypeConfirmed = false ; } let conn = new WsConnection ( { id : '' + this . _connIdCounter . getNext ( ) , ip : HttpUtil . getClientIp ( httpReq ) , server : this , ws : ws , httpReq : httpReq , onClose : this . _onClientClose , dataType : dataType , isDataTypeConfirmed : isDataTypeConfirmed } ) ; this . connections . push ( conn ) ; this . _id2Conn [ conn . id ] = conn ; conn . logger . log ( '[Connected]' , `ActiveConn=${ this . connections . length }` ) ; this . flows . postConnectFlow . exec ( conn , conn . logger ) ; } ; private _onClientClose = async ( conn , code , reason ) => { this . connections . removeOne ( v => v . id === conn . id ) ; delete this . _id2Conn [ conn . id ] ; conn . logger . log ( '[Disconnected]' , `Code=${ code } ${ reason ? `Reason=${ reason } ` : '' }ActiveConn=${ this . connections . length }` ) ; await this . flows . postDisconnectFlow . exec ( { conn : conn , reason : reason } , conn . logger ) ; } ; param msgName param msg param connIds returns async broadcastMsg < T extends keyof ServiceType [ 'msg' ] > ( msgName , msg , conns ? ) { let connStr ; if ( ! conns ) { conns = this . connections ; connStr = '*' ; } else { connStr = conns ? conns . map ( v => v . id ) . join ( ',' ) : '*' ; } if ( ! conns . length ) { return { isSucc : true } ; } if ( this . status !== ServerStatus . Opened ) { this . logger . warn ( '[BroadcastMsgErr]' , `[${ msgName }]` , `[To:${ connStr }]` , 'Server not open' ) ; return { isSucc : false , errMsg : 'Server not open' } ; } let service = this . serviceMap . msgName2Service [ msgName as string ] ; if ( ! service ) { this . logger . warn ( '[BroadcastMsgErr]' , `[${ msgName }]` , `[To:${ connStr }]` , 'Invalid msg name: ' + msgName ) ; return { isSucc : false , errMsg : 'Invalid msg name: ' + msgName } ; } let _opEncodeBuf ; let _opEncodeText ; const getOpEncodeBuf = ( ) => { if ( ! _opEncodeBuf ) { _opEncodeBuf = TransportDataUtil . encodeServerMsg ( this . tsbuffer , service ! , msg , 'buffer' , 'LONG' ) ; } return _opEncodeBuf ; } ; const getOpEncodeText = ( ) => { if ( ! _opEncodeText ) { _opEncodeText = TransportDataUtil . encodeServerMsg ( this . tsbuffer , service ! , msg , 'text' , 'LONG' ) ; } return _opEncodeText ; } ; let op = conns . some ( v => v . dataType === 'buffer' ) ? getOpEncodeBuf ( ) : getOpEncodeText ( ) ; if ( ! op . isSucc ) { this . logger . warn ( '[BroadcastMsgErr]' , `[${ msgName }]` , `[To:${ connStr }]` , op . errMsg ) ; return op ; } this . options . logMsg && this . logger . log ( `[BroadcastMsg]` , `[${ msgName }]` , `[To:${ connStr }]` , msg ) ; let errMsgs = [ ] ; return Promise . all ( conns . map ( async ( conn ) => { let pre = await this . flows . preSendMsgFlow . exec ( { conn : conn , service : service ! , msg : msg } , this . logger ) ; if ( ! pre ) { return { isSucc : false , errMsg : 'Prevented by preSendMsgFlow' } ; } msg = pre . msg ; let opSend = await conn . sendData ( ( conn . dataType === 'buffer' ? getOpEncodeBuf ( ) : getOpEncodeText ( ) ) ! . output ! ) ; if ( ! opSend . isSucc ) { return opSend ; } this . flows . postSendMsgFlow . exec ( pre , this . logger ) ; return { isSucc : true } ; } ) ) . then ( results => { for ( let i = 0 ; i < results . length ; ++ i ) { let op = results [ i ] ; if ( ! op . isSucc ) { errMsgs . push ( `Conn#conns[i].id: ${ op . errMsg }` ) ; } ; } if ( errMsgs . length ) { return { isSucc : false , errMsg : errMsgs . join ( '\n' ) } ; } else { return { isSucc : true } ; } } ) ; } ; } export interface WsServerOptions < ServiceType extends BaseServiceType > extends BaseServerOptions < ServiceType > { /** Which port the WebSocket server is listen to */ port : number ; } ; const defaultWsServerOptions = { ... defaultBaseServerOptions , port : 3000 } ;