Issue 189 | Add WebsocketProvider

example.js

@@ -1,13 +1,27 @@
 const ProviderEngine = require('./index.js')
 const ZeroClientProvider = require('./zero.js')
+const FetchProvider = require('./subproviders/fetch')
+
+const fetchEngine = ZeroClientProvider({
+  getAccounts: function(){},
+  dataSubprovider: new FetchProvider({
+    rpcUrl: 'https://mainnet.infura.io',
+  })
+})
 
 // create engine
-const engine = ZeroClientProvider({
+const websocketEngine = ZeroClientProvider({
   getAccounts: function(){},
-  rpcUrl: 'https://mainnet.infura.io/',
+  rpcUrl: 'wss://mainnet.infura.io/_ws',
+  //debug: true,
 })
 
 // log new blocks
-engine.on('block', function(block){
-  console.log('BLOCK CHANGED:', '#'+block.number.toString('hex'), '0x'+block.hash.toString('hex'))
-})
+websocketEngine.on('block', function(block) {
+  console.log('WS:', '#'+block.number.toString('hex'), '0x'+block.hash.toString('hex'))
+})
+
+// log new blocks
+fetchEngine.on('block', function(block) {
+  console.log('Fetch:', '#'+block.number.toString('hex'), '0x'+block.hash.toString('hex'))
+})

package.json

@@ -8,15 +8,15 @@
     "test": "node test/index.js",
     "bundle-zero": "browserify -s ZeroClientProvider -e zero.js > dist/ZeroClientProvider.js",
     "bundle-engine": "browserify -s ProviderEngine -e index.js > dist/ProviderEngine.js",
-    "bundle": "mkdir -p ./dist && npm run bundle-engine && npm run bundle-zero"
+    "bundle": "cross-env mkdir -p ./dist && npm run bundle-engine && npm run bundle-zero"
   },
   "author": "",
   "license": "ISC",
   "dependencies": {
     "async": "^2.5.0",
     "clone": "^2.0.0",
-    "eth-sig-util": "^1.4.2",
     "eth-block-tracker": "^2.2.2",
+    "eth-sig-util": "^1.4.2",
     "ethereumjs-block": "^1.2.2",
     "ethereumjs-tx": "^1.2.0",
     "ethereumjs-util": "^5.1.1",
@@ -30,15 +30,18 @@
     "semaphore": "^1.0.3",
     "solc": "^0.4.2",
     "tape": "^4.4.0",
+    "ws": "^5.1.0",
     "xhr": "^2.2.0",
     "xtend": "^4.0.1"
   },
   "devDependencies": {
     "babel-preset-es2015": "^6.24.1",
     "babel-preset-stage-0": "^6.24.1",
+    "backoff": "^2.5.0",
     "browserify": "^14.0.0"
   },
   "browser": {
-    "request": false
+    "request": false,
+    "ws": "./util/ws.js"
   }
 }

subproviders/fetch.js

