<commit_before>'use strict'

require('./lib/socket-hacks')

var Q = require('q')
var fs = require('fs')
var mkdirp = require('mkdirp')
var debug = require('debug')('bitkeeper-js')
var utils = require('tradle-utils')
var requireParam = utils.requireParam
var path = require('path')
var FSStorage = require('./lib/fsStorage')
var WebTorrent = require('webtorrent')
var reemit = require('re-emitter')
var common = require('./lib/common')
var EventEmitter = require('events').EventEmitter
var inherits = require('util').inherits
var Jobs = require('simple-jobs')
var Timeouts = require('timeouts')
var ports = require('promise-ports')
var DHT = require('bittorrent-dht/client')
var noop = function () {}

function Keeper (config) {
  EventEmitter.call(this)
  this._jobs = new Jobs()

  utils.bindPrototypeFunctions(this)

  this._config = config || {}
  this._initializers = []
  // this._pending = {}; // torrents that we're interested in but may not have files (or metadata) for yet
  this._seeding = {}
  this._timeouts = Timeouts()

  if (this.config('storage') === false) {
    this._storage = inMemoryStorage()
    this._loadDHT()
  } else this._initFromStorage()

  this.on('error', this._onerror)
}

Keeper.DHT = DHT
inherits(Keeper, EventEmitter)

Keeper.prototype._onerror = function (err, torrent) {
  this._debug(err, torrent && torrent.infoHash)
  if (torrent) {
    var infoHash = torrent.infoHash
    if (!infoHash) {
      // we may have tried to load a torrent with an invalid infoHash
      this._client.torrents.some(function (t) {
        if (t === torrent) {
          infoHash = t.infoHash
          return true
        }
      })
    }

    if (infoHash) {
      this.emit('error:' + torrent.infoHash, err)
      this.removeTorrent(infoHash)
    }
  }
}

Keeper.prototype._debug = function () {
  var args = [].slice.call(arguments)
  args[0] = '[' + this.torrentPort() + '] ' + args[0]
  return debug.apply(null, args)
}

Keeper.prototype.report = function () {
  var self = this
  var torrents = this._client.torrents
  if (!torrents || !torrents.length) return

  this._debug('REPLICATION REPORT')

  torrents.forEach(function (torrent) {
    self._debug('  ' + torrent.infoHash + ' : ' + self.peersCount(torrent))
  })

  this._debug('END REPLICATION REPORT')
}

Keeper.prototype._checkReady = function () {
  if (this.ready()) return
  if (!this._dht || !this._dht.ready) return
  // if (!this._portMapping) return
  if (!this._client || !this._client.ready) return

  this._ready = true
  this._debug('ready')
  this.emit('ready')
}

Keeper.prototype.ready = function () {
  return this._ready
}

Keeper.prototype._watchDHT = function () {
  var self = this
  var peers = this._dht.peers
  if (peers) {
    for (var infoHash in peers) {
      var addrs = peers[infoHash].index
      for (var addr in addrs) {
        if (addrs[addr]) this.addPeer(addr, infoHash)
      }
    }
  }

  this._dht.on('announce', function (addr, hash) {
    self.addPeer(addr, hash)
    self.emit('announce:' + hash, addr)
  })

  this.on('fullyReplicated', this.onFullyReplicated)
  this.monitorReplication()
}

Keeper.prototype._initTorrentClient = function () {
  if (this._destroyed) return

  this._client = new WebTorrent({
    dht: this._dht,
    tracker: false,
    dhtPort: this.dhtPort(),
    torrentPort: this.torrentPort()
  })

  this._client.on('ready', this._checkReady)
  this._client.on('torrent', this._ontorrent)
  reemit(this._client, this, ['warn', 'error'])

  this._checkReady()
}

Keeper.prototype.monitorReplication = function () {
  if (this._monitoringReplication) return

  this._monitoringReplication = true

  this._jobs.add('checkReplication', this.checkReplication, this.config('checkReplication') || 60000)
// this._jobs.add('report', this.report, 60000)
}

