<commit_before>// TODO:
// - Support magnet metadata protocol for trackerless torrents
// - What happens when a peer connects and handhsakes for an infoHash that we don't have?
// - Add reconnect behavior for incoming connections, too
//   - Need to discover their listening port first

module.exports = Swarm

var EventEmitter = require('events').EventEmitter
var inherits = require('inherits')
var net = require('net')
var once = require('once')
var Wire = require('bittorrent-protocol')

var MAX_SIZE = 100
var HANDSHAKE_TIMEOUT = 5000
var RECONNECT_WAIT = [1000, 5000, 15000, 30000, 60000, 120000, 300000, 600000]

/**
 * Peer
 * ====
 * A peer in the swarm. Comprised of a `net.Socket` and a `Wire`.
 *
 * @param {string} addr
 */
function Peer (addr) {
  this.addr = addr

  this.conn = null
  this.wire = null

  this.timeout = null
  this.retries = 0
}

/**
 * Called once the peer's `conn` has connected (i.e. fired 'connect')
 * @param {Socket} conn
 */
Peer.prototype.onconnect = function (conn) {
  this.conn = conn

  var wire = this.wire = new Wire()
  wire.remoteAddress = this.addr
  var destroy = once(function () {
    this.conn.destroy()
    this.conn = null
  }.bind(this))

  // Close the wire when the connection is destroyed
  conn.once('end', function () { destroy() })
  conn.once('error', function () { destroy() })
  conn.once('close', function () { wire.end() })

  wire.once('end', function () {
    this.wire = null
  }.bind(this))

  // Duplex streaming magic!
  conn.pipe(wire).pipe(conn)
}

/**
 * Pool
 * ====
 * A "pool" is a bunch of swarms all listening on the same TCP port for
 * incoming connections from peers who are interested in one of our swarms.
 * There is one Pool for every port that a swarm is listening on, and they are
 * all stored in the `Pool.pools` object. When a connection comes in, the pool
 * does the wire protocol handshake with the peer to determine which swarm they
 * are interested in, and routes the connection to the right swarm.
 *
 * @param {number} port
 */
function Pool (port) {
  this.port = port
  this.swarms = {} // infoHash -> Swarm
  this.listening = false

  // Keep track of incoming connections so we can destroy them if we need to
  // close the server later.
  this.conns = []

  this.server = net.createServer(this._onconn.bind(this))
  this.server.listen(this.port, this._onlistening.bind(this))
  this.server.on('error', this._onerror.bind(this))

  this._retries = 0
}

/**
 * In-use Pools (port -> Pool)
 */
Pool.pools = {}

/**
 * STATIC METHOD: Add a swarm to a pool, creating a new pool if necessary.
 * @param {Swarm} swarm
 */
Pool.add = function (swarm) {
  var port = swarm.port
  var pool = Pool.pools[port]

  if (!pool)
    pool = Pool.pools[port] = new Pool(port)

  pool.addSwarm(swarm)
}

/**
 * STATIC METHOD: Remove a swarm from its pool.
 * @param  {Swarm} swarm
 */
Pool.remove = function (swarm) {
  var port = swarm.port
  var pool = Pool.pools[port]
  if (!pool) return

  pool.removeSwarm(swarm)

  if (Object.keys(pool.swarms).length === 0)
    delete Pool.pools[port]
}

Pool.prototype._onlistening = function () {
  this.listening = true
  for (var infoHash in this.swarms) {
    var swarm = this.swarms[infoHash]
    swarm.emit('listening')
  }
}

Pool.prototype._onconn = function (conn) {
  // Track all conns in this pool
  this.conns.push(conn)
  conn.on('close', function () {
    this.conns.splice(this.conns.indexOf(conn))
  }.bind(this))

  var addr = conn.remoteAddress + ':' + conn.remotePort

  // On incoming connections, we expect the remote peer to send a handshake
  // first. Based on the infoHash in that handshake, route the peer to the
  // right swarm.
  var peer = new Peer(addr)
  peer.onconnect(conn)

  // Peer must send handshake in timely manner - they connected to us after all
  var timeout = setTimeout(function () {
    conn.destroy()
  }, HANDSHAKE_TIMEOUT)

  peer.wire.on('handshake', function (infoHash, peerId, extensions) {
    clearTimeout(timeout)
    var swarm = this.swarms[infoHash.toString('hex')]
    if (!swarm)
      return conn.destroy()

    swarm._onincoming(peer)
  }.bind(this))
}

