<commit_before>module.exports = Wire

var BitField = require('bitfield')
var bncode = require('bncode')
var extend = require('extend.js')
var inherits = require('inherits')
var speedometer = require('speedometer')
var stream = require('stream')

var BITFIELD_GROW = 400000

var MESSAGE_PROTOCOL     = new Buffer('\u0013BitTorrent protocol')
var MESSAGE_KEEP_ALIVE   = new Buffer([0x00,0x00,0x00,0x00])
var MESSAGE_CHOKE        = new Buffer([0x00,0x00,0x00,0x01,0x00])
var MESSAGE_UNCHOKE      = new Buffer([0x00,0x00,0x00,0x01,0x01])
var MESSAGE_INTERESTED   = new Buffer([0x00,0x00,0x00,0x01,0x02])
var MESSAGE_UNINTERESTED = new Buffer([0x00,0x00,0x00,0x01,0x03])

var MESSAGE_RESERVED     = [0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]
var MESSAGE_PORT         = [0x00,0x00,0x00,0x03,0x09,0x00,0x00]

function Request (piece, offset, length, callback) {
  this.piece = piece
  this.offset = offset
  this.length = length
  this.callback = callback
}

inherits(Wire, stream.Duplex)

function Wire () {
  if (!(this instanceof Wire)) return new Wire()
  stream.Duplex.call(this)

  this.amChoking = true // are we choking the peer?
  this.amInterested = false // are we interested in the peer?

  this.peerChoking = true // is the peer choking us?
  this.peerInterested = false // is the peer interested in us?

  // The largest torrent that I know of (the Geocities archive) is ~641 GB and has
  // ~41,000 pieces. Therefore, cap bitfield to 10x larger (400,000 bits) to support all
  // possible torrents but prevent malicious peers from growing bitfield to fill memory.
  this.peerPieces = new BitField(0, { grow: BITFIELD_GROW })

  this.peerExtensions = {}

  this.requests = []
  this.peerRequests = []

  /** @type {Object} number -> string, ex: 1 -> 'ut_metadata' */
  this.extendedMapping = {}
  /** @type {Object} string -> number, ex: 9 -> 'ut_metadata' */
  this.peerExtendedMapping = {}

  /**
   * The extended handshake to send, minus the "m" field, which gets automatically
   * filled from `this.extendedMapping`.
   * @type {Object}
   */
  this.extendedHandshake = {}
  this.peerExtendedHandshake = {}

  /** @type {Object} string -> function, ex 'ut_metadata' -> ut_metadata() */
  this._ext = {}
  this._nextExt = 1

  this.uploaded = 0
  this.downloaded = 0
  this.uploadSpeed = speedometer()
  this.downloadSpeed = speedometer()

  this._keepAlive = null
  this._timeout = null
  this._timeoutMs = 0

  this.destroyed = false // was the wire ended by calling `destroy`?
  this._finished = false

  this._buffer = []
  this._bufferSize = 0
  this._parser = null
  this._parserSize = 0

  this.on('finish', this._onfinish)

  this._parseHandshake()
}

/**
 * Set whether to send a "keep-alive" ping (sent every 60s)
 * @param {boolean} enable
 */
Wire.prototype.setKeepAlive = function (enable) {
  clearInterval(this._keepAlive)
  if (enable === false) return
  this._keepAlive = setInterval(this._push.bind(this, MESSAGE_KEEP_ALIVE), 60000)
}

/**
 * Set the amount of time to wait before considering a request to be "timed out"
 * @param {number} ms
 */
Wire.prototype.setTimeout = function (ms) {
  this._clearTimeout()
  this._timeoutMs = ms
  this._updateTimeout()
}

Wire.prototype.destroy = function () {
  this.destroyed = true
  this.end()
}

Wire.prototype.end = function () {
  this._onUninterested()
  this._onChoke()
  stream.Duplex.prototype.end.apply(this, arguments)
}

//
// PROTOCOL EXTENSION API
//

Wire.prototype.use = function (Extension) {
  if (!Extension.name) {
    throw new Error('Extension API requires a named function, e.g. function name() {}')
  }

  var ext = this._nextExt
  var handler = new Extension(this)

  function noop () {}

  if (typeof handler.onHandshake !== 'function') {
    handler.onHandshake = noop
  }
  if (typeof handler.onExtendedHandshake !== 'function') {
    handler.onExtendedHandshake = noop
  }
  if (typeof handler.onMessage !== 'function') {
    handler.onMessage = noop
  }

  this.extendedMapping[ext] = Extension.name
  this._ext[Extension.name] = handler
  this[Extension.name] = handler

  this._nextExt += 1
}

//
// OUTGOING MESSAGES
//

/**
 * Message: "handshake" <pstrlen><pstr><reserved><info_hash><peer_id>
 * @param  {Buffer|string} infoHash (as Buffer or *hex* string)
 * @param  {Buffer|string} peerId
 * @param  {Object} extensions
 */
