'use strict'; var web3 = require('web3'); var async = require('async'); var _ = require('lodash'); var os = require('os'); var shelljs = require('shelljs'); var debounce = require('debounce'); var registrar = require('./registrar.js'); var pjson = require('./../package.json'); var chalk = require('chalk'); var Primus = require('primus'), Emitter = require('primus-emit'), Latency = require('primus-spark-latency'), Socket, socket; var ETH_VERSION, NET_VERSION, PROTOCOL_VERSION, API_VERSION, COINBASE; var INSTANCE_NAME = process.env.INSTANCE_NAME; var WS_SECRET = process.env.WS_SECRET || "eth-net-stats-has-a-secret"; var Contract = null; var PENDING_WORKS = true; var MAX_BLOCKS_HISTORY = 40; var UPDATE_INTERVAL = 5000; var PING_INTERVAL = 2000; var MINERS_LIMIT = 5; var MAX_HISTORY_UPDATE = 50; Socket = Primus.createSocket({ transformer: 'websockets', pathname: '/api', timeout: 60000, strategy: 'disconnect,online', plugin: {emitter: Emitter, sparkLatency: Latency} }); if(process.env.NODE_ENV === 'production' && INSTANCE_NAME === "") { INSTANCE_NAME = shelljs.exec('ec2metadata --instance-id', {silent: true}).output; } function Node () { this.info = { name: INSTANCE_NAME || (process.env.EC2_INSTANCE_ID || os.hostname()), contact: (process.env.CONTACT_DETAILS || ""), coinbase: null, node: null, net: null, protocol: null, api: null, port: (process.env.LISTENING_PORT || 30303), os: os.platform(), os_v: os.release(), client: pjson.version, canUpdateHistory: true, }; this.id = _.camelCase(this.info.name); this.stats = { active: false, listening: false, mining: false, hashrate: 0, peers: 0, pending: 0, gasPrice: 0, block: {}, miners: [], uptime: 0 }; this._lastStats = JSON.stringify(this.stats); this._lastFetch = 0; this._tries = 0; this._down = 0; this._lastSent = 0; this._latency = 0; this._Registrar = null; this._knownMiners = []; this._web3 = false; this._socket = false; this.pendingFilter = false; this.chainFilter = false; this.updateInterval = false; this.pingInterval = false; this.connectionInterval = false; this._lastChainLog = 0; this._lastPendingLog = 0; this._called = 0 this.startWeb3Connection(); return this; } Node.prototype.startWeb3Connection = function() { console.info(chalk.blue('==>'), chalk.bold('Starting eth connection')); web3.setProvider( new web3.providers.HttpProvider('http://' + (process.env.RPC_HOST || 'localhost') + ':' + (process.env.RPC_PORT || '8080')) ); this.checkWeb3Connection(); } Node.prototype.checkWeb3Connection = function() { var self = this; if(!this._web3) { try { var tmp = web3.version.client; if( !_.isUndefined(tmp) ) { console.log('eth', tmp); console.info('==>', 'Ethereum connection established'); this._web3 = true; this.init(); return true; } } catch(err) { console.error('xx>', 'Ethereum connection attempt #' + this._called++ + ' failed;'); process.nextTick( function() { self.checkWeb3Connection(); }); } } } Node.prototype.startSocketConnection = function() { console.info("==> Starting socket connection"); socket = new Socket( process.env.WS_SERVER || 'ws://localhost:3000' ); this.setupSockets(); } Node.prototype.setupSockets = function() { var self = this; // Setup events socket.on('open', function open() { console.info('==> The connection has been opened.'); console.info('Trying to login'); socket.emit('hello', { id: self.id, info: self.info, secret: WS_SECRET }); }) .on('ready', function() { self._socket = true; console.info('==> The connection has been established.'); self.updateBlock(); self.update(true); }) .on('data', function incoming(data) { console.info('Received some data', data); }) .on('history', function (data) { console.info('==> Getting history'); var reqHistory = self.getHistory( data ); socket.emit('history', { id: self.id, history: reqHistory }); }) .on('node-pong', function(data) { var latency = Math.ceil( (_.now() - self._latency) / 2 ); socket.emit('latency', { id: self.id, latency: latency }); }) .on('end', function end() { self._socket = false; console.error('xx> Socket connection closed'); }) .on('error', function error(err) { console.error("socket:", err); }) .on('reconnecting', function reconnecting(opts) { console.warn('!!!', 'We are scheduling a reconnect operation', opts); }); } Node.prototype.emit = function(message, payload) { if(this._socket) { try { socket.emit(message, payload); } catch (err) { console.error("socket.emit:", err); } } } Node.prototype.setInactive = function() { this.stats.active = false; this.stats.peers = 0; this.stats.pending = 0; this.stats.mining = false; this.stats.hashrate = 0; this.stats.gasPrice = 0; this.stats.miner = false; this._down++; return this; } Node.prototype.getInfo = function() { var start = _.now(); console.info('==>', 'Getting info'); try { this.info.coinbase = web3.eth.coinbase; this.info.node = web3.version.client; this.info.net = web3.version.network; this.info.protocol = web3.toDecimal(web3.version.ethereum); this.info.api = web3.version.api; console.info('==>', 'Got info in', _.now() - start, 'ms'); console.info(' i ', this.info); return true; } catch (err) { console.error("Couldn't get version"); } return false; } Node.prototype.getBlock = function(number) { var block = { number: 0, hash: '?', difficulty: 0, timestamp: 0, miner: '' }; if( _.isUndefined(number) ) number = "latest"; try { block = web3.eth.getBlock(number, true); if( block.hash != '?' && !_.isUndefined(block.difficulty) && !_.isUndefined(block.totalDifficulty) ) { block.difficulty = web3.toDecimal( block.difficulty ); block.totalDifficulty = web3.toDecimal( block.totalDifficulty ); } } catch (err) { console.error("getBlock(" + number + "):", err); return false; } return block; } Node.prototype.getMinerName = function(miner) { var result = _.find(this._knownMiners, { miner: miner }); if (result !== undefined) { return result.name; } else { if (this._Registrar !== null) { var name = this._Registrar.name(miner); if(name.length > 0) { this._knownMiners.push({ miner: miner, name: name }); return name; } } this._knownMiners.push({ miner: miner, name: false }); } return false; } Node.prototype.uptime = function() { this.stats.uptime = ((this._tries - this._down) / this._tries) * 100; } Node.prototype.getStats = function(forced) { var self = this; var now = _.now(); var lastFetchAgo = now - this._lastFetch; this._lastFetch = now; if (this._socket) this._lastStats = JSON.stringify(this.stats); if (this._web3 && (lastFetchAgo >= UPDATE_INTERVAL || forced === true)) { console.log('==> Getting stats; last update:', lastFetchAgo, '- forced:', (forced === true)); async.parallel({ start: function (callback) { callback(null, _.now()); }, peers: function (callback) { async.nextTick(function () { var peers; try { peers = web3.toDecimal(web3.net.peerCount); } catch (err) { console.error('xx> PeerCount failed: ', err); callback(err, null); } callback(null, peers); }); }, pending: function (callback) { async.nextTick(function () { try { web3.eth.getBlockTransactionCount('pending', callback); } catch (err) { console.error('xx> Pending failed: ', err); callback(err, null); } }); }, mining: function (callback) { async.nextTick(function () { var mining; try { mining = web3.eth.mining; } catch (err) { console.error('xx> Mining failed: ', err); callback(err, null); } callback(null, mining); }); }, hashrate: function (callback) { async.nextTick(function () { var hashrate; try { hashrate = web3.eth.hashrate; } catch (err) { console.error('xx> Hashrate failed: ', err); callback(err, null); } callback(null, hashrate); }); }, gasPrice: function (callback) { async.nextTick(function () { var gasPrice; try { gasPrice = web3.toBigNumber(web3.eth.gasPrice).toString(10); } catch (err) { console.error('xx> gasPrice failed: ', err); callback(err, null); } callback(null, gasPrice); }); }, minerName: function (callback) { async.nextTick(function () { var minerName; try { minerName = self.getMinerName(self.stats.block.miner); } catch (err) { console.error('xx> minerName failed: ', err); callback(err, null); } callback(null, minerName); }); } }, function (err, results) { self._tries++; if (err) { console.error('xx> getStats error: ', err); self.setInactive(); return false; } results.end = _.now(); results.diff = results.end - results.start; // console.log('==> Got getStats results: ', results); console.log('==> Got getStats results in', results.diff, 'ms'); if(results.peers !== null) { self.stats.active = true; self.stats.peers = results.peers; self.stats.pending = results.pending; self.stats.mining = results.mining; self.stats.hashrate = results.hashrate; self.stats.gasPrice = results.gasPrice; } else { self.setInactive(); } self.uptime(); self.sendUpdate(); }); } } Node.prototype.getStatsBlock = function () { if(this._socket) this._lastStats = JSON.stringify(this.stats); if(this._web3) { var start = _.now(); var block = this.getBlock(); var end = _.now(); if( !_.isUndefined(block) && !_.isUndefined(block.number) && !_.isUndefined(block.hash) && block.hash !== '?' ) { console.log("==> Got block:", block.number, 'in', end - start, 'ms'); this.stats.block = block; } else { console.error("xx> getStatsBlock: couldn't fetch block..."); } } this.sendUpdate(); } Node.prototype.getHistory = function (range) { var history = []; var interv = {}; console.time('=== Got history in'); if( _.isUndefined(range) || range === null) { interv = { min: this.stats.block.number - MAX_HISTORY_UPDATE, max: this.stats.block.number - 1 }; } if( !_.isUndefined(range.list) ) { interv = { min: 0, max: range.list.length - 1 }; } for (var i = interv.min; i <= interv.max; i++) { var block = this.getBlock(( !_.isUndefined(range.list) ? range.list[i] : i)); if( block !== null && !_.isUndefined(block.number) ) { history.push( block ); } } console.timeEnd('=== Got history in'); return history.reverse(); } Node.prototype.updateBlock = function() { this.getStatsBlock(); return this; }; Node.prototype.update = function(forced) { this.getStats(forced); return this; }; Node.prototype.changed = function () { var changed = ! _.isEqual( this._lastStats, JSON.stringify(this.stats) ); return changed; } Node.prototype.prepareStats = function () { return { id: this.id, stats: this.stats }; } Node.prototype.sendUpdate = function (force) { if( this.changed() || force ) this.emit('update', this.prepareStats()); } Node.prototype.ping = function() { this._latency = _.now(); this.emit('node-ping', { id: this.id }); }; Node.prototype.setWatches = function() { var self = this; try { this.chainFilter = web3.eth.filter('latest'); this.chainFilter.watch( function (log) { var now = _.now(); var time = now - self._lastChainLog; self._lastChainLog = now; console.log('>>> Chain Filter triggered: ', now, '- last trigger:', time); if(time > 50) { self.updateBlock(); } else { debounce(function() { self.updateBlock(); }, 50); } }); } catch (err) { console.error("Couldn't set up chain filter"); console.error(err); } try { this.pendingFilter = web3.eth.filter('pending'); this.pendingFilter.watch( function (log) { var now = _.now(); var time = now - self._lastPendingLog; self._lastPendingLog = now; console.log('>>> Pending Filter triggered: ', now, '- last trigger:', time); if(time > 50) { self.update(true); } else { debounce(function() { self.update(true); }, 50); } }); } catch (err) { console.error("Couldn't set up pending filter"); console.error(err); } this.updateInterval = setInterval( function(){ self.update(); }, UPDATE_INTERVAL); this.pingInterval = setInterval( function(){ self.ping(); }, PING_INTERVAL); } Node.prototype.installContract = function() { var start = _.now(); try { Contract = web3.eth.contract( registrar.desc ); this._Registrar = new Contract( registrar.address ); console.log('==>', 'Installed Registrar contract in', _.now() - start, 'ms'); } catch (err) { console.error("!!!", "Couldn't set up registrar contract"); console.error(err); } } Node.prototype.init = function() { // Fetch node info this.getInfo(); // Install Registrar contract this.installContract(); // Start socket connection this.startSocketConnection(); // Set filters this.setWatches(); } Node.prototype.stop = function() { if(this._socket) socket.end(); if(this.updateInterval) clearInterval(this.updateInterval); if(this.pingInterval) clearInterval(this.pingInterval); web3.reset(); } module.exports = Node;