Pool.prototype._onerror = function (err) {
  if (err.code == 'EADDRINUSE' && this._retries < 5) {
    console.error('Address in use, retrying...')
    setTimeout(function () {
      this._retries += 1
      this.server.close()
      this.server.listen(this.port)
    }.bind(this), 1000)
  } else {
    this.swarms.forEach(function (swarm) {
      swarm.emit('error', 'Swarm listen error: ' + err.message)
    })<commit_msg>move todos to github issues<commit_after>module.exports = Swarm

var EventEmitter = require('events').EventEmitter
var inherits = require('inherits')
var net = require('net')
var once = require('once')
var Wire = require('bittorrent-protocol')

var MAX_SIZE = 100
var HANDSHAKE_TIMEOUT = 5000
var RECONNECT_WAIT = [1000, 5000, 15000, 30000, 60000, 120000, 300000, 600000]

/**
 * Peer
 * ====
 * A peer in the swarm. Comprised of a `net.Socket` and a `Wire`.
 *
 * @param {string} addr
 */
function Peer (addr) {
  this.addr = addr

  this.conn = null
  this.wire = null

  this.timeout = null
  this.retries = 0
}

/**
 * Called once the peer's `conn` has connected (i.e. fired 'connect')
 * @param {Socket} conn
 */
Peer.prototype.onconnect = function (conn) {
  this.conn = conn

  var wire = this.wire = new Wire()
  wire.remoteAddress = this.addr
  var destroy = once(function () {
    this.conn.destroy()
    this.conn = null
  }.bind(this))

  // Close the wire when the connection is destroyed
  conn.once('end', function () { destroy() })
  conn.once('error', function () { destroy() })
  conn.once('close', function () { wire.end() })

  wire.once('end', function () {
    this.wire = null
  }.bind(this))

  // Duplex streaming magic!
  conn.pipe(wire).pipe(conn)
}

/**
 * Pool
 * ====
 * A "pool" is a bunch of swarms all listening on the same TCP port for
 * incoming connections from peers who are interested in one of our swarms.
 * There is one Pool for every port that a swarm is listening on, and they are
 * all stored in the `Pool.pools` object. When a connection comes in, the pool
 * does the wire protocol handshake with the peer to determine which swarm they
 * are interested in, and routes the connection to the right swarm.
 *
 * @param {number} port
 */
function Pool (port) {
  this.port = port
  this.swarms = {} // infoHash -> Swarm
  this.listening = false

  // Keep track of incoming connections so we can destroy them if we need to
  // close the server later.
  this.conns = []

  this.server = net.createServer(this._onconn.bind(this))
  this.server.listen(this.port, this._onlistening.bind(this))
  this.server.on('error', this._onerror.bind(this))

  this._retries = 0
}

/**
 * In-use Pools (port -> Pool)
 */
Pool.pools = {}

/**
 * STATIC METHOD: Add a swarm to a pool, creating a new pool if necessary.
 * @param {Swarm} swarm
 */
Pool.add = function (swarm) {
  var port = swarm.port
  var pool = Pool.pools[port]

  if (!pool)
    pool = Pool.pools[port] = new Pool(port)

  pool.addSwarm(swarm)
}

/**
 * STATIC METHOD: Remove a swarm from its pool.
 * @param  {Swarm} swarm
 */
Pool.remove = function (swarm) {
  var port = swarm.port
  var pool = Pool.pools[port]
  if (!pool) return

  pool.removeSwarm(swarm)

  if (Object.keys(pool.swarms).length === 0)
    delete Pool.pools[port]
}

Pool.prototype._onlistening = function () {
  this.listening = true
  for (var infoHash in this.swarms) {
    var swarm = this.swarms[infoHash]
    swarm.emit('listening')
  }
}

Pool.prototype._onconn = function (conn) {
  // Track all conns in this pool
  this.conns.push(conn)
  conn.on('close', function () {
    this.conns.splice(this.conns.indexOf(conn))
  }.bind(this))

  var addr = conn.remoteAddress + ':' + conn.remotePort

  // On incoming connections, we expect the remote peer to send a handshake
  // first. Based on the infoHash in that handshake, route the peer to the
  // right swarm.
  var peer = new Peer(addr)
  peer.onconnect(conn)

  // Peer must send handshake in timely manner - they connected to us after all
  var timeout = setTimeout(function () {
    conn.destroy()
  }, HANDSHAKE_TIMEOUT)

  peer.wire.on('handshake', function (infoHash, peerId, extensions) {
    clearTimeout(timeout)
    var swarm = this.swarms[infoHash.toString('hex')]

    // Destroy connections from peers that handshake for an infoHash not in
    // this pool.
    if (!swarm)
      return conn.destroy()

    swarm._onincoming(peer)
  }.bind(this))
}

Pool.prototype._onerror = function (err) {
  if (err.code == 'EADDRINUSE' && this._retries < 5) {
    console.error('Address in use, retrying...')
    setTimeout(function () {
      this._retries += 1
      this.server.close()
      this.server.listen(this.port)
    }.bind(this), 1000)
  } else {
    this.swarms.forEach(function (swarm) {
      swarm.emit('error', 'Swarm listen error: ' + err.message)
    })
{'fraction_non_alphanumeric': 0.13270194986072423, 'fraction_numerical': 0.012701949860724234, 'mean_word_length': 3.235960358659745, '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': '6995371', 'n_tokens_mistral': 3079, 'n_tokens_neox': 2884, 'n_words': 917}