import { AppManager } from './app-managers/app-manager' ; import { Channel , PresenceChannel , PrivateChannel } from './channels' ; import { HttpApi } from './api' ; import { Log } from './log' ; import { Server } from './server' ; import { Stats } from './stats' ; const { constants } = require ( 'crypto' ) ; const dayjs = require ( 'dayjs' ) ; const packageFile = require ( '../package.json' ) ; /**
 * Echo server class.
 */ export class EchoServer { type { any } public options : any = { appManager : { driver : 'array' , api : { host : 'http://127.0.0.1' , endpoint : '/echo-server/app' , token : 'echo-app-token' , } , array : { apps : [ { id : 'echo-app' , key : 'echo-app-key' , secret : 'echo-app-secret' , maxConnections : - 1 , allowedOrigins : [ '*' ] , enableStats : false , } , ] , } , } , closingGracePeriod : 60 , cors : { credentials : true , origin : [ '*' ] , methods : [ 'GET' , 'POST' , 'PUT' , 'DELETE' , 'OPTIONS' ] , allowedHeaders : [ 'Origin' , 'Content-Type' , 'X-Auth-Token' , 'X-Requested-With' , 'Accept' , 'Authorization' , 'X-CSRF-TOKEN' , 'XSRF-TOKEN' , 'X-Socket-Id' , ] , } , database : { driver : 'redis' , redis : { host : '127.0.0.1' , port : 6379 , password : null , keyPrefix : '' , } , local : { } , } , development : false , host : null , headers : [ ] , stats : { enabled : true , driver : 'local' , snapshots : { interval : 60 * 60 , } , retention : { period : 7 * 24 * 60 * 60 , } , } , port : 6001 , protocol : 'http' , secureOptions : constants . SSL_OP_NO_TLSv1 , socketIoOptions : { } , ssl : { certPath : '' , keyPath : '' , caPath : '' , passphrase : '' , } , } ; type { Server } protected server : Server ; type { Channel } protected publicChannel : Channel ; type { PrivateChannel } protected privateChannel : PrivateChannel ; type { PresenceChannel } protected presenceChannel : PresenceChannel ; type { HttpApi } protected httpApi : HttpApi ; type { AppManager } protected appManager ; type { boolean } protected rejectNewConnections = false ; type { Stats } protected stats ; /**
     * Create a new Echo Server instance.
     */ constructor ( ) { } param { any } options return { Promise < any > } start ( options = { } ) { return new Promise ( ( resolve , reject ) => { this . options = Object . assign ( this . options , options ) ; this . server = new Server ( this . options ) ; Log . title ( `Echo Server v${ packageFile . version }\n` ) ; if ( this . options . development ) { Log . warning ( 'Starting the server in development mode...\n' ) ; } else { Log . info ( 'Starting the server...\n' ) ; } this . server . initialize ( ) . then ( io => { this . initialize ( io ) . then ( ( ) => { this . rejectNewConnections = false ; Log . info ( '\nServer ready!\n' ) ; if ( this . options . development ) { Log . info ( { options : JSON . stringify ( this . options ) } ) ; } resolve ( this ) ; } , error => Log . error ( error ) ) ; } , error => Log . error ( error ) ) ; } ) ; } param { any } io return { Promise < void > } initialize ( io ) { return new Promise ( ( resolve , reject ) => { this . appManager = new AppManager ( this . options ) ; this . stats = new Stats ( this . options ) ; this . publicChannel = new Channel ( io , this . stats , this . options ) ; this . privateChannel = new PrivateChannel ( io , this . stats , this . options ) ; this . presenceChannel = new PresenceChannel ( io , this . stats , this . options ) ; this . httpApi = new HttpApi ( this , io , this . server . express , this . options , this . appManager , this . stats ) ; this . httpApi . initialize ( ) ; let nsp = this . server . io . of ( /.*/ ) ; this . registerSocketMiddleware ( nsp ) ; this . registerConnectionCallbacks ( nsp ) ; this . registerStatsSnapshotter ( ) ; resolve ( ) ; } ) ; } return { Promise < void > } async stop ( ) { return new Promise ( ( resolve , reject ) => { console . log ( 'Stopping the server...' ) ; this . rejectNewConnections = true ; this . server . io . close ( async ( ) => { /**
                 * There is a timeout set to resolve the promise later
                 * since (in odd ways???) sync actions like disconnecting
                 * all sockets and decrementing the current connections count in stats
                 * do not finish for sync clients like Redis.
                 *
                 * So in order to do that, the server close should wait for
                 * a while before closing it, hence naming his timeout as "grace period"
                 * to resolve all socket.on() actions.
                 */ await setTimeout ( ( ) => resolve ( ) , this . options . closingGracePeriod * 1000 ) ; } ) ; } ) ; } param { any } socket return getNspForSocket ( socket ) { return socket ? socket . nsp . name : '/' ; } param { any } socket return { string | undefined } protected getAppKey ( socket ) { return this . getNspForSocket ( socket ) . replace ( /^\//g , '' ) ; } param { any } nsp return { void } protected registerSocketMiddleware ( nsp ) { nsp . use ( ( socket , next ) => { socket . id = this . generateSocketId ( ) ; this . checkIfSocketHasValidEchoApp ( socket ) . then ( socket => { this . checkIfSocketOriginIsAllowed ( socket ) . then ( socket => { next ( ) ; } , error => { if ( this . options . development ) { Log . error ( { time : new Date ( ) . toISOString ( ) , socketId : socket ? socket . id : null , action : 'origin_check' , status : 'failed' , error , } ) ; } next ( error ) ; } ) ; } , error => { if ( this . options . development ) { Log . error ( { time : new Date ( ) . toISOString ( ) , socketId : socket ? socket . id : null , action : 'app_check' , status : 'failed' , error , } ) ; } next ( error ) ; } ) ; } ) ; } param { any } nsp return { void } protected registerConnectionCallbacks ( nsp ) { nsp . on ( 'connection' , socket => { this . stats . markNewConnection ( socket . echoApp ) ; if ( this . rejectNewConnections ) { return socket . disconnect ( ) ; } this . checkIfSocketDidNotReachedLimit ( socket ) . then ( socket => { this . onSubscribe ( socket ) ; this . onUnsubscribe ( socket ) ; this . onDisconnecting ( socket ) ; this . onClientEvent ( socket ) ; } , error => { if ( this . options . development ) { Log . error ( { time : new Date ( ) . toISOString ( ) , socketId : socket ? socket . id : null , action : 'max_connections_check' , status : 'failed' , error , } ) ; } socket . disconnect ( ) ; } ) ; } ) ; } return { void } protected registerStatsSnapshotter ( ) { if ( this . options . stats . enabled ) { setInterval ( ( ) => { let time = dayjs ( ) . unix ( ) ; this . stats . getRegisteredApps ( ) . then ( apps => { apps . forEach ( name => { if ( this . options . development ) { Log . info ( { time , nsp : name , action : 'taking_snapshot' , } ) ; } let appKey = name . replace ( /^\//g , '' ) ; this . stats . takeSnapshot ( appKey , time ) . then ( ( ) => { this . stats . deleteStalePoints ( appKey , time ) ; } ) ; } ) ; } ) ; } , this . options . stats . snapshots . interval * 1000 ) ; } } param { any } socket return { void } protected onSubscribe ( socket ) { socket . on ( 'subscribe' , data => { this . getChannelInstance ( data . channel ) . join ( socket , data ) ; } ) ; } param { any } socket return { void } protected onUnsubscribe ( socket ) { socket . on ( 'unsubscribe' , data => { this . getChannelInstance ( data . channel ) . leave ( socket , data . channel , 'unsubscribed' ) ; } ) ; } param { any } socket return { void } protected onDisconnecting ( socket ) { socket . on ( 'disconnecting' , reason => { this . stats . markDisconnection ( socket . echoApp , reason ) . then ( ( ) => { socket . rooms . forEach ( room => { if ( room !== socket . id ) { this . getChannelInstance ( room ) . leave ( socket , room , reason ) ; } } ) ; } ) ; } ) ; } param { any } socket return { void } protected onClientEvent ( socket ) { socket . on ( 'client event' , data => { this . getChannelInstance ( data . channel ) . onClientEvent ( socket , data ) ; } ) ; } param { string } channel return { Channel | PrivateChannel | PresenceChannel } getChannelInstance ( channel ) { if ( Channel . isPresenceChannel ( channel ) ) { return this . presenceChannel ; } else if ( Channel . isPrivateChannel ( channel ) ) { return this . privateChannel ; } else { return this . publicChannel ; } } param { any } socket return { Promise < any > } protected checkIfSocketHasValidEchoApp ( socket ) { return new Promise ( ( resolve , reject ) => { if ( socket . echoApp ) { return resolve ( socket ) ; } let appKey = this . getAppKey ( socket ) ; this . appManager . findByKey ( appKey , socket , { } ) . then ( app => { console . log ( { socket } ) ; if ( ! app ) { reject ( { reason : `The app ${ appKey } does not exist` } ) ; } else { socket . echoApp = app ; resolve ( socket ) ; } } , error => { if ( this . options . development ) { Log . error ( { time : new Date ( ) . toISOString ( ) , socketId : socket ? socket . id : null , action : 'find_app' , status : 'failed' , error , } ) ; } reject ( error ) ; } ) ; } ) ; } param { any } socket return { Promise < any > } protected checkIfSocketDidNotReachedLimit ( socket ) { return new Promise ( ( resolve , reject ) => { if ( socket . disconnected || ! socket . echoApp ) { return reject ( { reason : 'The app connection limit cannot be checked because the socket is not authenticated.' } ) ; } this . server . io . of ( this . getNspForSocket ( socket ) ) . allSockets ( ) . then ( clients => { let maxConnections = parseInt ( socket . echoApp . maxConnections ) || - 1 ; if ( maxConnections < 0 ) { return resolve ( socket ) ; } if ( maxConnections >= clients . size ) { resolve ( socket ) ; } else { reject ( { reason : 'The current app reached connections limit.' } ) ; } } , error => { Log . error ( error ) ; reject ( error ) ; } ) ; } ) ; } param { any } socket return { Promise < any > } protected checkIfSocketOriginIsAllowed ( socket ) { return new Promise ( ( resolve , reject ) => { if ( socket . disconnected || ! socket . echoApp ) { return reject ( { reason : 'The origin cannot be checked because the socket is not authenticated.' } ) ; } let originIsAllowed = false ; let allowedOrigins = socket . echoApp . allowedOrigins || [ '*' ] ; let socketOrigin = socket . handshake . headers . origin || null ; if ( ! socketOrigin ) { return reject ( { reason : 'The origin header is not existent' } ) ; } allowedOrigins . forEach ( pattern => { let regex = new RegExp ( pattern . replace ( /(\.|\||\+|\?|\$|\/|\\)/g , '\\$1' ) . replace ( '*' , '.*' ) ) ; if ( regex . test ( socketOrigin ) ) { originIsAllowed = true ; } } ) ; if ( originIsAllowed ) { resolve ( socket ) ; } else { reject ( { reason : `The origin ${ socketOrigin } is not allowed.` } ) ; } } ) ; } return { string } protected generateSocketId ( ) { let min = 0 ; let max = 10000000000 ; let randomNumber = ( min , max ) => Math . floor ( Math . random ( ) * ( max - min + 1 ) + min ) ; return randomNumber ( min , max ) + '.' + randomNumber ( min , max ) ; } }