make it work

lib/cmds/blockchain/miner.js

@@ -1,3 +1,4 @@
+const async = require('async');
 const NetcatClient = require('netcat/client');
 
 //Constants
@@ -7,7 +8,9 @@ const getHashRate = 'miner_getHashrate';
 const getCoinbase = 'eth_coinbase';
 const getBalance = 'eth_getBalance';
 const newBlockFilter = 'eth_newBlockFilter';
+const pendingBlockFilter = 'eth_newPendingTransactionFilter';
 const getChanges = 'eth_getFilterChanges';
+const getBlockCount = 'eth_getBlockTransactionCountByNumber';
 
 class GethMiner {
   constructor() {
@@ -15,6 +18,13 @@ class GethMiner {
     // TODO: Find a way to load mining config from YML.
     // In the meantime, just set an empty config object
     this.config = {};
+    self.interval = null;
+    self.callback = null;
+
+    self.commandQueue = async.queue((task, callback) => {
+      self.callback = callback;
+      self.client.send(JSON.stringify({"jsonrpc": "2.0", "method": task.method, "params": task.params || [], "id": 1}));
+    }, 1);
 
     const defaults = {
       interval_ms: 15000,
@@ -44,7 +54,7 @@ class GethMiner {
     this.client.unixSocket(ipcPath)
       .enc('utf8')
       .connect()
-      .on('data', function(response){
+      .on('data', (response) => {
         try {
           response = JSON.parse(response);
         } catch (e) {
@@ -53,7 +63,6 @@ class GethMiner {
         }
         if (self.callback) {
           self.callback(response.error, response.result);
-          self.callback = null;
         }
       });
 
@@ -67,8 +76,8 @@ class GethMiner {
           console.error(err);
           return;
         }
-        if (this.config.mine_periodically) self.start_periodic_mining();
+        if (self.config.mine_periodically) self.start_periodic_mining();
-        if (this.config.mine_pending_txns) self.start_transaction_mining();
+        if (self.config.mine_pending_txns) self.start_transaction_mining();
       });
     });
 
@@ -79,21 +88,21 @@ class GethMiner {
       callback = params;
       params = [];
     }
-    if (callback) {
+    if (!callback) {
-      this.callback = callback;
+      callback = function() {};
     }
-    this.client.send(JSON.stringify({"jsonrpc": "2.0", "method": method, "params": params || [], "id": 1}));
+    this.commandQueue.push({method, params: params || []}, callback);
   }
 
   getCoinbase(callback) {
     if (this.coinbase) {
       return callback(null, this.coinbase);
     }
-    this.sendCommand(getCoinbase, (err, response) => {
+    this.sendCommand(getCoinbase, (err, result) => {
       if (err) {
         return callback(err);
       }
-      this.coinbase = response.result;
+      this.coinbase = result;
       if (!this.coinbase) {
         return callback('Failed getting coinbase account');
       }
@@ -107,7 +116,7 @@ class GethMiner {
       if (err) {
         return callback(err);
       }
-      self.sendCommand(getBalance, [coinbase], (err, result) => {
+      self.sendCommand(getBalance, [coinbase, 'latest'], (err, result) => {
         if (err) {
           return callback(err);
         }
@@ -116,27 +125,44 @@ class GethMiner {
     });
   }
 
-  watchNewBlocks() {
+  watchBlocks(filterCommand, callback) {
     const self = this;
-    self.sendCommand(newBlockFilter, (err, filterId) => {
+    self.sendCommand(filterCommand, (err, filterId) => {
+      if (err) {
+        callback(err);
+        return;
+      }
+      self.interval = setInterval(() => {
+        self.sendCommand(getChanges, [filterId], (err, changes) => {
           if (err) {
             console.error(err);
             return;
           }
-      const interval = setInterval(() => {
-        self.sendCommand(getChanges, [filterId], (err, changes) => {
           if (!changes || !changes.length) {
             return;
           }
+          callback(null, changes);
+        });
+      }, 1000);
+    });
+  }
+
+  mineUntilFunded(callback) {
+    const self = this;
+    this.sendCommand(minerStart);
+    self.watchBlocks(newBlockFilter, (err) => {
+      if (err) {
+        console.error(err);
+        return;
+      }
       self.accountFunded((err, funded) => {
         if (funded) {
-              clearTimeout(interval);
+          clearTimeout(self.interval);
           self.sendCommand(minerStop);
+          callback();
         }
       });
     });
-      }, 1000);
-    });
   }
 
   fundAccount(callback) {
@@ -151,64 +177,94 @@ class GethMiner {
       }
 
       console.log("== Funding account");
-      this.sendCommand(minerStart);
+      self.mineUntilFunded(callback);
-
-      self.watchNewBlocks();
     });
   }
 
-  pendingTransactions() {
+  pendingTransactions(callback) {
-    if (web3.eth.pendingTransactions === undefined || web3.eth.pendingTransactions === null) {
+    const self = this;
-      return txpool.status.pending || txpool.status.queued;
+    self.sendCommand(getBlockCount, ['pending'], (err, hexCount) => {
-    }
+      if (err) {
-    else if (typeof web3.eth.pendingTransactions === "function") {
+        return callback(err);
-      return web3.eth.pendingTransactions().length > 0;
-    }
-    else {
-      return web3.eth.pendingTransactions.length > 0 || web3.eth.getBlock('pending').transactions.length > 0;
       }
+      callback(null, parseInt(hexCount, 16));
+    });
   }
 
   start_periodic_mining() {
     const self = this;
+    const WAIT = 'wait';
     let last_mined_ms = Date.now();
     let timeout_set = false;
+    let next_block_in_ms;
 
     self.sendCommand(minerStart);
-    web3.eth.filter("latest").watch(function () {
+    self.watchBlocks(newBlockFilter, (err) => {
-      if ((self.config.mine_pending_txns && self.pendingTransactions()) || timeout_set) {
+      if (err) {
+        console.error(err);
         return;
       }
-
+      if (timeout_set) {
+        return;
+      }
+      async.waterfall([
+        function checkPendingTransactions(next) {
+          if (!self.config.mine_pending_txns) {
+            return next();
+          }
+          self.pendingTransactions((err, count) => {
+            if (err) {
+              return next(err);
+            }
+            if (count) {
+              return next(WAIT);
+            }
+            next();
+          });
+        },
+        function stopMiner(next) {
           timeout_set = true;
 
           const now = Date.now();
           const ms_since_block = now - last_mined_ms;
           last_mined_ms = now;
 
-      let next_block_in_ms;
-
           if (ms_since_block > self.config.interval_ms) {
             next_block_in_ms = 0;
           } else {
             next_block_in_ms = (self.config.interval_ms - ms_since_block);
           }
-
           self.sendCommand(minerStop);
           console.log("== Looking for next block in " + next_block_in_ms + "ms");
-
+          next();
+        },
+        function startAfterTimeout(next) {
           setTimeout(function () {
             console.log("== Looking for next block");
             timeout_set = false;
             //miner_obj.start(config.threads);
             self.sendCommand(minerStart);
+            next();
           }, next_block_in_ms);
+        }
+      ], (err) => {
+        if (err === WAIT) {
+          return;
+        }
+        if (err) {
+          console.error(err);
+        }
+      });
     });
   }
 
   start_transaction_mining() {
     const self = this;
-    web3.eth.filter("pending").watch(function () {
+    self.watchBlocks(pendingBlockFilter, (err) => {
+      if (err) {
+        console.error(err);
+        return;
+      }
       self.sendCommand(getHashRate, (err, result) => {
         if (result > 0) return;
 
@@ -217,14 +273,25 @@ class GethMiner {
       });
     });
 
+
     if (self.config.mine_periodically) return;
 
-    web3.eth.filter("latest").watch(function () {
+    self.watchBlocks(newBlockFilter, (err) => {
-      if (!self.pendingTransactions()) {
+      if (err) {
+        console.error(err);
+        return;
+      }
+      self.pendingTransactions((err, count) => {
+        if (err) {
+          console.error(err);
+          return;
+        }
+        if (!count) {
           console.log("== No transactions left. Stopping miner...");
           self.sendCommand(minerStop);
         }
       });
+    });
   }
 }