https://chronobank.atlassian.net/browse/MD-973 - middleware eth / move to web3 1.0 / block processor

Author: egor

index.js

@@ -132,9 +132,17 @@ const init = async () => {
   blockWatchingService.events.on('block', blockEventCallback);
   blockWatchingService.events.on('tx', txEventCallback);
 
-  await blockWatchingService.startSync();
+  //await blockWatchingService.startSync();
 };
 
+
+/*
+process.on('unhandledRejection', (reason, promise) => {//todo remove
+  console.log('Unhandled Rejection at:', reason.stack || reason)
+  process.exit(0)
+})
+*/
+
 module.exports = init().catch(err => {
   log.error(err);
   process.exit(0);

package.json

@@ -16,7 +16,7 @@
     "require-all": "^2.2.0",
     "semaphore": "^1.1.0",
     "uniqid": "^4.1.1",
-    "web3": "^0.19.0"
+    "web3": "^1.0.0-beta.36"
   },
   "scripts": {
     "lint": "node ./node_modules/eslint/bin/eslint.js -c .eslintrc.js --ext .js --fix --ignore-path .eslintignore .",

services/blockWatchingService.js

@@ -45,7 +45,7 @@ class BlockWatchingService {
     this.isSyncing = true;
     let web3 = await providerService.get();
 
-    const pendingBlock = await Promise.promisify(web3.eth.getBlock)('pending').timeout(5000);
+    const pendingBlock = await web3.eth.getBlock('pending');
 
     if (!pendingBlock)
       await removeUnconfirmedTxs();
@@ -106,7 +106,7 @@ class BlockWatchingService {
   async unconfirmedTxEvent (hash) {
 
     let web3 = await providerService.get();
-    let tx = await Promise.promisify(web3.eth.getTransaction)(hash);
+    let tx = await web3.eth.getTransaction(hash);
 
     if (!_.has(tx, 'hash'))
       return;
@@ -137,13 +137,13 @@ class BlockWatchingService {
   async processBlock () {
 
     let web3 = await providerService.get();
-    const block = await Promise.promisify(web3.eth.getBlockNumber)().timeout(2000).catch(() => 0);
+    const block = await web3.eth.getBlockNumber().catch(() => 0);
 
     if (block === this.currentHeight - 1)
       return Promise.reject({code: 0});
 
     const lastBlock = this.currentHeight === 0 ? null :
-      await Promise.promisify(web3.eth.getBlock)(this.currentHeight - 1, false).timeout(60000).catch(() => null);
+      await web3.eth.getBlock(this.currentHeight - 1, false);
 
 
     if (_.get(lastBlock, 'hash')) {

services/providerService.js

@@ -42,23 +42,29 @@ class providerService {
    */
   makeWeb3FromProviderURI (providerURI) {
 
-    const provider = /^http/.test(providerURI) ?
-      new Web3.providers.HttpProvider(providerURI) :
-      new Web3.providers.IpcProvider(`${/^win/.test(process.platform) ? '\\\\.\\pipe\\' : ''}${providerURI}`, net);
+    if (/^http/.test(providerURI) || /^ws/.test(providerURI))
+      return new Web3(providerURI);
 
-    const web3 = new Web3();
-    web3.setProvider(provider);
-    return web3;
+    providerURI = `${/^win/.test(process.platform) ? '\\\\.\\pipe\\' : ''}${providerURI}`;
+    return new Web3(providerURI, net);
   }
 
   /** @function
    * @description reset the current connection
    * @return {Promise<void>}
    */
   async resetConnector () {
-    await this.connector.reset();
+
+    if (this.filter) {
+      //await Promise.promisify(this.filter.unsubscribe.bind(this.connector))();
+      await new Promise(res => this.filter.unsubscribe(res));
+      this.filter = null;
+    }
+
+    if (_.has(this.connector, 'currentProvider.connection.close'))
+      this.connector.currentProvider.connection.close();
     this.switchConnector();
-    this.events.emit('disconnected');
+
   }
 
   /**
@@ -68,16 +74,20 @@ class providerService {
    */
   async switchConnector () {
 
+    console.log('switching connector')
     const providerURI = await Promise.any(config.web3.providers.map(async providerURI => {
       const web3 = this.makeWeb3FromProviderURI(providerURI);
-      await Promise.promisify(web3.eth.getBlockNumber)().timeout(5000);
-      web3.reset();
+      await web3.eth.getBlockNumber();
+      if (_.has(web3, 'currentProvider.connection.close'))
+        web3.currentProvider.connection.close();
       return providerURI;
     })).catch(() => {
       log.error('no available connection!');
-      process.exit(0);
+      process.exit(0); //todo move to root
     });
 
+    console.log('switched connector')
+
     const fullProviderURI = !/^http/.test(providerURI) ? `${/^win/.test(process.platform) ? '\\\\.\\pipe\\' : ''}${providerURI}` : providerURI;
     const currentProviderURI = this.connector ? this.connector.currentProvider.path || this.connector.currentProvider.host : '';
 
@@ -87,31 +97,28 @@ class providerService {
     this.connector = this.makeWeb3FromProviderURI(providerURI);
 
     if (_.get(this.connector.currentProvider, 'connection')) {
-      this.connector.currentProvider.connection.on('end', () => this.resetConnector());
-      this.connector.currentProvider.connection.on('error', () => this.resetConnector());
-    } else 
+
+/*      this.connector.currentProvider.connection.on('end', () => this.resetConnector());
+      this.connector.currentProvider.connection.on('error', () => this.resetConnector());*/
+      this.connector.currentProvider.connection.onerror(()=>this.resetConnector());
+      this.connector.currentProvider.connection.onclose(()=>this.resetConnector());
+
+
+    } else
       this.pingIntervalId = setInterval(async () => {
 
-        const isConnected = await new Promise((res, rej) => {
-          this.connector.currentProvider.sendAsync({
-            id: 9999999999,
-            jsonrpc: '2.0',
-            method: 'net_listening',
-            params: []
-          }, (err, result) => err ? rej(err) : res(result.result));
-        });
+        const isConnected = await this.connector.eth.getProtocolVersion().catch(() => null);
 
         if (!isConnected) {
           clearInterval(this.pingIntervalId);
           this.resetConnector();
         }
       }, 5000);
-    
 
-    this.filter = this.connector.eth.filter('pending');
-    this.filter.watch((err, result) => {
-      if (!err)
-        this.events.emit('unconfirmedTx', result);
+
+    this.filter = this.connector.eth.subscribe('pendingTransactions');
+    this.filter.on('data', (transaction) => {
+      this.events.emit('unconfirmedTx', transaction);
     });
 
     this.events.emit('provider_set');
@@ -126,6 +133,8 @@ class providerService {
    */
   async switchConnectorSafe () {
 
+    console.log('going to switch connector')
+
     return new Promise(res => {
       sem.take(async () => {
         await this.switchConnector();
@@ -141,7 +150,12 @@ class providerService {
    * @return {Promise<*|bluebird>}
    */
   async get () {
-    return this.connector && this.connector.isConnected() ? this.connector : await this.switchConnectorSafe();
+
+    if(this.connector){
+      console.log('is listening: ', await this.connector.eth.getProtocolVersion().catch(()=>null))
+    }
+
+    return this.connector && await this.connector.eth.getProtocolVersion().catch(() => false) ? this.connector : await this.switchConnectorSafe();
   }
 
 }

services/syncCacheService.js

@@ -93,21 +93,26 @@ class SyncCacheService {
    */
   async runPeer(bucket) {
 
+    console.log('getting web3 instance')
     let web3 = await providerService.get();
 
-    const lastBlock = await Promise.promisify(web3.eth.getBlock)(_.last(bucket), false).timeout(60000);
-
+    const lastBlock = await web3.eth.getBlock(_.last(bucket), false);
 
     if (!lastBlock || (_.last(bucket) !== 0 && !lastBlock.number))
       return await Promise.delay(10000);
 
     log.info(`web3 provider took chuck of blocks ${bucket[0]} - ${_.last(bucket)}`);
 
     await Promise.mapSeries(bucket, async (blockNumber) => {
+      console.log('getting block')
       const block = await getBlock(blockNumber);
+      console.log('adding block')
       await addBlock(block);
 
+      console.log('remove block from bucket')
       _.pull(bucket, blockNumber);
+
+      console.log('emitting event...')
       this.events.emit('block', block);
     });
 

utils/blocks/addBlock.js

@@ -9,7 +9,6 @@ const bunyan = require('bunyan'),
   removeUnconfirmedTxs = require('../txs/removeUnconfirmedTxs'),
   crypto = require('crypto'),
   sem = require('semaphore')(3),
-  Promise = require('bluebird'),
   config = require('../../config'),
   models = require('../../models'),
   BigNumber = require('bignumber.js'),

utils/blocks/allocateBlockBuckets.js

@@ -60,7 +60,7 @@ module.exports = async function () {
 
   let web3 = await providerService.get();
 
-  let currentNodeHeight = await Promise.promisify(web3.eth.getBlockNumber)().timeout(10000).catch(() => -1);
+  let currentNodeHeight = await web3.eth.getBlockNumber().catch(() => -1);
   currentNodeHeight = parseInt(currentNodeHeight);
 
   if (currentNodeHeight === -1)

utils/blocks/getBlock.js

@@ -4,8 +4,7 @@
  * @author Egor Zuev <[email protected]>
  */
 
-const Promise = require('bluebird'),
-  providerService = require('../../services/providerService'),
+const providerService = require('../../services/providerService'),
   _ = require('lodash');
 
 /**
@@ -23,9 +22,9 @@ module.exports = async (blockNumber) => {
 
   let web3 = await providerService.get();
 
-  let rawBlock = await Promise.promisify(web3.eth.getBlock)(blockNumber, true).timeout(10000);
+  let rawBlock = await web3.eth.getBlock(blockNumber, true);
 
-  if(!rawBlock)
+  if (!rawBlock)
     return Promise.reject({code: 2});
 
   rawBlock.uncleAmount = rawBlock.uncles.length;
@@ -35,10 +34,7 @@ module.exports = async (blockNumber) => {
     return rawBlock;
   }
 
-  let logs = await new Promise((res, rej) =>
-    web3.eth.filter({fromBlock: blockNumber, toBlock: blockNumber})
-      .get((err, result) => err ? rej(err) : res(result))
-  ).timeout(30000);
+  let logs = await web3.eth.getPastLogs({fromBlock: blockNumber, toBlock: blockNumber});
 
   rawBlock.transactions = rawBlock.transactions.map(tx => {
     tx.logs = _.chain(logs)

utils/txs/removeUnconfirmedTxs.js

@@ -12,7 +12,7 @@ module.exports = async () => {
 
   let web3 = await providerService.get();
 
-  const pendingBlock = await Promise.promisify(web3.eth.getBlock)('pending').timeout(5000);
+  const pendingBlock = await web3.eth.getBlock('pending');
 
   if (!_.get(pendingBlock, 'transactions', []).length)
     return;