<commit_before>var protobuf = require('protocol-buffers')
var varint = require('varint')
var stream = require('stream')
var util = require('util')
var debug = require('debug')('protocol-buffers-stream')
var fs = require('fs')
var path = require('path')

var Handshake = protobuf(fs.readFileSync(path.join(__dirname, 'handshake.proto'))).Handshake

var camelize = function(key) {
  return key[0].toLowerCase()+key.slice(1)
}

var compile = function(schema) {
  var messages = protobuf(schema)
  var encodings = []

  var handshake = {
    messages: Object.keys(messages)
  }

  var Messages = function() {
    if (!(this instanceof Messages)) return new Messages()
    stream.Duplex.call(this)

    this.destroyed = false
    this._missing = 0
    this._buffer = null
    this._header = new Buffer(50)
    this._ptr = 0
    this._events = []
    this._decodings = []
    this._encodings = encodings
    this._push(0, handshake)
  }

  util.inherits(Messages, stream.Duplex)

  Object.keys(messages).forEach(function(key, i) {
    encodings.push(messages[key])
    Messages.prototype[camelize(key)] = function(obj) {
      this._push(i+1, obj)
    }
  })

  Messages.prototype._push = function(id, obj) {
    if (this.destroyed) return

    var enc = id === 0 ? Handshake : this._encodings[id-1]
    var len = enc.encodingLength(obj)

    debug('writing %s (id: %d, length: %d)', enc.name, id, len)

    len += varint.encodingLength(id)

    var offset = 0
    var buf = new Buffer(len+varint.encodingLength(len))

    varint.encode(len, buf, offset)
    offset += varint.encode.bytes

    varint.encode(id, buf, offset)
    offset += varint.encode.bytes

    enc.encode(obj, buf, offset)
    this.push(buf)
  }

  Messages.prototype.destroy = function(err) {
    if (this.destroyed) return
    this.destroyed = true
    if (err) this.emit('error', err)
    this.emit('close')
  }

  Messages.prototype.finalize = function() {
    this.push(null)
  }

  Messages.prototype._read = function() {
    // do nothing
  }

  Messages.prototype._write = function(data, enc, cb) {
    while (data && data.length && !this.destroyed) {
      if (this._missing) data = this._onmessage(data)
      else data = this._onheader(data)
    }
    cb()
  }

  Messages.prototype._emit = function() {
    this._missing = 0
    this._ptr = 0

    var id = varint.decode(this._buffer)
    var enc = id === 0 ? Handshake : this._decodings[id-1]
    var evt = id > 0 && this._events[id-1]

    if (!enc) {
      debug('unknown (id: %d)', id)
      return this.emit('unknown', id)
    }

    try {
      var obj = enc.decode(this._buffer, varint.decode.bytes)
    } catch (err) {
      debug('invalid %s (id: %d, length: %d)', enc.name, id, this._buffer.length)
      return this.emit('invalid', evt, this._buffer)
    }

    debug('received %s (id: %d, length: %d)', enc.name, id, this._buffer.length)
    if (enc !== Handshake) return this.emit(evt, obj)

    for (var i = 0; i < obj.messages.length; i++) {
      this._decodings[i] = messages[obj.messages[i]] || null
      this._events[i] = this._decodings[i] && camelize(this._decodings[i].name)
    }

    return false
  }

  Messages.prototype._onmessage = function(data) {
    if (!this._buffer) {
      if (data.length === this._missing) {
        this._buffer = data
        this._emit()
        return null
      }

      if (data.length > this._missing) {
        this._buffer = data.slice(0, this._missing)
        var overflow = data.slice(this._missing)
        this._emit()
        return overflow
      }

      this._buffer = new Buffer(this._missing)
    }

    if (data.length === this._missing) {
      data.copy(this._buffer, this._ptr)
      this._emit()<commit_msg>fix handshake when enums are present<commit_after>var protobuf = require('protocol-buffers')
var varint = require('varint')
var stream = require('stream')
var util = require('util')
var debug = require('debug')('protocol-buffers-stream')
var fs = require('fs')
var path = require('path')

var Handshake = protobuf(fs.readFileSync(path.join(__dirname, 'handshake.proto'))).Handshake

var camelize = function(key) {
  return key[0].toLowerCase()+key.slice(1)
}

var compile = function(schema) {
  var messages = protobuf(schema)
  var encodings = []

  var toMessage = function(key) {
    var m = messages[key]
    return m && typeof m.encode === 'function' && typeof m.decode === 'function' && m.name && m
  }

  var handshake = {
    messages: Object.keys(messages).filter(toMessage)
  }

  var Messages = function() {
    if (!(this instanceof Messages)) return new Messages()
    stream.Duplex.call(this)

    this.destroyed = false
    this._missing = 0
    this._buffer = null
    this._header = new Buffer(50)
    this._ptr = 0
    this._events = []
    this._decodings = []
    this._encodings = encodings
    this._push(0, handshake)
  }

  util.inherits(Messages, stream.Duplex)

  handshake.messages.forEach(function(key, i) {
    encodings.push(messages[key])
    Messages.prototype[camelize(key)] = function(obj) {
      this._push(i+1, obj)
    }
  })

  Messages.prototype._push = function(id, obj) {
    if (this.destroyed) return

    var enc = id === 0 ? Handshake : this._encodings[id-1]
    var len = enc.encodingLength(obj)

    debug('writing %s (id: %d, length: %d)', enc.name, id, len)

    len += varint.encodingLength(id)

    var offset = 0
    var buf = new Buffer(len+varint.encodingLength(len))

    varint.encode(len, buf, offset)
    offset += varint.encode.bytes

    varint.encode(id, buf, offset)
    offset += varint.encode.bytes

    enc.encode(obj, buf, offset)
    this.push(buf)
  }

  Messages.prototype.destroy = function(err) {
    if (this.destroyed) return
    this.destroyed = true
    if (err) this.emit('error', err)
    this.emit('close')
  }

  Messages.prototype.finalize = function() {
    this.push(null)
  }

  Messages.prototype._read = function() {
    // do nothing
  }

  Messages.prototype._write = function(data, enc, cb) {
    while (data && data.length && !this.destroyed) {
      if (this._missing) data = this._onmessage(data)
      else data = this._onheader(data)
    }
    cb()
  }

  Messages.prototype._emit = function() {
    this._missing = 0
    this._ptr = 0

    var id = varint.decode(this._buffer)
    var enc = id === 0 ? Handshake : this._decodings[id-1]
    var evt = id > 0 && this._events[id-1]

    if (!enc) {
      debug('unknown (id: %d)', id)
      return this.emit('unknown', id)
    }

    try {
      var obj = enc.decode(this._buffer, varint.decode.bytes)
    } catch (err) {
      debug('invalid %s (id: %d, length: %d)', enc.name, id, this._buffer.length)
      return this.emit('invalid', evt, this._buffer)
    }

    debug('received %s (id: %d, length: %d)', enc.name, id, this._buffer.length)
    if (enc !== Handshake) return this.emit(evt, obj)

    for (var i = 0; i < obj.messages.length; i++) {
      this._decodings[i] = toMessage(obj.messages[i]) || null
      this._events[i] = this._decodings[i] && camelize(this._decodings[i].name)
    }

    return false
  }

  Messages.prototype._onmessage = function(data) {
    if (!this._buffer) {
      if (data.length === this._missing) {
        this._buffer = data
        this._emit()
        return null
      }

      if (data.length > this._missing) {
        this._buffer = data.slice(0, this._missing)
        var overflow = data.slice(this._missing)
        this._emit()
        return overflow
      }

      this._buffer = new Buffer(this._missing)
    }

    if (data.length === this._missing) {
      data.copy(this._buffer, this._ptr)
      this._emit()
{'fraction_non_alphanumeric': 0.14926547743966423, 'fraction_numerical': 0.004984260230849948, 'mean_word_length': 2.8646730866700456, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '6969016', 'n_tokens_mistral': 2609, 'n_tokens_neox': 2519, 'n_words': 566}