<commit_before>var EventEmitter = require('events').EventEmitter
var fs = require('fs')
var path = require('path')
var protobuf = require('protocol-buffers')
var headers = protobuf(fs.readFileSync(path.join(__dirname, 'headers.proto')))
var hyperlog = require('hyperlog')
var net = require('net')
var cuid = require('cuid')
var os = require('os')
var pump = require('pump')
var fastparallel = require('fastparallel')
var eos = require('end-of-stream')
var through2 = require('through2')
var xtend = require('xtend')
var duplexify = require('duplexify')
var deepEqual = require('deep-equal')
var STOREID = '!!STOREID!!'
var PEERS = '!!PEERS!!'
var MYEVENTPEER = '!!MYEVENTPEER!!'
var defaults = {
  reconnectTimeout: 1000
}

function HyperEmitter (db, schema, opts) {
  if (!(this instanceof HyperEmitter)) {
    return new HyperEmitter(db, schema, opts)
  }

  this.messages = protobuf(schema)
  this._db = db
  this._hyperlog = hyperlog(db)
  this._last = null
  this._opts = xtend(defaults, opts)
  this._listening = false
  this._parallel = fastparallel()
  this._clients = {}
  this._lastClient = 0
  this._listeners = {}
  this._closed = false

  var that = this
  this._hyperlog.heads(function (err, heads) {
    if (err) { return that.status.emit('error', err) }

    that._last = heads
  })

  Object.keys(headers).forEach(function (header) {
    that.messages[header] = headers[header]
  })

  this._hyperlog.ready(function () {<commit_msg>Disable results/error for fastparallel.<commit_after>var EventEmitter = require('events').EventEmitter
var fs = require('fs')
var path = require('path')
var protobuf = require('protocol-buffers')
var headers = protobuf(fs.readFileSync(path.join(__dirname, 'headers.proto')))
var hyperlog = require('hyperlog')
var net = require('net')
var cuid = require('cuid')
var os = require('os')
var pump = require('pump')
var fastparallel = require('fastparallel')
var eos = require('end-of-stream')
var through2 = require('through2')
var xtend = require('xtend')
var duplexify = require('duplexify')
var deepEqual = require('deep-equal')
var STOREID = '!!STOREID!!'
var PEERS = '!!PEERS!!'
var MYEVENTPEER = '!!MYEVENTPEER!!'
var defaults = {
  reconnectTimeout: 1000
}

function HyperEmitter (db, schema, opts) {
  if (!(this instanceof HyperEmitter)) {
    return new HyperEmitter(db, schema, opts)
  }

  this.messages = protobuf(schema)
  this._db = db
  this._hyperlog = hyperlog(db)
  this._last = null
  this._opts = xtend(defaults, opts)
  this._listening = false
  this._parallel = fastparallel({ results: false })
  this._clients = {}
  this._lastClient = 0
  this._listeners = {}
  this._closed = false

  var that = this
  this._hyperlog.heads(function (err, heads) {
    if (err) { return that.status.emit('error', err) }

    that._last = heads
  })

  Object.keys(headers).forEach(function (header) {
    that.messages[header] = headers[header]
  })

  this._hyperlog.ready(function () {
{'fraction_non_alphanumeric': 0.14921821889870837, 'fraction_numerical': 0.004758667573079538, 'mean_word_length': 4.692456479690522, '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': '3462339', 'n_tokens_mistral': 975, 'n_tokens_neox': 954, 'n_words': 235}