n.js

var level = require('level-hyper');
var LevelBatch = require('level-batch-stream');
var BatchStream = require('batch-stream');
var writeonly = require('write-only-stream');
var through = require('through2');
var streamify = require('stream-array');
var monotonic = require('monotonic-timestamp');
var uuid = require('uuid');

var db = level('test.db');

function writable() {
  var stream = through.obj();

  stream
    .pipe(through.obj(function (data, enc, cb) {
      cb(null, {
        type: 'put',
        key: monotonic(),
        value: data
      });
    }))
    .pipe(new BatchStream({ size: 1 }))
    .pipe(new LevelBatch(db));

  return writeonly(stream);
}

var exited = false;

var stream = writable()
  .on('error', function (err) {
    console.error(err);
    process.exit(1);
  })
  .on('finish', function () {
    db.createReadStream().on('data', function (d) {
      console.dir(d);
    }).on('error', function (err) {
      if (exited) return;
      exited = true;
      console.error(err);
      process.exit(1);
    }).on('end', function () {
      if (exited) return;
      exited = true;
      console.log('data?????');
      db.close(function () {
        process.exit(0);
      });
    })
  });

streamify(
  [ uuid.v4(), uuid.v4(), uuid.v4() ]
    .map(function (i) {
      return { id: i, createdAt: uuid.v1() };
    }
  )
).pipe(stream);