How to use ndjson - 10 common examples

const server = http.createServer(function (req, res) {
  const count = ++requestCount
  console.log('[server req#%d] new request', count)

  // decompress the request body and parse it as ndjson
  pump(req, zlib.createGunzip(), ndjson.parse(), function (err) {
    if (err) {
      console.error(err.stack)
      res.statusCode = 500
    }
    console.log('[server req#%d] request body ended - responding with status code %d', count, res.statusCode)
    res.end()
  }).on('data', function (obj) {
    console.log('[server req#%d] got an ndjson object: %j', count, obj)
  })
})

const server = http.createServer(function (req, res) {
  const count = ++requestCount
  console.log('[server req#%d] new request', count)

  // parse the request body as ndjson
  pump(req, ndjson.parse(), function (err) {
    if (err) {
      console.error(err.stack)
      res.statusCode = 500
    }
    console.log('[server req#%d] request body ended - responding with status code %d', count, res.statusCode)
    res.end()
  }).on('data', function (obj) {
    console.log('[server req#%d] got an ndjson object: %j', count, obj)
  })
})

this.carpenter.build({ data: { data: pack, promote } }, (err, buildLog) => {
            if (err) return reject(err);
            //
            // Log the streaming responses of the builder.
            // TODO: can these be streamed back to npm?
            //
            buildLog.pipe(ndjson.parse())
              .on('error', reject)
              .on('data', (data) => {
                if (data.event === 'error') {
                  return reject(err);
                }

                this.log.info(data);
              }).on('end', resolve);
          });
        });

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;

latestSeq(function (err, seq) {
      if (err) throw err
    
      seq = Math.max(0, seq - 1) // sub 1 incase of errors

      if (seq) log('Continuing fetching npm data from seq: %d', seq)

      var url = 'https://skimdb.npmjs.com/registry/_changes?heartbeat=30000&include_docs=true&feed=continuous' + (seq ? '&since=' + seq : '')

      pump(request(url), ndjson.parse(), save(), startUpdating)
    })
  }

constructor (stream) {
    // TODO: rate limiting
    super()
    this.nonce = 0
    this.stream = pumpify(
      json.serialize(),
      stream,
      json.parse()
    )
    this.stream.on('data', this.onMessage.bind(this))
    this.stream.on('error', this.error.bind(this))
  }

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))
  }
}