stream-rpc-example.js

var duplexify = require('duplexify')
var lpstream = require('length-prefixed-stream')
var ids = require('numeric-id-map')
var util = require('util')

function RPC () {
  duplexify.call(this) // subclass yolo
  this._encode = lpstream.encode()
  this._decode = lpstream.decode()
  this._pending = ids()
  this.setReadable(this._encode) // read from the encoder
  this.setWritable(this._decode) // write to the decoder

  var self = this
  this._decode.on('data', function (data) {
    data = JSON.parse(data)
    if (data.type === 'request') {
      self.emit('request', data.data, function (err, res) {
        if (err) self._encode.write(JSON.stringify({id: data.id, type: 'error', error: err.message}))
        else self._encode.write(JSON.stringify({id: data.id, type: 'response', data: res}))
      })
    }
    if (data.type === 'error') {
      var req = self._pending.remove(data.id)
      if (req.callback) req.callback(new Error(req.error))
    }
    if (data.type === 'response') {
      var req = self._pending.remove(data.id)
      if (req.callback) req.callback(null, data.data)
    }
  })
}

util.inherits(RPC, duplexify)

RPC.prototype.request = function (msg, cb) {
  var req = {type: 'request', data: msg}
  req.id = this._pending.add(req)
  this._encode.write(JSON.stringify(req))
  req.callback = cb || function () {} // hack to not stringify callback
}


// usage

var net = require('net')

net.createServer(function (stream) {
  var rpc = new RPC()
  rpc.on('request', function (req, cb) {
    cb(null, {echo: req})
  })
  stream.pipe(rpc).pipe(stream)
}).listen(10000, function() {
  var rpc = new RPC()
  var stream = net.connect(10000)

  rpc.request({hello: 'world'}, console.log) // prints {echo: {hello: 'world'}}
  stream.pipe(rpc).pipe(stream)
})