var entityStream  = require('seneca-entity-save-stream')
  , actStream     = require('seneca-act-stream')
  , csvStream     = require('csv2')
  , through       = require('through2')
  , xtend         = require('xtend')
  , Joi           = require('joi')

function csvToObj(dest, opts) {
  var csv   = csvStream()
    , header = null
    , count = 0

  opts = xtend({ skip: 0 }, opts)

  function rowToObject(row) {
    var obj = {}

    for (var i = 0; i < header.length; i++) {
      obj[header[i]] = row[i]
    }

    return obj
  }

  csv.pipe(through.obj({ highWaterMark: 16 }, function (chunk, enc, callback) {
    if (header === null) {
      header = chunk
      return callback()
    }

    if (count++ < opts.skip) {
      return callback()
    }

    var obj   = rowToObject(chunk)
      , that  = this

    if (!opts.schema) {
      this.push(obj)
      callback()
    } else {
      Joi.validate(obj, opts.schema, {
        convert: true
            , stripUnknown: true
      }, function(err, obj) {
        if (!err) {
          that.push(obj)
        }
        callback()
      })
    }
  })).pipe(dest)

  dest.on('finish', function() {
    csv.emit('importCompleted')
  })

  dest.on('one', function() {
    csv.rowsImported++
    csv.emit('rowImported')
  })

  dest.on('oneError', function(err) {
    csv.emit('rowError', err)
  })

  csv.rowsImported = 0

  return csv
}

module.exports.entity = function importEntity(seneca, entity, opts) {
  var dest = entityStream(seneca, { name$: entity })
  return csvToObj(dest, opts)
}

module.exports.act = function importAct(seneca, pattern, opts) {
  var dest = actStream(seneca, pattern)
  return csvToObj(dest, opts)
}

@@ -304,16 +304,37 @@
 vStream(
+%7B highWaterMark: 16 %7D
 )%0A    , 

Set highWaterMark on the csv stream.