How to use the ndjson.serialize function in ndjson

function command(t) {
  var env = extend({}, process.env, {PATH:PATH})
  var proc = execspawn(t.command, {env:env})
  var dup = duplexer(proc.stdin, proc.stdout)

  proc.stderr.resume() // drain it!
  proc.unref()
  proc.stderr.unref()
  proc.stdout.unref()
  proc.stdin.unref()
  
  var serializer = ldjson.serialize()
  var parser = ldjson.parse()
  
  if (process.env.DEBUG) {
    serializer.pipe(debugStream('transform input: '))
    dup.pipe(debugStream('transform output: '))
  }
  
  return splicer.obj([serializer, dup, parser])
}

function WritableStreamPouch(opts, callback) {
  var api = this;
  api.instanceId = Math.random().toString();
  api.ndj = ndj.serialize();
  api.localStore = {};
  api.originalName = opts.name;

  // TODO: I would pass this in as a constructor opt, but
  // PouchDB changed how it clones in 5.0.0 so this broke
  api.setupStream = function (stream) {
    api.ndj.pipe(stream);
  };

  /* istanbul ignore next */
  api.type = function () {
    return 'writableStream';
  };

  api._remote = false;

return () => {
    const stream = ndjson.serialize();

    stream.write({
      event: error ? 'error' : 'info',
      id: 'blah',
      message: 'things'
    });

    stream.end();

    return readonly(stream);
  };
};

var path = require('path')
var fs = require('fs')

var usage = fs.readFileSync(__dirname + '/usage.txt').toString()

var transform = args._[0]
if (args.file) transform = require(path.resolve(process.cwd(), args.file))

if (!transform) {
  console.error(usage)
  process.exit(1)
}

process.stdin
  .pipe(map(transform, args))
  .pipe(ndjson.serialize())
  .pipe(process.stdout)

_startIPC () {
    var stdin = json.serialize()
    stdin.on('error', (err) => this.error(err))
    stdin.pipe(this.child.stdin)

    var stdout = json.parse()
    stdout.on('error', (err) => this.error(err))
    this.child.stdout.pipe(stdout)

    this.child.send = (data) => stdin.write(data)
    stdout.on('data', (data) => this.child.emit('message', data))
  }
}

function handleError(cout, dumpFile, error) {
  if (error.status === 206) {
    const
      errorFile = `${dumpFile.split('.').slice(0, -1).join('.')}-errors.jsonl`,
      writeStream = fs.createWriteStream(errorFile, { flags: 'a' }),
      serialize = ndjson.serialize().pipe(writeStream)

    serialize.on('data', line => {
      writeStream.write(line);
    });

    for (const partialError of error.errors) {
      serialize.write(partialError);
    }

    serialize.end();
    cout.warn(`[ℹ] Error importing ${dumpFile}. See errors in ${errorFile}`);
  } else {
    cout.warn(error.message);
  }
  process.exit(1);
}

RestHandler.prototype.stats = function(req, res) {
  var statsStream = this.dat.createStatsStream()
  statsStream.on('error', function(err) {
    var errObj = {
      type: 'statsStreamError',
      message: err.message
    }
    res.statusCode = 400
    serializer.write(errObj)
    serializer.end()
  })
  var serializer = ldj.serialize()
  pump(statsStream, serializer, res)
}

function writeNDJson(path, collection) {
  const serialize = ndjson.serialize();
  fd = fs.openSync(path, 'w');
  serialize.on('data', line => {
    fs.appendFileSync(fd, line, 'utf8');
  });

  for (const item of collection) {
    serialize.write(item);
  }
}

module.exports = function (opts) {
  return ping(xtend(config.ping, opts)).pipe(ndjson.serialize())
}