Pub sub blocks (#6139)

* fun start

* playin

* linting

* cleanup kk 1

* cleanup kk 2

* package

* oops

* grumbles

* lint

* lint 2

* all-possible-cases

* conflict resolution

* conflict resolution

* ready

* lint

* remove package-lock

* Restart subscription on transport close.

* Bring back the formatter.

* Fix formatter.

hw/src/lib.rs

@@ -155,7 +155,6 @@ impl HardwareWalletManager {
 		self.ledger.lock().set_key_path(key_path);
 	}
 
-
 	/// List connected wallets. This only returns wallets that are ready to be used.
 	pub fn list_wallets(&self) -> Vec<WalletInfo> {
 		self.ledger.lock().list_devices()

js/package-lock.json

@@ -7722,7 +7722,7 @@
     "minimatch": {
       "version": "3.0.4",
       "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-3.0.4.tgz",
-      "integrity": "sha1-UWbihkV/AzBgZL5Ul+jbsMPTIIM=",
+      "integrity": "sha512-yJHVQEhyqPLUTgt9B83PXu6W3rx4MvvHvSUvToogpwoGDOUQ+yDrR0HRot+yOCdCO7u4hX3pWft6kWBBcqh0UA==",
       "requires": {
         "brace-expansion": "1.1.8"
       }
@@ -10081,7 +10081,7 @@
     "react-qr-reader": {
       "version": "1.1.3",
       "resolved": "https://registry.npmjs.org/react-qr-reader/-/react-qr-reader-1.1.3.tgz",
-      "integrity": "sha1-dDmnZvyZPLj17u/HLCnblh1AswI=",
+      "integrity": "sha512-ruBF8KaSwUW9nbzjO4rA7/HOCGYZuNUz9od7uBRy8SRBi24nwxWWmwa2z8R6vPGDRglA0y2Qk1aVBuC1olTnHw==",
       "requires": {
         "jsqr": "git+https://github.com/JodusNodus/jsQR.git#5ba1acefa1cbb9b2bc92b49f503f2674e2ec212b",
         "prop-types": "15.5.10",

js/src/api/pubsub/parity/parity.js

@@ -183,16 +183,17 @@ export default class Parity extends PubsubBase {
 
   localTransactions (callback) {
     return this.addListener(this._api, 'parity_localTransactions', (error, transactions) => {
-      error
-        ? callback(error)
-        : callback(null, transactions => {
+      if (error) {
+        return callback(error);
+      }
+
       Object.values(transactions)
         .filter(tx => tx.transaction)
         .map(tx => {
           tx.transaction = outTransaction(tx.transaction);
         });
-          return transactions;
-        });
+
+      callback(null, transactions);
     });
   }
 

js/src/views/Signer/store.js

@@ -25,11 +25,13 @@ export default class SignerStore {
 
   constructor (api, withLocalTransactions = false, externalLink = '') {
     this._api = api;
-    this._timeoutId = 0;
     this.externalLink = externalLink;
 
     if (withLocalTransactions) {
-      this.fetchLocalTransactions();
+      this._api.transport.on('close', () => {
+        this.subscribeLocalTransactions();
+      });
+      this.subscribeLocalTransactions();
     }
   }
 
@@ -49,9 +51,7 @@ export default class SignerStore {
   }
 
   @action unsubscribe () {
-    if (this._timeoutId) {
-      clearTimeout(this._timeoutId);
-    }
+    this.subscription.then(id => this._api.pubsub.unsubscribe([id]));
   }
 
   fetchBalance (address) {
@@ -87,21 +87,17 @@ export default class SignerStore {
       });
   }
 
-  fetchLocalTransactions = () => {
-    const nextTimeout = () => {
-      this._timeoutId = setTimeout(this.fetchLocalTransactions, 1500);
-    };
-
-    this._api.parity
-      .localTransactions()
-      .then((localTransactions) => {
+  subscribeLocalTransactions = () => {
+    this.subscription = this._api.pubsub.parity.localTransactions((error, transactions) => {
+      if (error) {
+        console.warn('subscribeLocalTransactions', error);
+        return;
+      }
       const keys = Object
-          .keys(localTransactions)
-          .filter((key) => localTransactions[key].status !== 'canceled');
+        .keys(transactions)
+        .filter((key) => transactions[key].status !== 'canceled');
 
       this.setLocalHashes(keys);
-      })
-      .then(() => nextTimeout())
-      .catch(() => nextTimeout());
+    });
   }
 }

js/src/views/Status/NodeStatus/nodeStatus.js

@@ -18,7 +18,7 @@ import bytes from 'bytes';
 import moment from 'moment';
 import React, { Component, PropTypes } from 'react';
 import { FormattedMessage } from 'react-intl';
-import { connect } from 'react-redux';
+import { observer } from 'mobx-react';
 
 import { Container, ContainerTitle, Input } from '~/ui';
 
@@ -27,18 +27,12 @@ import StatusStore from './store';
 
 import styles from './nodeStatus.css';
 
+@observer
 class NodeStatus extends Component {
   static contextTypes = {
     api: PropTypes.object.isRequired
   };
 
-  static propTypes = {
-    blockNumber: PropTypes.object,
-    blockTimestamp: PropTypes.object,
-    netChain: PropTypes.string,
-    netPeers: PropTypes.object
-  };
-
   statusStore = new StatusStore(this.context.api);
 
   componentWillMount () {
@@ -50,8 +44,7 @@ class NodeStatus extends Component {
   }
 
   render () {
-    const { blockNumber, blockTimestamp, netPeers } = this.props;
-    const { hashrate } = this.statusStore;
+    const { blockNumber, blockTimestamp, netPeers, hashrate } = this.statusStore;
 
     if (!netPeers || !blockNumber) {
       return null;
@@ -157,8 +150,7 @@ class NodeStatus extends Component {
   }
 
   renderSettings () {
-    const { netChain } = this.props;
-    const { enode, rpcSettings, netPort = '' } = this.statusStore;
+    const { chain, enode, rpcSettings, netPort = '' } = this.statusStore;
 
     if (!rpcSettings) {
       return null;
@@ -185,7 +177,7 @@ class NodeStatus extends Component {
               defaultMessage='chain'
             />
           }
-          value={ netChain }
+          value={ chain }
         />
         <div className={ styles.row }>
           <div className={ styles.col6 }>
@@ -279,23 +271,4 @@ class NodeStatus extends Component {
   }
 }
 
-function mapStateToProps (state) {
-  const {
-    blockNumber,
-    blockTimestamp,
-    netChain,
-    netPeers
-  } = state.nodeStatus;
-
-  return {
-    blockNumber,
-    blockTimestamp,
-    netChain,
-    netPeers
-  };
-}
-
-export default connect(
-  mapStateToProps,
-  null
-)(NodeStatus);
+export default NodeStatus;

js/src/views/Status/NodeStatus/store.js

@@ -15,11 +15,15 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 import BigNumber from 'bignumber.js';
-import { action, observable, transaction } from 'mobx';
+import { action, observable } from 'mobx';
 
 export default class StatusStore {
   @observable defaultExtraData = '';
   @observable enode = '';
+  @observable blockNumber = new BigNumber(0);
+  @observable blockTimestamp = new Date();
+  @observable chain = '';
+  @observable netPeers = new BigNumber(0);
   @observable hashrate = new BigNumber(0);
   @observable netPort = new BigNumber(0);
   @observable nodeName = '';
@@ -35,39 +39,68 @@ export default class StatusStore {
 
   constructor (api) {
     this.api = api;
+    this.api.transport.on('close', () => {
+      if (this.isPolling) {
+        this.startPolling();
+      }
+    });
   }
 
-  @action setLongStatus ({ defaultExtraData, enode, netPort, rpcSettings }) {
-    transaction(() => {
+  @action setStatuses ({ chain, defaultExtraData, enode, netPeers, netPort, rpcSettings, hashrate }) {
+    this.chain = chain;
     this.defaultExtraData = defaultExtraData;
     this.enode = enode;
+    this.netPeers = netPeers;
     this.netPort = netPort;
     this.rpcSettings = rpcSettings;
-    });
-  }
-
-  @action setStatus ({ hashrate }) {
-    transaction(() => {
     this.hashrate = hashrate;
-    });
   }
 
   @action setMinerSettings ({ coinbase, extraData, gasFloorTarget, minGasPrice }) {
-    transaction(() => {
     this.coinbase = coinbase;
     this.extraData = extraData;
     this.gasFloorTarget = gasFloorTarget;
     this.minGasPrice = minGasPrice;
-    });
-  }
-
-  startPolling () {
-    this._pollStatus();
-    this._pollLongStatus();
   }
 
   stopPolling () {
-    Object.keys(this._timeoutIds).forEach((key) => clearTimeout(this._timeoutIds[key]));
+    this.isPolling = false;
+    this.subscription.then(id => this.api.pubsub.unsubscribe([id]));
+  }
+
+  startPolling () {
+    this.isPolling = true;
+    this.subscription = this.api.pubsub.parity.getBlockHeaderByNumber((error, block) => {
+      if (error) {
+        console.warn('_startPolling', error);
+        return;
+      }
+      this.subscribed = true;
+      this.blockNumber = block.number;
+      this.blockTimestamp = block.timestamp;
+      this._pollMinerSettings();
+      Promise
+        .all([
+          this.api.parity.chain(),
+          this.api.parity.defaultExtraData(),
+          this.api.parity.enode().then((enode) => enode).catch(() => '-'),
+          this.api.parity.netPeers(),
+          this.api.parity.netPort(),
+          this.api.parity.rpcSettings(),
+          this.api.eth.hashrate()
+        ])
+        .then(([
+          chain, defaultExtraData, enode, netPeers, netPort, rpcSettings, hashrate
+        ]) => {
+          this.setStatuses({
+            chain, defaultExtraData, enode, netPeers, netPort, rpcSettings, hashrate
+          });
+        })
+        .catch((error) => {
+          console.error('_pollStatuses', error);
+          return;
+        });
+    });
   }
 
   /**
@@ -100,60 +133,6 @@ export default class StatusStore {
       });
   }
 
-  _pollStatus () {
-    const nextTimeout = (timeout = 1000) => {
-      clearTimeout(this._timeoutIds.short);
-      this._timeoutIds.short = setTimeout(() => this._pollStatus(), timeout);
-    };
-
-    return Promise
-      .all([
-        this.api.eth.hashrate()
-      ])
-      .then(([
-        hashrate
-      ]) => {
-        this.setStatus({
-          hashrate
-        });
-      })
-      .catch((error) => {
-        console.error('_pollStatus', error);
-      })
-      .then(() => {
-        nextTimeout();
-      });
-  }
-
-  _pollLongStatus () {
-    const nextTimeout = (timeout = 30000) => {
-      clearTimeout(this._timeoutIds.long);
-      this._timeoutIds.long = setTimeout(() => this._pollLongStatus(), timeout);
-    };
-
-    this._pollMinerSettings();
-    return Promise
-      .all([
-        this.api.parity.defaultExtraData(),
-        this.api.parity.enode().then((enode) => enode).catch(() => '-'),
-        this.api.parity.netPort(),
-        this.api.parity.rpcSettings()
-      ])
-      .then(([
-        defaultExtraData, enode, netPort, rpcSettings
-      ]) => {
-        this.setLongStatus({
-          defaultExtraData, enode, netPort, rpcSettings
-        });
-      })
-      .catch((error) => {
-        console.error('_pollLongStatus', error);
-      })
-      .then(() => {
-        nextTimeout();
-      });
-  }
-
   handleUpdateSetting = () => {
     return this._pollMinerSettings();
   };