import { GethBin } from './GethBin' ; import { Web3 } from './Web3' ; import { Socket } from 'net' ; import * as event from './Constants' ; import { EventEmitter } from 'events' ; import { ChildProcess , exec , spawn } from 'child_process' ; import * as Promise from 'bluebird' ; import { homedir , type as osType } from 'os' ; import { join as pathJoin } from 'path' ; const platform = osType ( ) ; const symbolEnforcer = Symbol ( 'go-ethereum' ) ; const symbol = Symbol ( 'go-ethereum' ) ; const cpuPriority = { unix : { [ event . PriorityCode . HIGH ] : '-17' , [ event . PriorityCode . NORMAL ] : '0' , [ event . PriorityCode . LOW ] : '17' , } , win : { [ event . PriorityCode . HIGH ] : 'high' , [ event . PriorityCode . NORMAL ] : 'normal' , [ event . PriorityCode . LOW ] : 'below normal' , } , } ; export default class GethConnector extends EventEmitter { public downloadManager = new GethBin ( ) ; public ipcStream = new Web3 ( ) ; public logger : any = console ; public spawnOptions = new Map ( ) ; public gethService : ChildProcess ; public serviceStatus : { process : boolean ; api : boolean ; version : string ; } = { process : false , api : false , version : '' , } ; public watchers = new Map ( ) ; private socket : Socket ; private connectedToLocal : boolean = false ; private cpuPriority = event . PriorityCode . LOW ; private _downloadEventsEnabled = false ; param enforcer constructor ( enforcer ) { super ( ) ; if ( enforcer !== symbolEnforcer ) { throw new Error ( 'Use .getInstance() instead of new constructor' ) ; } } returns { Web3Factory } get web3 ( ) { return this . ipcStream . web3 ; } returns { GethConnector } public static getInstance ( ) { if ( ! this [ symbol ] ) { this [ symbol ] = new GethConnector ( symbolEnforcer ) ; } return this [ symbol ] ; } returns { string } static getDefaultDatadir ( ) { let dataDir ; switch ( platform ) { case 'Linux' : dataDir = pathJoin ( homedir ( ) , '.ethereum' ) ; break ; case 'Darwin' : dataDir = pathJoin ( homedir ( ) , 'Library' , 'Ethereum' ) ; break ; case 'Windows_NT' : dataDir = pathJoin ( process . env . APPDATA , '/Ethereum' ) ; break ; default : throw new Error ( 'Platform not supported' ) ; } return dataDir ; } returns { string } static getDefaultIpcPath ( ) { const dataDirPath = GethConnector . getDefaultDatadir ( ) ; let ipcPath ; switch ( platform ) { case 'Windows_NT' : ipcPath = '\\\\.\\pipe\\geth.ipc' ; break ; default : ipcPath = pathJoin ( dataDirPath , 'geth.ipc' ) ; break ; } return ipcPath ; } param logger public setLogger ( logger ) { this . logger = logger ; } param path public setBinPath ( path ) { this . downloadManager = new GethBin ( path ) ; this . _downloadEventsEnabled = false ; } fires param options public start ( options ? ) { event this . emit ( event . STARTING ) ; this . setOptions ( options ) ; this . socket = new Socket ( ) ; return this . _checkBin ( ) . then ( ( binPath ) => { if ( ! binPath ) { event this . emit ( event . FAILED , 'gethBin' ) ; this . serviceStatus . process = false ; return false ; } if ( this . connectedToLocal ) { return false ; } this . gethService = spawn ( binPath , this . _flattenOptions ( ) , { detached : true } ) ; return true ; } ) . then ( ( passed ) => { if ( passed ) { this . _attachEvents ( ) ; } return passed ; } ) ; } fires fires returns { Bluebird < U > } public stop ( ) { event this . emit ( event . STOPPING ) ; this . _flushEvents ( ) ; if ( ! this . gethService ) { event this . emit ( event . STOPPED ) ; } const killProcess = ( this . gethService ) ? this . gethService . kill ( ) : true ; return Promise . delay ( 500 ) ; } returns { any } public getChainFolder ( ) { return this . web3 . admin . getDatadir ( ) . then ( ( datadir ) => { return pathJoin ( datadir , 'chaindata' ) ; } ) ; } /**
     * connect to existing geth ipc
     */ public connectToLocal ( ) { this . _connectToIPC ( ) ; this . connectedToLocal = true ; } public setCpuPriority ( level , immediate ? ) { if ( ! event . PriorityCode . hasOwnProperty ( level ) ) { throw new Error ( 'Invalid priority level' ) ; } this . cpuPriority = level ; if ( immediate ) { this . executeCpuPriority ( ) ; } } public executeCpuPriority ( ) { if ( ! this . gethService . pid ) { this . logger . warn ( 'cpu:geth process is not started' ) ; return ; } const command = ( process . platform === 'win32' ) ? `wmic process where name="geth.exe" CALL setpriority "${ cpuPriority . win [ this . cpuPriority ] }"` : `renice -n ${ cpuPriority . unix [ this . cpuPriority ] } -p ${ this . gethService . pid }` ; exec ( command , ( error , stdout , stderr ) => { if ( error ) { this . logger . error ( `cpu:geth:exec error: ${ error }` ) ; return ; } this . logger . debug ( `cpu:geth:stdout: ${ stdout }` ) ; this . logger . debug ( `cpu:geth:stderr: ${ stderr }` ) ; } ) ; } public getCpuPriority ( ) { return this . cpuPriority ; } public enableDownloadEvents ( ) { if ( this . _downloadEventsEnabled ) { return ; } this . downloadManager . wrapper . downloadProgress . on ( event . DOWNLOAD_STARTED , ( ) => { this . emit ( event . DOWNLOAD_STARTED ) ; } ) ; this . downloadManager . wrapper . downloadProgress . on ( event . DOWNLOAD_PROGRESS , ( progress ) => { this . emit ( event . DOWNLOAD_PROGRESS , progress ) ; } ) ; this . downloadManager . wrapper . downloadProgress . on ( event . DOWNLOAD_ERROR , ( error ) => { this . emit ( event . DOWNLOAD_ERROR , error ) ; } ) ; this . _downloadEventsEnabled = true ; } param options returns { Map < any , any > } public setOptions ( options ? ) { let localOptions ; if ( options ) { if ( options . hasOwnProperty ( 'cpu' ) ) { this . setCpuPriority ( options . cpu ) ; delete options . cpu ; } if ( platform === 'Windows_NT' && options . hasOwnProperty ( 'ipcpath' ) ) { options . ipcpath = pathJoin ( '\\\\.\\pipe' , options . ipcpath ) ; } if ( platform !== 'Windows_NT' && options . hasOwnProperty ( 'datadir' ) && ! options . hasOwnProperty ( 'ipcpath' ) ) { options . ipcpath = pathJoin ( options . datadir , 'geth.ipc' ) ; } } if ( this . spawnOptions . size ) { if ( ! options ) { return this . spawnOptions ; } localOptions = options ; } else { localOptions = Object . assign ( { datadir : GethConnector . getDefaultDatadir ( ) , ipcpath : GethConnector . getDefaultIpcPath ( ) , } , event . START_OPTIONS , options ) ; } for ( let option in localOptions ) { if ( localOptions . hasOwnProperty ( option ) ) { this . spawnOptions . set ( option , localOptions [ option ] ) ; } } if ( this . spawnOptions . get ( event . BIN_PATH ) ) { this . setBinPath ( this . spawnOptions . get ( event . BIN_PATH ) ) ; this . spawnOptions . delete ( event . BIN_PATH ) ; } return this . spawnOptions ; } param waitTime returns { Bluebird < U > } public restart ( waitTime = 5000 ) { return Promise . resolve ( this . stop ( ) ) . delay ( waitTime ) . then ( ( ) => this . start ( ) ) ; } param genesisPath param cb writeGenesis ( genesisPath , cb ) { this . _checkBin ( ) . then ( ( binPath ) => { if ( binPath ) { const dataDir = ( this . spawnOptions . get ( 'datadir' ) ) ? this . spawnOptions . get ( 'datadir' ) : '' ; let command = ( dataDir ) ? `--datadir "${ dataDir }"` : '' ; command += ` init "${ genesisPath }"` ; exec ( `"${ binPath }" ${ command }` , ( error , stdout ) => { cb ( error , stdout ) ; } ) ; } } ) ; } private private _flushEvents ( ) { this . web3 . reset ( ) ; this . serviceStatus . api = false ; this . serviceStatus . process = false ; this . serviceStatus . version = '' ; if ( this . socket ) { this . socket . removeAllListeners ( ) ; } if ( ! this . connectedToLocal ) { if ( this . watchers . get ( event . START_FILTER ) ) { this . gethService . stderr . removeListener ( 'data' , this . watchers . get ( event . START_FILTER ) ) ; this . watchers . delete ( event . START_FILTER ) ; } if ( this . watchers . get ( event . INFO_FILTER ) ) { this . gethService . stdout . removeListener ( 'data' , this . watchers . get ( event . INFO_FILTER ) ) ; this . gethService . stderr . removeListener ( 'data' , this . watchers . get ( event . INFO_FILTER ) ) ; this . watchers . delete ( event . INFO_FILTER ) ; } } if ( this . socket ) { this . socket . end ( ) ; } this . socket = null ; return true ; } fires fires returns { Bluebird < U > } private private _checkBin ( ) { return new Promise ( ( resolve , reject ) => { this . downloadManager . check ( ( err , data ) => { if ( err ) { this . logger . error ( err ) ; event this . emit ( event . BINARY_CORRUPTED , err ) ; this . downloadManager . deleteBin ( ) ; return reject ( err ) ; } if ( data . binPath ) { return resolve ( data . binPath ) ; } } ) ; } ) ; } returns { any [ ] } private private _flattenOptions ( ) { const arrayOptions = [ ] ; for ( let [ key , value ] of this . spawnOptions ) { arrayOptions . push ( `--${ key }` ) ; if ( value ) { arrayOptions . push ( value ) ; } } return arrayOptions ; } private _attachEvents ( ) { this . __listenProcess ( ) ; this . _watchGethStd ( ) ; this . _attachListeners ( ) ; } fires fires private private __listenProcess ( ) { this . gethService . once ( 'exit' , ( code , signal ) => { let message ; if ( code ) { message = `geth: exited with code: ${ code }` ; this . logger . error ( message ) ; this . emit ( event . ERROR , message ) ; } else { message = `geth: received signal: ${ signal }` ; this . logger . debug ( message ) ; } this . _flushEvents ( ) ; this . serviceStatus . process = false ; } ) ; this . gethService . once ( 'close' , ( code , signal ) => { this . logger . debug ( 'geth:spawn:close:' , code , signal ) ; event this . emit ( event . STOPPED ) ; } ) ; this . gethService . once ( 'error' , ( code ) => { this . logger . error ( `geth:spawn:error: ${ code }` ) ; this . serviceStatus . process = false ; event this . emit ( event . FAILED , 'gethService' ) ; } ) ; } private private _tailGethLog ( ) { if ( this . watchers . get ( event . START_FILTER ) ) { this . gethService . stderr . removeListener ( 'data' , this . watchers . get ( event . START_FILTER ) ) ; this . watchers . delete ( event . START_FILTER ) ; } const infoFilter = ( data ) => { this . logger . debug ( data . toString ( ) ) ; } ; this . watchers . set ( event . INFO_FILTER , infoFilter ) ; this . gethService . stdout . on ( 'data' , this . watchers . get ( event . INFO_FILTER ) ) ; this . gethService . stderr . on ( 'data' , this . watchers . get ( event . INFO_FILTER ) ) ; } fires fires fires fires private private _watchGethStd ( ) { let started = false ; this . serviceStatus . process = false ; const timeout = setTimeout ( ( ) => { this . gethService . stderr . removeAllListeners ( 'data' ) ; event this . emit ( event . ERROR , 'geth connection timeout' ) ; } , 20000 ) ; const startFilter = ( data ) => { let log = data . toString ( ) ; if ( log . includes ( 'clock seems off' ) ) { event this . emit ( event . TIME_NOT_SYNCED , log ) ; } if ( log . includes ( 'Fatal' ) || log . includes ( 'Synchronisation failed' ) ) { this . logger . error ( `geth:sync:error: ${ log }` ) ; event this . emit ( event . FATAL , log ) ; clearTimeout ( timeout ) ; } if ( log . includes ( 'IPC endpoint opened' ) ) { this . serviceStatus . process = true ; clearTimeout ( timeout ) ; this . _connectToIPC ( ) ; } this . logger . debug ( log ) ; } ; this . watchers . set ( event . START_FILTER , startFilter ) ; this . gethService . stderr . on ( 'data' , this . watchers . get ( event . START_FILTER ) ) ; } listens private private _attachListeners ( ) { this . removeListener ( event . STARTED , this . _tailGethLog ) ; this . removeListener ( event . FATAL , this . socket . end ) ; this . once ( event . STARTED , this . _tailGethLog ) ; this . once ( event . FATAL , ( ) => this . socket ? this . socket . end : null ) ; } fires fires private private _connectToIPC ( ) { this . ipcStream . setProvider ( this . spawnOptions . get ( 'ipcpath' ) , this . socket ) ; this . socket . once ( 'connect' , ( ) => { this . logger . info ( 'Connection to go-ethereum ipc established!' ) ; event this . emit ( event . IPC_CONNECTED ) ; this . _checkRunningSevice ( ) . then ( ( status ) => { if ( status ) { this . serviceStatus . api = true ; } } ) ; event this . emit ( event . STARTED ) ; } ) ; this . socket . once ( 'error' , ( error ) => { this . web3 . reset ( ) ; this . logger . error ( error . message ) ; this . serviceStatus . api = false ; event this . emit ( event . IPC_DISCONNECTED ) ; } ) ; } fires fires fires returns { Bluebird < boolean > } private private _checkRunningSevice ( ) { const requiredVersion = GethBin . requiredVersion ( ) ; const runningServiceProps = [ this . web3 . version . getNodeAsync ( ) , this . web3 . version . getNetworkAsync ( ) , ] ; return Promise . all ( runningServiceProps ) . then ( ( [ buildVersion , networkId ] ) => { let message ; networkId = parseInt ( networkId , 10 ) ; this . serviceStatus . version = buildVersion ; if ( ! buildVersion . includes ( requiredVersion ) ) { message = `required geth version: ${ requiredVersion }, found: ${ buildVersion }, updating executable...` ; this . logger . warn ( message ) ; this . emit ( event . UPDATING_BINARY , message ) ; return this . stop ( ) . delay ( 5000 ) . then ( ( ) => this . downloadManager . deleteBin ( ) ) . then ( ( ) => this . start ( ) ) ; } let netId = this . spawnOptions . get ( 'networkid' ) ; if ( this . spawnOptions . has ( 'testnet' ) ) { netId = 2 ; } if ( netId && networkId !== netId ) { message = `required ethereum network: ${ netId }, found: ${ networkId }` ; this . logger . error ( message ) ; this . emit ( event . FATAL , message ) ; return false ; } event this . emit ( event . ETH_NODE_OK ) ; return true ; } ) ; } }