@@ -33,6 +33,7 @@ RpcSource.prototype.handleRequest = function (payload, next, end) {
 
   // overwrite id to not conflict with other concurrent users
   const newPayload = createPayload(payload)
+  newPayload.jsonrpc = '2.0'
   // remove extra parameter from request
   delete newPayload.origin
 

subproviders/websocket.js

@@ -0,0 +1,130 @@
+const Backoff = require('backoff')
+const createPayload = require('../util/create-payload')
+const inherits = require('util').inherits
+const Subprovider = require('./subprovider')
+const WebSocket = require('ws')
+
+class WebsocketProvider extends Subprovider {
+  constructor({ rpcUrl, debug }) {
+    super()
+
+    Object.defineProperties(this, {
+      _backoff: {
+        value: Backoff.exponential({
+          randomisationFactor: 0.2,
+          maxDelay: 5000
+        })
+      },
+      _connectTime: {
+        value: null,
+        writable: true
+      },
+      _log: {
+        value: debug
+          ? (...args) => console.info.apply(console, ['[WSProvider]', ...args])
+          : () => { }
+      },
+      _pendingRequests: {
+        value: new Map()
+      },
+      _socket: {
+        value: null,
+        writable: true
+      },
+      _unhandledRequests: {
+        value: []
+      },
+      _url: {
+        value: rpcUrl
+      }
+    })
+
+    this._handleSocketClose = this._handleSocketClose.bind(this)
+    this._handleSocketMessage = this._handleSocketMessage.bind(this)
+    this._handleSocketOpen = this._handleSocketOpen.bind(this)
+
+    // Called when a backoff timeout has finished. Time to try reconnecting.
+    this._backoff.on('ready', () => {
+      this._openSocket()
+    })
+
+    this._openSocket()
+  }
+
+  handleRequest(payload, next, end) {
+    if (!this._socket || this._socket.readyState !== 1) {
+      this._unhandledRequests.push(Array.from(arguments))
+      this._log('Socket not open. Request queued.')
+      return;
+    }
+
+    this._pendingRequests.set(payload.id, Array.from(arguments))
+
+    const newPayload = createPayload(payload)
+    delete newPayload.origin
+
+    this._socket.send(JSON.stringify(newPayload))
+    this._log('Request sent:', newPayload.method)
+  }
+
+  _handleSocketClose({ reason, code }) {
+    this._log(`Socket closed, code ${code} (${reason || 'no reason'})`);
+    // If the socket has been open for longer than 5 seconds, reset the backoff
+    if (this._connectTime && Date.now() - this._connectTime > 5000) {
+      this._backoff.reset()
+    }
+
+    this._socket.removeEventListener('close', this._handleSocketClose)
+    this._socket.removeEventListener('message', this._handleSocketMessage)
+    this._socket.removeEventListener('open', this._handleSocketOpen)
+
+    this._socket = null
+    this._backoff.backoff()
+  }
+
+  _handleSocketMessage(message) {
+    let data;
+
+    try {
+      data = JSON.parse(message.data)
+    } catch (e) {
+      this._log('Received a message that is not valid JSON:', message)
+      return;
+    }
+
+    this._log('Received message ID:', data.id)
+
+    if (!this._pendingRequests.has(data.id)) {
+      return
+    }
+
+    const [payload, next, end] = this._pendingRequests.get(data.id)
+    this._pendingRequests.delete(data.id)
+    end(data.error && data.error.message, data.result)
+  }
+
+  _handleSocketOpen() {
+    this._log('Socket open.')
+    this._connectTime = Date.now();
+
+    // Any pending requests need to be resent because our session was lost
+    // and will not get responses for them in our new session.
+    this._pendingRequests.forEach(value => this._unhandledRequests.push(value))
+    this._pendingRequests.clear()
+
+    const unhandledRequests = this._unhandledRequests.splice(0, this._unhandledRequests.length)
+    unhandledRequests.forEach(request => {
+      this.handleRequest.apply(this, request)
+    })
+  }
+
+  _openSocket() {
+    this._log('Opening socket...')
+    this._socket = new WebSocket(this._url)
+    this._socket.addEventListener('close', this._handleSocketClose)
+    this._socket.addEventListener('message', this._handleSocketMessage)
+    this._socket.addEventListener('open', this._handleSocketOpen)
+  }
+}
+
+module.exports = WebsocketProvider

util/create-payload.js

@@ -4,11 +4,10 @@ const extend = require('xtend')
 module.exports = createPayload
 
 
-function createPayload(data){
+function createPayload(data) {
   return extend({
     // defaults
     id: getRandomId(),
-    jsonrpc: '2.0',
     params: [],
     // user-specified
   }, data)

util/ws.js

@@ -0,0 +1 @@
+module.exports = WebSocket;

zero.js

@@ -7,7 +7,7 @@ const InflightCacheSubprovider = require('./subproviders/inflight-cache')
 const HookedWalletSubprovider = require('./subproviders/hooked-wallet.js')
 const SanitizingSubprovider = require('./subproviders/sanitizer.js')
 const RpcSubprovider = require('./subproviders/rpc.js')
-const FetchSubprovider = require('./subproviders/fetch.js')
+const WebsocketSubprovider = require('./subproviders/websocket.js')
 
 
 module.exports = ZeroClientProvider
@@ -67,9 +67,9 @@ function ZeroClientProvider(opts){
   engine.addProvider(idmgmtSubprovider)
 
   // data source
-  const dataSubprovider = opts.dataSubprovider || new FetchSubprovider({
-    rpcUrl: opts.rpcUrl || 'https://mainnet.infura.io/',
-    originHttpHeaderKey: opts.originHttpHeaderKey,
+  const dataSubprovider = opts.dataSubprovider || new WebsocketSubprovider({
+    rpcUrl: opts.rpcUrl || 'wss://mainnet.infura.io/_ws',
+    debug: opts.debug,
   })
   engine.addProvider(dataSubprovider)