plumbing - pipe web3 to the popup.js

app/scripts/background.js

@@ -1,4 +1,6 @@
 const Dnode = require('dnode')
+const Multiplex = require('multiplex')
+const Through = require('through2')
 const eos = require('end-of-stream')
 const extend = require('xtend')
 const EthStore = require('eth-store')
@@ -15,17 +17,22 @@ console.log('ready to roll')
 chrome.runtime.onConnect.addListener(connectRemote)
 function connectRemote(remotePort){
   var isMetaMaskInternalProcess = (remotePort.name === 'popup')
+  var portStream = new PortStream(remotePort)
   if (isMetaMaskInternalProcess) {
     // communication with popup
-    handleInternalCommunication(remotePort)
+    handleInternalCommunication(portStream)
   } else {
     // communication with page
-    handleExternalCommunication(remotePort)
+    handleEthRpcRequestStream(portStream)
   }
 }
 
-function handleExternalCommunication(remotePort){
-  remotePort.onMessage.addListener(onRpcRequest.bind(null, remotePort))
+function handleEthRpcRequestStream(stream){
+  // portStream
+  stream.on('data', function(data){
+    console.log(data)
+  })
+  stream.on('data', onRpcRequest.bind(null, stream))
 }
 
 //
@@ -63,15 +70,15 @@ function getState(){
 }
 
 // handle rpc requests
-function onRpcRequest(remotePort, payload){
+function onRpcRequest(remoteStream, payload){
   // console.log('MetaMaskPlugin - incoming payload:', payload)
   zeroClient.sendAsync(payload, function onPayloadHandled(err, response){
     // provider engine errors are included in response objects
-    if (!payload.isMetamaskInternal) console.log('MetaMaskPlugin - RPC complete:', payload, '->', response)
+    // if (!payload.isMetamaskInternal) console.log('MetaMaskPlugin - RPC complete:', payload, '->', response)
     try {
-      remotePort.postMessage(response)
-    } catch (_) {
-      // port disconnected
+      remoteStream.push(response)
+    } catch (err) {
+      console.error(err)
     }
   })
 }
@@ -81,8 +88,28 @@ function onRpcRequest(remotePort, payload){
 // popup integration
 //
 
-function handleInternalCommunication(remotePort){
-  var duplex = new PortStream(remotePort)
+function handleInternalCommunication(portStream){
+  // setup multiplexing
+  var mx = Multiplex()
+  portStream.pipe(mx).pipe(portStream)
+  mx.on('error', function(err) {
+    console.error(err)
+    // portStream.destroy()
+  })
+  portStream.on('error', function(err) {
+    console.error(err)
+    mx.destroy()
+  })
+  var dnodeStream = mx.createSharedStream('dnode')
+  var providerStream =
+    jsonStringifyStream()
+    .pipe(mx.createSharedStream('provider'))
+    .pipe(jsonParseStream())
+  linkDnode(dnodeStream)
+  handleEthRpcRequestStream(providerStream)
+}
+
+function linkDnode(stream){
   var connection = Dnode({
     getState:           function(cb){ cb(null, getState()) },
     setRpcTarget:       setRpcTarget,
@@ -94,14 +121,14 @@ function handleInternalCommunication(remotePort){
     cancelTransaction:  idStore.cancelTransaction.bind(idStore),
     setLocked:          idStore.setLocked.bind(idStore),
   })
-  duplex.pipe(connection).pipe(duplex)
+  stream.pipe(connection).pipe(stream)
   connection.on('remote', function(remote){
     
     // push updates to popup
     ethStore.on('update', sendUpdate)
     idStore.on('update', sendUpdate)
     // teardown on disconnect
-    eos(duplex, function unsubscribe(){
+    eos(stream, function unsubscribe(){
       ethStore.removeListener('update', sendUpdate)
     })
     function sendUpdate(){
@@ -148,4 +175,20 @@ function getConfig(){
 
 function setConfig(state){
   localStorage['config'] = JSON.stringify(state)
+}
+
+// util
+
+function jsonParseStream(){
+  return Through.obj(function(serialized){
+    this.push(JSON.parse(serialized))
+    cb()
+  })
+}
+
+function jsonStringifyStream(){
+  return Through.obj(function(obj){
+    this.push(JSON.stringify(obj))
+    cb()
+  })
 }

app/scripts/lib/port-stream.js

@@ -18,12 +18,18 @@ function PortDuplexStream(port){
 // private
 
 PortDuplexStream.prototype._onMessage = function(msg){
-  // console.log('PortDuplexStream - saw message', msg)
-  this.push(msg)
+  if (Buffer.isBuffer(msg)) {
+    delete msg._isBuffer
+    var data = new Buffer(msg)
+    // console.log('PortDuplexStream - saw message as buffer', data)
+    this.push(data)
+  } else {
+    // console.log('PortDuplexStream - saw message', msg)
+    this.push(msg)
+  }
 }
 
-PortDuplexStream.prototype._onDisconnect = function(msg){
-  // console.log('PortDuplexStream - saw message', msg)
+PortDuplexStream.prototype._onDisconnect = function(){
   try {
     this.end()
   } catch(err){
@@ -36,9 +42,16 @@ PortDuplexStream.prototype._onDisconnect = function(msg){
 PortDuplexStream.prototype._read = noop
 
 PortDuplexStream.prototype._write = function(msg, encoding, cb){
-  // console.log('PortDuplexStream - sent message', msg)
   try {
-    this._port.postMessage(msg)
+    if (Buffer.isBuffer(msg)) {
+      var data = msg.toJSON()
+      data._isBuffer = true
+      // console.log('PortDuplexStream - sent message as buffer', data)
+      this._port.postMessage(data)
+    } else {
+      // console.log('PortDuplexStream - sent message', msg)
+      this._port.postMessage(msg)
+    }
     cb()
   } catch(err){
     // this.emit('error', err)

app/scripts/popup.js

@@ -1,11 +1,13 @@
 const url = require('url')
 const EventEmitter = require('events').EventEmitter
 const async = require('async')
+const Multiplex = require('multiplex')
 const Dnode = require('dnode')
 const MetaMaskUi = require('metamask-ui')
 const MetaMaskUiCss = require('metamask-ui/css')
 const injectCss = require('inject-css')
 const PortStream = require('./lib/port-stream.js')
+const StreamProvider = require('./lib/stream-provider.js')
 
 // setup app
 var css = MetaMaskUiCss()
@@ -19,15 +21,39 @@ async.parallel({
 function connectToAccountManager(cb){
   // setup communication with background
   var pluginPort = chrome.runtime.connect({name: 'popup'})
-  var duplex = new PortStream(pluginPort)
+  var portStream = new PortStream(pluginPort)
+  // setup multiplexing
+  var mx = Multiplex()
+  portStream.pipe(mx).pipe(portStream)
+  mx.on('error', function(err) {
+    console.error(err)
+    portStream.destroy()
+  })
+  portStream.on('error', function(err) {
+    console.error(err)
+    mx.destroy()
+  })
+  var dnodeStream = mx.createSharedStream('dnode')
+  var providerStream = mx.createSharedStream('provider')
+  linkDnode(dnodeStream, cb)
+  linkWeb3(providerStream)
+}
+
+function linkWeb3(stream){
+  var remoteProvider = new StreamProvider()
+  remoteProvider.pipe(stream).pipe(remoteProvider)
+  stream.on('error', console.error.bind(console))
+  remoteProvider.on('error', console.error.bind(console))
+}
+
+function linkDnode(stream, cb){
   var eventEmitter = new EventEmitter()
   var background = Dnode({
-    // setUnconfirmedTxs: setUnconfirmedTxs,
     sendUpdate: function(state){
       eventEmitter.emit('update', state)
     },
   })
-  duplex.pipe(background).pipe(duplex)
+  stream.pipe(background).pipe(stream)
   background.once('remote', function(accountManager){
     // setup push events
     accountManager.on = eventEmitter.on.bind(eventEmitter)

package.json

@@ -19,7 +19,9 @@
     "faux-jax": "git+https://github.com/kumavis/faux-jax.git#c3648de04804f3895c5b4972750cae5b51ddb103",
     "inject-css": "^0.1.1",
     "metamask-ui": "^1.5.0",
+    "multiplex": "^6.7.0",
     "readable-stream": "^2.0.5",
+    "through2": "^2.0.1",
     "web3": "^0.15.1",
     "web3-provider-engine": "^7.0.0",
     "xtend": "^4.0.1"