gistfile1.js

var noop = function() {}
var jsonpath = require('jsonpath-rep')
var duplexer2 = require('duplexer2.jbenet')

function processJsonpathStream(path, otherStream) {
  var queue = []
  var spliter = through2.obj(split)
  var joiner = through2.obj(join)
  spliter.pipe(otherStream).pipe(joiner)
  return duplexer2(spliter, joiner)
  
  function split(item, enc, next) {
    queue.unshift(item)
    var val = jsonpath.eval(item, path)
    this.push(val)
    next()
  }
  
  function join(item, enc, nect) {
    var orig = queue.pop()
    var replaced = jsonpath.eval(orig, path, undefined, item)
    this.push(replaced)
    next()
  }
}

using https://github.com/mafintosh/write-transform-read

var wtr = require('write-transform-read')
var through = require('through2')

var wrap = function(otherStream) {
  var transform = wtr(otherStream)
  return through.obj(function(data, enc, cb) {
    transform(data.bar, function(err, result) {
      if (err) return cb(err)
      data.bar = result
      cb(null, data)
    }, function(cb) {
      otherStream.end(cb)
    })
  })
}

use it like so

var squareBar = wrap(square)
squareBar.write({bar:5}) -> {bar:25}