Wire.prototype.handshake = function (infoHash, peerId, extensions) {
  if (typeof infoHash === 'string')
    infoHash = new Buffer(infoHash, 'hex')
  if (typeof peerId === 'string')
    peerId = new Buffer(peerId, 'hex')
  if (infoHash.length !== 20 || peerId.length !== 20)
    throw new Error('infoHash and peerId MUST have length 20')

  var reserved = new Buffer(MESSAGE_RESERVED)

  // enable extended message<commit_msg>extension api: support `name` on prototype<commit_after>module.exports = Wire

var BitField = require('bitfield')
var bncode = require('bncode')
var extend = require('extend.js')
var inherits = require('inherits')
var speedometer = require('speedometer')
var stream = require('stream')

var BITFIELD_GROW = 400000

var MESSAGE_PROTOCOL     = new Buffer('\u0013BitTorrent protocol')
var MESSAGE_KEEP_ALIVE   = new Buffer([0x00,0x00,0x00,0x00])
var MESSAGE_CHOKE        = new Buffer([0x00,0x00,0x00,0x01,0x00])
var MESSAGE_UNCHOKE      = new Buffer([0x00,0x00,0x00,0x01,0x01])
var MESSAGE_INTERESTED   = new Buffer([0x00,0x00,0x00,0x01,0x02])
var MESSAGE_UNINTERESTED = new Buffer([0x00,0x00,0x00,0x01,0x03])

var MESSAGE_RESERVED     = [0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]
var MESSAGE_PORT         = [0x00,0x00,0x00,0x03,0x09,0x00,0x00]

function Request (piece, offset, length, callback) {
  this.piece = piece
  this.offset = offset
  this.length = length
  this.callback = callback
}

inherits(Wire, stream.Duplex)

function Wire () {
  if (!(this instanceof Wire)) return new Wire()
  stream.Duplex.call(this)

  this.amChoking = true // are we choking the peer?
  this.amInterested = false // are we interested in the peer?

  this.peerChoking = true // is the peer choking us?
  this.peerInterested = false // is the peer interested in us?

  // The largest torrent that I know of (the Geocities archive) is ~641 GB and has
  // ~41,000 pieces. Therefore, cap bitfield to 10x larger (400,000 bits) to support all
  // possible torrents but prevent malicious peers from growing bitfield to fill memory.
  this.peerPieces = new BitField(0, { grow: BITFIELD_GROW })

  this.peerExtensions = {}

  this.requests = []
  this.peerRequests = []

  /** @type {Object} number -> string, ex: 1 -> 'ut_metadata' */
  this.extendedMapping = {}
  /** @type {Object} string -> number, ex: 9 -> 'ut_metadata' */
  this.peerExtendedMapping = {}

  /**
   * The extended handshake to send, minus the "m" field, which gets automatically
   * filled from `this.extendedMapping`.
   * @type {Object}
   */
  this.extendedHandshake = {}
  this.peerExtendedHandshake = {}

  /** @type {Object} string -> function, ex 'ut_metadata' -> ut_metadata() */
  this._ext = {}
  this._nextExt = 1

  this.uploaded = 0
  this.downloaded = 0
  this.uploadSpeed = speedometer()
  this.downloadSpeed = speedometer()

  this._keepAlive = null
  this._timeout = null
  this._timeoutMs = 0

  this.destroyed = false // was the wire ended by calling `destroy`?
  this._finished = false

  this._buffer = []
  this._bufferSize = 0
  this._parser = null
  this._parserSize = 0

  this.on('finish', this._onfinish)

  this._parseHandshake()
}

/**
 * Set whether to send a "keep-alive" ping (sent every 60s)
 * @param {boolean} enable
 */
Wire.prototype.setKeepAlive = function (enable) {
  clearInterval(this._keepAlive)
  if (enable === false) return
  this._keepAlive = setInterval(this._push.bind(this, MESSAGE_KEEP_ALIVE), 60000)
}

/**
 * Set the amount of time to wait before considering a request to be "timed out"
 * @param {number} ms
 */
Wire.prototype.setTimeout = function (ms) {
  this._clearTimeout()
  this._timeoutMs = ms
  this._updateTimeout()
}

Wire.prototype.destroy = function () {
  this.destroyed = true
  this.end()
}

Wire.prototype.end = function () {
  this._onUninterested()
  this._onChoke()
  stream.Duplex.prototype.end.apply(this, arguments)
}

//
// PROTOCOL EXTENSION API
//

Wire.prototype.use = function (Extension) {
  var name = Extension.name || Extension.prototype.name
  if (!name) {
    throw new Error('Extension API requires a named function, e.g. function name() {}')
  }

  var ext = this._nextExt
  var handler = new Extension(this)

  function noop () {}

  if (typeof handler.onHandshake !== 'function') {
    handler.onHandshake = noop
  }
  if (typeof handler.onExtendedHandshake !== 'function') {
    handler.onExtendedHandshake = noop
  }
  if (typeof handler.onMessage !== 'function') {
    handler.onMessage = noop
  }

  this.extendedMapping[ext] = name
  this._ext[name] = handler
  this[name] = handler

  this._nextExt += 1
}

//
// OUTGOING MESSAGES
//

/**
 * Message: "handshake" <pstrlen><pstr><reserved><info_hash><peer_id>
 * @param  {Buffer|string} infoHash (as Buffer or *hex* string)
 * @param  {Buffer|string} peerId
 * @param  {Object} extensions
 */
Wire.prototype.handshake = function (infoHash, peerId, extensions) {
  if (typeof infoHash === 'string')
    infoHash = new Buffer(infoHash, 'hex')
  if (typeof peerId === 'string')
    peerId = new Buffer(peerId, 'hex')
  if (infoHash.length !== 20 || peerId.length !== 20)
    throw new Error('infoHash and peerId MUST have length 20')

  var reserved = new Buffer(MESSAGE_RESERVED)

  // enable extended message
{'fraction_non_alphanumeric': 0.1307893361212755, 'fraction_numerical': 0.034709879769994775, 'mean_word_length': 3.828874305906108, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5020005', 'n_tokens_mistral': 3465, 'n_tokens_neox': 3224, 'n_words': 836}