WebSocket Improvments #3544

This commit is contained in:
Nicolas Gotchac 2016-11-23 17:37:04 +01:00
parent a22469ade5
commit f800dd1eee
5 changed files with 193 additions and 55 deletions

View File

@ -29,21 +29,33 @@ export default class Ws extends JsonRpcBase {
this._token = token; this._token = token;
this._messages = {}; this._messages = {};
this._connecting = true; this._connecting = false;
this._connected = false;
this._lastError = null; this._lastError = null;
this._autoConnect = false; this._autoConnect = true;
this._retries = 0;
this._reconnectTimeoutId = null;
this._connect(); this._connect();
} }
updateToken (token) { updateToken (token) {
this._token = token; this._token = token;
this._autoConnect = false; this._autoConnect = true;
this._connect(); this._connect();
} }
_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 time = parseInt(new Date().getTime() / 1000, 10);
const sha3 = keccak_256(`${this._token}:${time}`); const sha3 = keccak_256(`${this._token}:${time}`);
const hash = `${sha3}_${time}`; const hash = `${sha3}_${time}`;
@ -53,6 +65,7 @@ export default class Ws extends JsonRpcBase {
this._ws.onopen = null; this._ws.onopen = null;
this._ws.onclose = null; this._ws.onclose = null;
this._ws.onmessage = null; this._ws.onmessage = null;
this._ws.close();
this._ws = null; this._ws = null;
} }
@ -65,6 +78,42 @@ export default class Ws extends JsonRpcBase {
this._ws.onopen = this._onOpen; this._ws.onopen = this._onOpen;
this._ws.onclose = this._onClose; this._ws.onclose = this._onClose;
this._ws.onmessage = this._onMessage; 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;
}
}
_checkNodeUp () {
const url = process.env.PARITY_URL || window.location.host;
return fetch(
`http://${url}/api/ping`,
{ method: 'HEAD' }
)
.then((r) => {
return r.status === 200;
}, () => {
return false;
})
.catch(() => false);
} }
_onOpen = (event) => { _onOpen = (event) => {
@ -72,6 +121,7 @@ export default class Ws extends JsonRpcBase {
this._connected = true; this._connected = true;
this._connecting = false; this._connecting = false;
this._autoConnect = true; this._autoConnect = true;
this._retries = 0;
Object.keys(this._messages) Object.keys(this._messages)
.filter((id) => this._messages[id].queued) .filter((id) => this._messages[id].queued)
@ -79,19 +129,51 @@ export default class Ws extends JsonRpcBase {
} }
_onClose = (event) => { _onClose = (event) => {
console.log('ws:onClose', event);
this._connected = false; this._connected = false;
this._connecting = false; this._connecting = false;
if (this._autoConnect) { this._checkNodeUp()
setTimeout(() => this._connect(), 500); .then((up) => {
// If the connection has been closed and the node
// is up, it means we have a wrong token
// Event code 1006 for WS means there is an error
// (not just closed by server)
if (up && event.code === 1006) {
event.status = 403;
} }
this._lastError = event;
if (this._autoConnect) {
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) => { _onError = (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); console.error('ws:onError', event);
this._lastError = event; this._lastError = event;
} }
}, 50);
}
_onMessage = (event) => { _onMessage = (event) => {
// Event sent by Signer Broadcaster // Event sent by Signer Broadcaster
@ -127,11 +209,16 @@ export default class Ws extends JsonRpcBase {
_send = (id) => { _send = (id) => {
const message = this._messages[id]; const message = this._messages[id];
message.queued = !this._connected;
if (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) { execute (method, ...params) {
@ -159,4 +246,27 @@ export default class Ws extends JsonRpcBase {
get lastError () { get lastError () {
return this._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);
}
} }

View File

@ -21,24 +21,38 @@ export default class Registry {
this._api = api; this._api = api;
this._contracts = []; this._contracts = [];
this._instance = null; this._instance = null;
this._fetching = false;
this._queue = [];
this.getInstance(); this.getInstance();
} }
getInstance () { getInstance () {
return new Promise((resolve, reject) => {
if (this._instance) { if (this._instance) {
resolve(this._instance); return Promise.resolve(this._instance);
return;
} }
this._api.parity if (this._fetching) {
return new Promise((resolve) => {
this._queue.push({ resolve });
});
}
this._fetching = true;
return this._api.parity
.registryAddress() .registryAddress()
.then((address) => { .then((address) => {
this._fetching = false;
this._instance = this._api.newContract(abis.registry, address).instance; this._instance = this._api.newContract(abis.registry, address).instance;
resolve(this._instance);
}) this._queue.forEach((queued) => {
.catch(reject); queued.resolve(this._instance);
});
this._queue = [];
return this._instance;
}); });
} }

View File

@ -17,6 +17,7 @@
import { getBalances, getTokens } from './balancesActions'; import { getBalances, getTokens } from './balancesActions';
import { setAddressImage } from './imagesActions'; import { setAddressImage } from './imagesActions';
import Contracts from '../../contracts';
import * as abis from '../../contracts/abi'; import * as abis from '../../contracts/abi';
import imagesEthereum from '../../../assets/images/contracts/ethereum-black-64x64.png'; import imagesEthereum from '../../../assets/images/contracts/ethereum-black-64x64.png';
@ -84,15 +85,9 @@ export default class Balances {
return Promise.resolve(this._tokenreg); return Promise.resolve(this._tokenreg);
} }
return this._api.parity return Contracts.get().tokenReg
.registryAddress() .getContract()
.then((registryAddress) => { .then((tokenreg) => {
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);
this._tokenreg = tokenreg; this._tokenreg = tokenreg;
this.attachToTokens(); this.attachToTokens();

View File

@ -71,8 +71,8 @@ export default class Status {
* @see src/views/Connection/connection.js * @see src/views/Connection/connection.js
*/ */
_shouldPing = () => { _shouldPing = () => {
const { isConnected, isConnecting } = this._apiStatus; const { isConnected } = this._apiStatus;
return isConnecting || !isConnected; return !isConnected;
} }
_stopPollPing = () => { _stopPollPing = () => {
@ -102,7 +102,7 @@ export default class Status {
}, timeout); }, timeout);
}; };
fetch('/', { method: 'HEAD' }) fetch('/api/ping', { method: 'HEAD' })
.then((response) => dispatch(!!response.ok)) .then((response) => dispatch(!!response.ok))
.catch(() => dispatch(false)); .catch(() => dispatch(false));
} }
@ -119,7 +119,7 @@ export default class Status {
_pollStatus = () => { _pollStatus = () => {
const nextTimeout = (timeout = 1000) => { const nextTimeout = (timeout = 1000) => {
setTimeout(this._pollStatus, timeout); setTimeout(() => this._pollStatus(), timeout);
}; };
const { isConnected, isConnecting, needsToken, secureToken } = this._api; const { isConnected, isConnecting, needsToken, secureToken } = this._api;
@ -134,7 +134,8 @@ export default class Status {
const gotReconnected = !this._apiStatus.isConnected && apiStatus.isConnected; const gotReconnected = !this._apiStatus.isConnected && apiStatus.isConnected;
if (gotReconnected) { if (gotReconnected) {
this._pollLongStatus(); this._pollLongStatus(true);
this._store.dispatch(statusCollection({ isPingable: true }));
} }
if (!isEqual(apiStatus, this._apiStatus)) { if (!isEqual(apiStatus, this._apiStatus)) {
@ -175,13 +176,12 @@ export default class Status {
this._store.dispatch(statusCollection(status)); this._store.dispatch(statusCollection(status));
this._status = status; this._status = status;
} }
nextTimeout();
}) })
.catch((error) => { .catch((error) => {
console.error('_pollStatus', error); console.error('_pollStatus', error);
nextTimeout(250);
}); });
nextTimeout();
} }
/** /**
@ -223,7 +223,11 @@ export default class Status {
* fetched every 30s just in case, and whenever * fetched every 30s just in case, and whenever
* the client got reconnected. * the client got reconnected.
*/ */
_pollLongStatus = () => { _pollLongStatus = (newConnection = false) => {
if (!this._api.isConnected) {
return;
}
const nextTimeout = (timeout = 30000) => { const nextTimeout = (timeout = 30000) => {
if (this._longStatusTimeoutId) { if (this._longStatusTimeoutId) {
clearTimeout(this._longStatusTimeoutId); clearTimeout(this._longStatusTimeoutId);
@ -242,7 +246,7 @@ export default class Status {
this._api.parity.netChain(), this._api.parity.netChain(),
this._api.parity.netPort(), this._api.parity.netPort(),
this._api.parity.rpcSettings(), this._api.parity.rpcSettings(),
this._api.parity.enode() newConnection ? Promise.resolve(null) : this._api.parity.enode()
]) ])
.then(([ .then(([
clientVersion, defaultExtraData, netChain, netPort, rpcSettings, enode clientVersion, defaultExtraData, netChain, netPort, rpcSettings, enode
@ -255,21 +259,23 @@ export default class Status {
netChain, netChain,
netPort, netPort,
rpcSettings, rpcSettings,
enode,
isTest isTest
}; };
if (enode) {
longStatus.enode = enode;
}
if (!isEqual(longStatus, this._longStatus)) { if (!isEqual(longStatus, this._longStatus)) {
this._store.dispatch(statusCollection(longStatus)); this._store.dispatch(statusCollection(longStatus));
this._longStatus = longStatus; this._longStatus = longStatus;
} }
nextTimeout();
}) })
.catch((error) => { .catch((error) => {
console.error('_pollLongStatus', error); console.error('_pollLongStatus', error);
nextTimeout(250);
}); });
nextTimeout(newConnection ? 5000 : 30000);
} }
_pollLogs = () => { _pollLogs = () => {

View File

@ -25,12 +25,13 @@ export default class SecureApi extends Api {
this._isConnecting = true; this._isConnecting = true;
this._connectState = sysuiToken === 'initial' ? 1 : 0; this._connectState = sysuiToken === 'initial' ? 1 : 0;
this._needsToken = false; this._needsToken = false;
this._nextToken = nextToken;
this._dappsPort = 8080; this._dappsPort = 8080;
this._dappsInterface = null; this._dappsInterface = null;
this._signerPort = 8180; this._signerPort = 8180;
this._followConnectionTimeoutId = null;
console.log('SecureApi:constructor', sysuiToken); // Try tokens from hash, then from localstorage
this._tokensToTry = [ nextToken, sysuiToken ].filter((t) => t && t.length);
this._followConnection(); this._followConnection();
} }
@ -42,13 +43,19 @@ export default class SecureApi extends Api {
_followConnection = () => { _followConnection = () => {
const nextTick = () => { const nextTick = () => {
setTimeout(() => this._followConnection(), 250); if (this._followConnectionTimeoutId) {
clearTimeout(this._followConnectionTimeoutId);
}
this._followConnectionTimeoutId = setTimeout(() => this._followConnection(), 250);
}; };
const setManual = () => { const setManual = () => {
this._connectState = 100; this._connectState = 100;
this._needsToken = true; this._needsToken = true;
this._isConnecting = false; this._isConnecting = false;
}; };
const lastError = this._transport.lastError; const lastError = this._transport.lastError;
const isConnected = this._transport.isConnected; const isConnected = this._transport.isConnected;
@ -58,12 +65,18 @@ export default class SecureApi extends Api {
if (isConnected) { if (isConnected) {
return this.connectSuccess(); return this.connectSuccess();
} else if (lastError) { } else if (lastError) {
const nextToken = this._nextToken || 'initial'; const nextToken = this._tokensToTry[0] || 'initial';
const nextState = this._nextToken ? 0 : 1; const nextState = nextToken !== 'initial' ? 0 : 1;
this._nextToken = null; // If previous token was wrong, delete it
if (lastError.status === 403) {
this._tokensToTry = this._tokensToTry.slice(1);
}
if (nextToken !== this._transport.token) {
this.updateToken(nextToken, nextState); this.updateToken(nextToken, nextState);
} }
}
break; break;
// token = 'initial' // token = 'initial'