code: function createStream (options) {
          options = options || {}
  
          var stream = through2.obj(function transform (chunk, encoding, cb) {
              stream.push(options.objectMode ? chunk : JSON.stringify(chunk)+'\n')
              cb()
          })
          stream.close = function close () {
              setImmediate(function () {
                  stream.push(null)
                  changes.unpipe(stream)
              })
              numStreams--
          }
          changes.pipe(stream)
          numStreams++
          return stream
      }
docs: create stream of changes