openethereum/js/packages/api/transport/ws/ws.js

398 lines
10 KiB
JavaScript

// Copyright 2015-2017 Parity Technologies (UK) Ltd.
// This file is part of Parity.
// Parity is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Parity is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
import { keccak_256 } from 'js-sha3'; // eslint-disable-line camelcase
import { Logging } from '../../subscriptions';
import JsonRpcBase from '../jsonRpcBase';
import TransportError from '../error';
/* global WebSocket */
export default class Ws extends JsonRpcBase {
// token is optional (secure API)
constructor (url, token = null, autoconnect = true) {
super();
this._url = url;
this._token = token;
this._messages = {};
this._subscriptions = { 'eth_subscription': [], 'parity_subscription': [] };
this._sessionHash = null;
this._connecting = false;
this._connected = false;
this._lastError = null;
this._autoConnect = autoconnect;
this._retries = 0;
this._reconnectTimeoutId = null;
this._connectPromise = null;
this._connectPromiseFunctions = {};
if (autoconnect) {
this.connect();
}
}
updateToken (token, connect = true) {
this._token = token;
// this._autoConnect = true;
if (connect) {
this.connect();
}
}
connect () {
if (this._connected) {
return Promise.resolve();
}
if (this._connecting) {
return this._connectPromise || Promise.resolve();
}
if (this._reconnectTimeoutId) {
window.clearTimeout(this._reconnectTimeoutId);
this._reconnectTimeoutId = null;
}
if (this._ws) {
this._ws.onerror = null;
this._ws.onopen = null;
this._ws.onclose = null;
this._ws.onmessage = null;
this._ws.close();
this._ws = null;
this._sessionHash = null;
}
this._connecting = true;
this._connected = false;
this._lastError = null;
// rpc secure API
if (this._token) {
const time = parseInt(new Date().getTime() / 1000, 10);
const sha3 = keccak_256(`${this._token}:${time}`);
const hash = `${sha3}_${time}`;
this._sessionHash = sha3;
this._ws = new WebSocket(this._url, hash);
// non-secure API
} else {
this._ws = new WebSocket(this._url);
}
this._ws.onerror = this._onError;
this._ws.onopen = this._onOpen;
this._ws.onclose = this._onClose;
this._ws.onmessage = this._onMessage;
// Get counts in dev mode only
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}`, `(+${n})`);
}
this._lastCount = {
timestamp: Date.now(),
count: this._count
};
}, 5000);
window._parityWS = this;
}
this._connectPromise = new Promise((resolve, reject) => {
this._connectPromiseFunctions = { resolve, reject };
});
return this._connectPromise;
}
_onOpen = (event) => {
this._setConnected();
this._connecting = false;
this._retries = 0;
Object.keys(this._messages)
.filter((id) => this._messages[id].queued)
.forEach(this._send);
this._connectPromiseFunctions.resolve();
this._connectPromise = null;
this._connectPromiseFunctions = {};
}
_onClose = (event) => {
this._setDisconnected();
this._connecting = false;
event.timestamp = Date.now();
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;
}
if (this._connectPromise) {
this._connectPromiseFunctions.reject(event);
this._connectPromise = null;
this._connectPromiseFunctions = {};
}
console.log('ws:onClose');
}
_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.timestamp = Date.now();
this._lastError = event;
if (this._connectPromise) {
this._connectPromiseFunctions.reject(event);
this._connectPromise = null;
this._connectPromiseFunctions = {};
}
}
}, 50);
}
_extract = (result) => {
const { result: res, id, method, params } = result;
const msg = this._messages[id];
// initial pubsub ACK
if (id && msg.subscription) {
// save subscription to map subId -> messageId
this._subscriptions[msg.subscription][res] = id;
// resolve promise with messageId because subId's can collide (eth/parity)
msg.resolve(id);
// save subId for unsubscribing later
msg.subId = res;
return msg;
}
// normal message
if (id) {
return msg;
}
// pubsub format
if (method.includes('subscription')) {
const messageId = this._messages[this._subscriptions[method][params.subscription]];
if (messageId) {
return messageId;
} else {
throw Error(`Received Subscription which is already unsubscribed ${JSON.stringify(result)}`);
}
}
throw Error(`Unknown message format: No ID or subscription ${JSON.stringify(result)}`);
}
_onMessage = (event) => {
try {
const result = JSON.parse(event.data);
const { method, params, json, resolve, reject, callback, subscription } = this._extract(result);
Logging.send(method, params, { json, result });
result.error = (result.params && result.params.error) || result.error;
if (result.error) {
this.error(event.data);
// Don't print error if request rejected or not is not yet up...
if (!/(rejected|not yet up)/.test(result.error.message)) {
console.error(`${method}(${JSON.stringify(params)}): ${result.error.code}: ${result.error.message}`);
}
const error = new TransportError(method, result.error.code, result.error.message);
if (result.id) {
reject(error);
} else {
callback(error);
}
delete this._messages[result.id];
return;
}
// if not initial subscription message resolve & delete
if (result.id && !subscription) {
resolve(result.result);
delete this._messages[result.id];
} else if (result.params) {
callback(null, result.params.result);
}
} catch (e) {
console.error('ws::_onMessage', event.data, e);
}
}
_send = (id) => {
const message = this._messages[id];
if (this._connected) {
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) {
return new Promise((resolve, reject) => {
const id = this.id;
const json = this.encode(method, params);
this._messages[id] = { id, method, params, json, resolve, reject };
this._send(id);
});
}
_methodsFromApi (api) {
const method = `${api}_subscribe`;
const uMethod = `${api}_unsubscribe`;
const subscription = `${api}_subscription`;
return { method, uMethod, subscription };
}
subscribe (api, callback, params) {
return new Promise((resolve, reject) => {
const id = this.id;
const { method, uMethod, subscription } = this._methodsFromApi(api);
const json = this.encode(method, params);
this._messages[id] = { id, method, uMethod, params, json, resolve, reject, callback, subscription };
this._send(id);
});
}
unsubscribeAll () {
return new Promise((resolve, reject) => {
var unsubscribed = 0;
let keys = Object.keys(this._messages);
keys.forEach(i => this._messages[i].subscription ? this.unsubscribe(this._messages[i].subId).then(_ => unsubscribed++, reject) : null);
resolve(unsubscribed);
});
}
unsubscribe (messageId) {
return new Promise((resolve, reject) => {
const id = this.id;
const { subId, uMethod, subscription } = this._messages[messageId];
const params = [subId];
const json = this.encode(uMethod, params);
const uResolve = (v) => {
delete this._messages[messageId];
delete this._subscriptions[subscription][subId];
resolve(v);
};
this._messages[id] = { id, method: uMethod, params, json, resolve: uResolve, reject };
this._send(id);
});
}
set url (url) {
this._url = url;
}
get token () {
return this._token;
}
get sessionHash () {
return this._sessionHash;
}
get isAutoConnect () {
return this._autoConnect;
}
get isConnecting () {
return this._connecting;
}
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);
}
}