Keeper.prototype.checkTorrentReplication = function (torrent) {
  requireParam('torrent', torrent)

  var self = this

  this.judge(torrent)
    .then(function (verdict) {
      if (verdict.replicate) {
        self.replicate(torrent)
      } else if (verdict.drop) {<commit_msg>ping known nodes periodically to keep udp holes open<commit_after>'use strict'

require('./lib/socket-hacks')

var Q = require('q')
var fs = require('fs')
var mkdirp = require('mkdirp')
var debug = require('debug')('bitkeeper-js')
var utils = require('tradle-utils')
var requireParam = utils.requireParam
var path = require('path')
var FSStorage = require('./lib/fsStorage')
var WebTorrent = require('webtorrent')
var reemit = require('re-emitter')
var common = require('./lib/common')
var EventEmitter = require('events').EventEmitter
var inherits = require('util').inherits
var Jobs = require('simple-jobs')
var Timeouts = require('timeouts')
var ports = require('promise-ports')
var DHT = require('bittorrent-dht/client')
var noop = function () {}

function Keeper (config) {
  EventEmitter.call(this)
  this._jobs = new Jobs()

  utils.bindPrototypeFunctions(this)

  this._config = config || {}
  this._initializers = []
  // this._pending = {}; // torrents that we're interested in but may not have files (or metadata) for yet
  this._seeding = {}
  this._timeouts = Timeouts()

  if (this.config('storage') === false) {
    this._storage = inMemoryStorage()
    this._loadDHT()
  } else this._initFromStorage()

  this.on('error', this._onerror)
}

Keeper.DHT = DHT
inherits(Keeper, EventEmitter)

Keeper.prototype._onerror = function (err, torrent) {
  this._debug(err, torrent && torrent.infoHash)
  if (torrent) {
    var infoHash = torrent.infoHash
    if (!infoHash) {
      // we may have tried to load a torrent with an invalid infoHash
      this._client.torrents.some(function (t) {
        if (t === torrent) {
          infoHash = t.infoHash
          return true
        }
      })
    }

    if (infoHash) {
      this.emit('error:' + torrent.infoHash, err)
      this.removeTorrent(infoHash)
    }
  }
}

Keeper.prototype._debug = function () {
  var args = [].slice.call(arguments)
  args[0] = '[' + this.torrentPort() + '] ' + args[0]
  return debug.apply(null, args)
}

Keeper.prototype.report = function () {
  var self = this
  var torrents = this._client.torrents
  if (!torrents || !torrents.length) return

  this._debug('REPLICATION REPORT')

  torrents.forEach(function (torrent) {
    self._debug('  ' + torrent.infoHash + ' : ' + self.peersCount(torrent))
  })

  this._debug('END REPLICATION REPORT')
}

Keeper.prototype._checkReady = function () {
  if (this.ready()) return
  if (!this._dht || !this._dht.ready) return
  // if (!this._portMapping) return
  if (!this._client || !this._client.ready) return

  this._ready = true
  this._debug('ready')
  this.emit('ready')
}

Keeper.prototype.ready = function () {
  return this._ready
}

Keeper.prototype._watchDHT = function () {
  var self = this
  var peers = this._dht.peers
  if (peers) {
    for (var infoHash in peers) {
      var addrs = peers[infoHash].index
      for (var addr in addrs) {
        if (addrs[addr]) this.addPeer(addr, infoHash)
      }
    }
  }

  this._dht.on('announce', function (addr, hash) {
    self.addPeer(addr, hash)
    self.emit('announce:' + hash, addr)
  })

  this._dht.on('node', function(addr) {
    self._dht._sendPing(addr, noop)
  })

  this.on('fullyReplicated', this.onFullyReplicated)
  this.monitorReplication()
  this.keepAlive()
}

Keeper.prototype._initTorrentClient = function () {
  if (this._destroyed) return

  this._client = new WebTorrent({
    dht: this._dht,
    tracker: false,
    dhtPort: this.dhtPort(),
    torrentPort: this.torrentPort()
  })

  this._client.on('ready', this._checkReady)
  this._client.on('torrent', this._ontorrent)
  reemit(this._client, this, ['warn', 'error'])

  this._checkReady()
}

Keeper.prototype.keepAlive = function () {
  if (this._jobs.has('keepAlive')) return

  this._jobs.add('_pingNodes', this._pingNodes, 60000)
}

Keeper.prototype._pingNodes = function () {
  this._dht.nodes.toArray().forEach(function(n) {
    this._sendPing(n.addr, noop)
  }, this._dht)
}

Keeper.prototype.monitorReplication = function () {
  if (this._jobs.has('checkReplication')) return

  this._jobs.add('checkReplication', this.checkReplication, this.config('checkReplication') || 60000)
// this._jobs.add('report', this.report, 60000)
}

Keeper.prototype.checkTorrentReplication = function (torrent) {
  requireParam('torrent', torrent)

  var self = this

  this.judge(torrent)
    .then(function (verdict) {
      if (verdict.replicate) {
        self.replicate(torrent)
      } else if (verdict.drop) {
{'fraction_non_alphanumeric': 0.14902327757632472, 'fraction_numerical': 0.0033922096151596678, 'mean_word_length': 3.6391752577319587, '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': '6007512', 'n_tokens_mistral': 2963, 'n_tokens_neox': 2841, 'n_words': 611}