import * as net from "net" ; import { Node } from "./node" ; import { Socket } from "net" ; import { SocketHandler , SocketMessage } from "../core/socket-handler" ; import { Server } from "net" ; import { Global } from "../core/global" ; import { LoggingHelper } from "../core/logging-helper" ; import { BstStatistics , BstCommand , BstEvent } from "../statistics/bst-statistics" ; const Logger = "NODEMGR" ; export interface OnConnectCallback { ( node ) : void ; } export class NodeManager { public host : string = "0.0.0.0" ; public nodes : { [ id ] : Node ; } = { } ; private servers : NodeServer [ ] ; public onConnect : OnConnectCallback = null ; public onNodeRemoved : ( node : Node ) => void = null ; constructor ( private ports ) { } public node ( nodeID ) { return this . nodes [ nodeID ] ; } public async start ( ) { this . servers = [ ] ; for ( const port of this . ports ) { const server = new NodeServer ( this , port ) ; await server . start ( ) ; this . servers . push ( server ) ; } } public static onKeepAliveReceived ( node ) { node . socketHandler . send ( new SocketMessage ( Global . KeepAliveMessage ) ) ; } param callback public async stop ( ) { for ( let key of Object . keys ( this . nodes ) ) { const node = this . node ( key ) ; node . socketHandler . disconnect ( ) ; LoggingHelper . info ( Logger , "NODE CLOSING: " + node . id ) ; } for ( const server of this . servers ) { await server . stop ( ) ; } } } class NodeServer { private server : Server ; public constructor ( private nodeManager , private port ) { } public start ( ) { this . server = net . createServer ( ( socket ) => { let initialConnection = true ; let node = null ; const socketHandler = new SocketHandler ( socket , ( socketMessage ) => { const strMessage = socketMessage . asString ( ) ; if ( initialConnection ) { if ( ! socketMessage . isJSON ( ) ) { LoggingHelper . error ( Logger , "Error on parsing initial message: " + strMessage ) ; socketHandler . disconnect ( ) ; return ; } const connectData = socketMessage . asJSON ( ) ; node = new Node ( connectData . id , socketHandler ) ; this . nodeManager . nodes [ node . id ] = node ; socketHandler . send ( new SocketMessage ( "ACK" ) ) ; initialConnection = false ; if ( this . nodeManager . onConnect != null ) { this . nodeManager . onConnect ( node ) ; } BstStatistics . instance ( ) . record ( BstCommand . proxy , BstEvent . connect , node . id ) ; } else if ( strMessage === Global . KeepAliveMessage ) { NodeManager . onKeepAliveReceived ( node ) ; } else if ( node . handlingRequest ( ) ) { node . onReply ( socketMessage ) ; } } ) ; socketHandler . onCloseCallback = ( ) => { if ( node !== null ) { LoggingHelper . info ( Logger , "NODE CLOSED: " + node . id ) ; delete this . nodeManager . nodes [ node . id ] ; if ( this . nodeManager . onNodeRemoved !== undefined && this . nodeManager . onNodeRemoved !== null ) { this . nodeManager . onNodeRemoved ( node ) ; } } } ; LoggingHelper . info ( Logger , "NODE CONNECTED: " + socket . remoteAddress + ":" + socket . remotePort ) ; } ) . listen ( this . port , this . nodeManager . host ) ; return new Promise ( ( resolve ) => { this . server . on ( "listening" , ( ) => { LoggingHelper . info ( Logger , "NodeMgr Listening on " + this . nodeManager . host + ":" + this . port ) ; resolve ( ) ; } ) ; } ) ; } stop ( ) { return new Promise ( ( resolve , reject ) => { this . server . close ( function ( error ) { if ( error !== undefined ) { LoggingHelper . error ( Logger , "ERROR! NodeManager not stopped: " + error ) ; reject ( error ) ; } else { LoggingHelper . info ( Logger , "STOPPED" ) ; resolve ( ) ; } } ) ; } ) ; } }