Skip to content
This repository has been archived by the owner on Nov 6, 2020. It is now read-only.

Commit

Permalink
Merge pull request #3587 from ethcore/ng-ws-improved
Browse files Browse the repository at this point in the history
Work around WS in UI
  • Loading branch information
gavofyork committed Nov 23, 2016
2 parents 820e291 + 0c3d87f commit 7e800b7
Show file tree
Hide file tree
Showing 6 changed files with 198 additions and 59 deletions.
104 changes: 94 additions & 10 deletions js/src/api/transport/ws/ws.js
Original file line number Diff line number Diff line change
Expand Up @@ -29,21 +29,33 @@ export default class Ws extends JsonRpcBase {
this._token = token;
this._messages = {};

this._connecting = true;
this._connecting = false;
this._connected = false;
this._lastError = null;
this._autoConnect = false;
this._autoConnect = true;
this._retries = 0;
this._reconnectTimeoutId = null;

this._connect();
}

updateToken (token) {
this._token = token;
this._autoConnect = false;
this._autoConnect = true;

this._connect();
}

_connect () {
if (this._connecting) {
return;
}

if (this._reconnectTimeoutId) {
window.clearTimeout(this._reconnectTimeoutId);
this._reconnectTimeoutId = null;
}

const time = parseInt(new Date().getTime() / 1000, 10);
const sha3 = keccak_256(`${this._token}:${time}`);
const hash = `${sha3}_${time}`;
Expand All @@ -53,6 +65,7 @@ export default class Ws extends JsonRpcBase {
this._ws.onopen = null;
this._ws.onclose = null;
this._ws.onmessage = null;
this._ws.close();
this._ws = null;
}

Expand All @@ -65,32 +78,75 @@ export default class Ws extends JsonRpcBase {
this._ws.onopen = this._onOpen;
this._ws.onclose = this._onClose;
this._ws.onmessage = this._onMessage;

// Get counts in dev mode
if (process.env.NODE_ENV === 'development') {
this._count = 0;
this._lastCount = {
timestamp: Date.now(),
count: 0
};

window.setInterval(() => {
const n = this._count - this._lastCount.count;
const t = (Date.now() - this._lastCount.timestamp) / 1000;
const s = Math.round(1000 * n / t) / 1000;

if (this._debug) {
console.log('::parityWS', `speed: ${s} req/s`, `count: ${this._count}`);
}
}, 5000);

window._parityWS = this;
}
}

_onOpen = (event) => {
console.log('ws:onOpen', event);
this._connected = true;
this._connecting = false;
this._autoConnect = true;
this._retries = 0;

Object.keys(this._messages)
.filter((id) => this._messages[id].queued)
.forEach(this._send);
}

_onClose = (event) => {
console.log('ws:onClose', event);
this._connected = false;
this._connecting = false;

this._lastError = event;

if (this._autoConnect) {
setTimeout(() => this._connect(), 500);
const timeout = this.retryTimeout;

const time = timeout < 1000
? Math.round(timeout) + 'ms'
: (Math.round(timeout / 10) / 100) + 's';

console.log('ws:onClose', `trying again in ${time}...`);

this._reconnectTimeoutId = setTimeout(() => {
this._connect();
}, timeout);

return;
}

console.log('ws:onClose', event);
}

_onError = (event) => {
console.error('ws:onError', event);
this._lastError = event;
// Only print error if the WS is connected
// ie. don't print if error == closed
window.setTimeout(() => {
if (this._connected) {
console.error('ws:onError', event);
this._lastError = event;
}
}, 50);
}

_onMessage = (event) => {
Expand Down Expand Up @@ -127,11 +183,16 @@ export default class Ws extends JsonRpcBase {
_send = (id) => {
const message = this._messages[id];

message.queued = !this._connected;

if (this._connected) {
this._ws.send(message.json);
if (process.env.NODE_ENV === 'development') {
this._count++;
}

return this._ws.send(message.json);
}

message.queued = !this._connected;
message.timestamp = Date.now();
}

execute (method, ...params) {
Expand Down Expand Up @@ -159,4 +220,27 @@ export default class Ws extends JsonRpcBase {
get lastError () {
return this._lastError;
}

/**
* Exponential Timeout for Retries
*
* @see http://dthain.blogspot.de/2009/02/exponential-backoff-in-distributed.html
*/
get retryTimeout () {
// R between 1 and 2
const R = Math.random() + 1;
// Initial timeout (100ms)
const T = 100;
// Exponential Factor
const F = 2;
// Max timeout (4s)
const M = 4000;
// Current number of retries
const N = this._retries;

// Increase retries number
this._retries++;

return Math.min(R * T * Math.pow(F, N), M);
}
}
40 changes: 27 additions & 13 deletions js/src/contracts/registry.js
Original file line number Diff line number Diff line change
Expand Up @@ -21,25 +21,39 @@ export default class Registry {
this._api = api;
this._contracts = [];
this._instance = null;
this._fetching = false;
this._queue = [];

this.getInstance();
}

getInstance () {
return new Promise((resolve, reject) => {
if (this._instance) {
resolve(this._instance);
return;
}
if (this._instance) {
return Promise.resolve(this._instance);
}

this._api.parity
.registryAddress()
.then((address) => {
this._instance = this._api.newContract(abis.registry, address).instance;
resolve(this._instance);
})
.catch(reject);
});
if (this._fetching) {
return new Promise((resolve) => {
this._queue.push({ resolve });
});
}

this._fetching = true;

return this._api.parity
.registryAddress()
.then((address) => {
this._fetching = false;
this._instance = this._api.newContract(abis.registry, address).instance;

this._queue.forEach((queued) => {
queued.resolve(this._instance);
});

this._queue = [];

return this._instance;
});
}

getContract (_name) {
Expand Down
13 changes: 4 additions & 9 deletions js/src/redux/providers/balances.js
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import { getBalances, getTokens } from './balancesActions';
import { setAddressImage } from './imagesActions';

import Contracts from '../../contracts';
import * as abis from '../../contracts/abi';

import imagesEthereum from '../../../assets/images/contracts/ethereum-black-64x64.png';
Expand Down Expand Up @@ -84,15 +85,9 @@ export default class Balances {
return Promise.resolve(this._tokenreg);
}

return this._api.parity
.registryAddress()
.then((registryAddress) => {
const registry = this._api.newContract(abis.registry, registryAddress);

return registry.instance.getAddress.call({}, [this._api.util.sha3('tokenreg'), 'A']);
})
.then((tokenregAddress) => {
const tokenreg = this._api.newContract(abis.tokenreg, tokenregAddress);
return Contracts.get().tokenReg
.getContract()
.then((tokenreg) => {
this._tokenreg = tokenreg;
this.attachToTokens();

Expand Down
32 changes: 19 additions & 13 deletions js/src/redux/providers/status.js
Original file line number Diff line number Diff line change
Expand Up @@ -71,8 +71,8 @@ export default class Status {
* @see src/views/Connection/connection.js
*/
_shouldPing = () => {
const { isConnected, isConnecting } = this._apiStatus;
return isConnecting || !isConnected;
const { isConnected } = this._apiStatus;
return !isConnected;
}

_stopPollPing = () => {
Expand Down Expand Up @@ -119,7 +119,7 @@ export default class Status {

_pollStatus = () => {
const nextTimeout = (timeout = 1000) => {
setTimeout(this._pollStatus, timeout);
setTimeout(() => this._pollStatus(), timeout);
};

const { isConnected, isConnecting, needsToken, secureToken } = this._api;
Expand All @@ -134,7 +134,8 @@ export default class Status {
const gotReconnected = !this._apiStatus.isConnected && apiStatus.isConnected;

if (gotReconnected) {
this._pollLongStatus();
this._pollLongStatus(true);
this._store.dispatch(statusCollection({ isPingable: true }));
}

if (!isEqual(apiStatus, this._apiStatus)) {
Expand Down Expand Up @@ -175,13 +176,12 @@ export default class Status {
this._store.dispatch(statusCollection(status));
this._status = status;
}

nextTimeout();
})
.catch((error) => {
console.error('_pollStatus', error);
nextTimeout(250);
});

nextTimeout();
}

/**
Expand Down Expand Up @@ -223,7 +223,11 @@ export default class Status {
* fetched every 30s just in case, and whenever
* the client got reconnected.
*/
_pollLongStatus = () => {
_pollLongStatus = (newConnection = false) => {
if (!this._api.isConnected) {
return;
}

const nextTimeout = (timeout = 30000) => {
if (this._longStatusTimeoutId) {
clearTimeout(this._longStatusTimeoutId);
Expand All @@ -242,7 +246,7 @@ export default class Status {
this._api.parity.netChain(),
this._api.parity.netPort(),
this._api.parity.rpcSettings(),
this._api.parity.enode()
newConnection ? Promise.resolve(null) : this._api.parity.enode()
])
.then(([
clientVersion, defaultExtraData, netChain, netPort, rpcSettings, enode
Expand All @@ -255,21 +259,23 @@ export default class Status {
netChain,
netPort,
rpcSettings,
enode,
isTest
};

if (enode) {
longStatus.enode = enode;
}

if (!isEqual(longStatus, this._longStatus)) {
this._store.dispatch(statusCollection(longStatus));
this._longStatus = longStatus;
}

nextTimeout();
})
.catch((error) => {
console.error('_pollLongStatus', error);
nextTimeout(250);
});

nextTimeout(newConnection ? 5000 : 30000);
}

_pollLogs = () => {
Expand Down
Loading

0 comments on commit 7e800b7

Please sign in to comment.