How to use the ndjson.parse function in ndjson

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

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

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

swarm.handle('/cyclon/0.1.0', (conn) => {
    var stream = ndjson.parse()
    stream.on('data', function (peers) {
      cb(peers, (err, reply) => {
        if (err) {
          console.log('do not know how to handle error yet')
          return
        }
        var serialize = ndjson.serialize()
        serialize.write(reply)
        serialize.end()
        serialize.pipe(conn)
      })
    })
    conn.pipe(stream)
  })
}

var listener = Net.createServer(function(connection) {
      seneca.log.debug(
        'listen',
        'connection',
        listenOptions,
        'remote',
        connection.remoteAddress,
        connection.remotePort
      )

      var parser = Ndjson.parse()
      var stringifier = Ndjson.stringify()
      parser.on('error', function(error) {
        console.error(error)
        connection.end()
      })
      parser.on('data', function(data) {
        if (data instanceof Error) {
          var out = transportUtil.prepareResponse(seneca, {})
          out.input = data.input
          out.error = transportUtil.error('invalid_json', { input: data.input })

          stringifier.write(out)
          return
        }

        transportUtil.handle_request(seneca, data, options, function(out) {

#!/usr/bin/env node
var csv = require('./')
var fs = require('fs')
var minimist = require('minimist')
var ndj = require('ndjson')

var args = minimist(process.argv.slice(2))


process.stdin
  .pipe(ndj.parse())
  .pipe(csv(args))
  .pipe(process.stdout)

async function loadNDJSON(path) {
  const records = [];

  const parser = fs.createReadStream(path).pipe(ndjson.parse());

  parser.on('data', obj => records.push(obj));

  return new Promise(function(resolve, reject) {
    parser.on('end', () => resolve(records));
    parser.on('error', e => reject(e));
  });
}