<commit_before>var through = require('through2')
var bl = require('bl')

function parser (data, opts) {
  var toLine = opts.json ? toLineJSON : toLineString

  var result = through.obj(opts.tty ? parseTty : parse)

  result.payload = bl()
  result.headers = bl()
  result.currentLength = 0
  result.payloadLength = 0
  result.readingHeader = true
  result.publish = publish

  return result

  function parseTty (chunk, enc, cb) {
    this.payload.append(chunk)

    this.publish()

    return cb()
  }

  function parse (chunk, enc, cb) {
    var buffer = bl(chunk)

    while (buffer.length) {
      if (this.readingHeader) {
        readAndConsume(buffer, this.headers, 8 - this.headers.length)

        if (this.headers.length !== 8) {
          return cb()
        }

        this.readingHeader = false
        this.payloadLength = this.headers.readUInt32BE(4)
        this.headers.consume(8)
      }

      if (!buffer.length) {
        return cb()
      }

      var pointer = this.payloadLength - this.currentLength

      readAndConsume(buffer, this.payload, pointer)

      this.currentLength += pointer

      this.readingHeader = true
      this.currentLength = 0
      this.payloadLength = 0
    }

    this.publish()

    return cb()
  }

  function publish () {
    if (opts.newline) {
      var lines = this.payload.toString().split('\n').slice(0, -1)

      lines.forEach(function (line) {
        this.push(buildObject(line))
        this.payload.consume(new Buffer(line + '\n').length)<commit_msg>optimize publish() function with newline option, 3x faster, less memory usage<commit_after>var through = require('through2')
var bl = require('bl')

function parser (data, opts) {
  var toLine = opts.json ? toLineJSON : toLineString

  var result = through.obj(opts.tty ? parseTty : parse)

  result.payload = bl()
  result.headers = bl()
  result.currentLength = 0
  result.payloadLength = 0
  result.readingHeader = true
  result.publish = publish

  return result

  function parseTty (chunk, enc, cb) {
    this.payload.append(chunk)

    this.publish()

    return cb()
  }

  function parse (chunk, enc, cb) {
    var buffer = bl(chunk)

    while (buffer.length) {
      if (this.readingHeader) {
        readAndConsume(buffer, this.headers, 8 - this.headers.length)

        if (this.headers.length !== 8) {
          return cb()
        }

        this.readingHeader = false
        this.payloadLength = this.headers.readUInt32BE(4)
        this.headers.consume(8)
      }

      if (!buffer.length) {
        return cb()
      }

      var pointer = this.payloadLength - this.currentLength

      readAndConsume(buffer, this.payload, pointer)

      this.currentLength += pointer

      this.readingHeader = true
      this.currentLength = 0
      this.payloadLength = 0
    }

    this.publish()

    return cb()
  }

  function publish () {
    if (opts.newline) {
      var lines = this.payload.toString().split('\n').slice(0, -1)
      lines.forEach(function (line) {
        this.push(buildObject(line))
        this.payload.consume(Buffer.byteLength(line) + 1)
{'fraction_non_alphanumeric': 0.11934023285899094, 'fraction_numerical': 0.009055627425614488, 'mean_word_length': 2.5227790432801824, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4958848', 'n_tokens_mistral': 988, 'n_tokens_neox': 927, 'n_words': 218}