index.js

/* global FileList */

module.exports = WebTorrent

var Buffer = require('safe-buffer').Buffer
var concat = require('simple-concat')
var createTorrent = require('create-torrent')
var debug = require('debug')('webtorrent')
var DHT = require('bittorrent-dht/client') // browser exclude
var EventEmitter = require('events').EventEmitter
var extend = require('xtend')
var inherits = require('inherits')
var loadIPSet = require('load-ip-set') // browser exclude
var parallel = require('run-parallel')
var parseTorrent = require('parse-torrent')
var path = require('path')
var Peer = require('simple-peer')
var randombytes = require('randombytes')
var speedometer = require('speedometer')
var zeroFill = require('zero-fill')

var TCPPool = require('./lib/tcp-pool') // browser exclude
var Torrent = require('./lib/torrent')

/**
 * WebTorrent version.
 */
var VERSION = require('./package.json').version

/**
 * Version number in Azureus-style. Generated from major and minor semver version.
 * For example:
 *   '0.16.1' -> '0016'
 *   '1.2.5' -> '0102'
 */
var VERSION_STR = VERSION.match(/([0-9]+)/g)
  .slice(0, 2)
  .map(function (v) { return zeroFill(2, v) })
  .join('')

/**
 * Version prefix string (used in peer ID). WebTorrent uses the Azureus-style
 * encoding: '-', two characters for client id ('WW'), four ascii digits for version
 * number, '-', followed by random numbers.
 * For example:
 *   '-WW0102-'...
 */
var VERSION_PREFIX = '-WW' + VERSION_STR + '-'

inherits(WebTorrent, EventEmitter)

/**
 * WebTorrent Client
 * @param {Object=} opts
 */
function WebTorrent (opts) {
  var self = this
  if (!(self instanceof WebTorrent)) return new WebTorrent(opts)
  EventEmitter.call(self)

  if (!opts) opts = {}

  if (typeof opts.peerId === 'string') {
    self.peerId = opts.peerId
  } else if (Buffer.isBuffer(opts.peerId)) {
    self.peerId = opts.peerId.toString('hex')
  } else {
    self.peerId = Buffer.from(VERSION_PREFIX + randombytes(9).toString('base64')).toString('hex')
  }
  self.peerIdBuffer = Buffer.from(self.peerId, 'hex')

  if (typeof opts.nodeId === 'string') {
    self.nodeId = opts.nodeId
  } else if (Buffer.isBuffer(opts.nodeId)) {
    self.nodeId = opts.nodeId.toString('hex')
  } else {
    self.nodeId = randombytes(20).toString('hex')
  }
  self.nodeIdBuffer = Buffer.from(self.nodeId, 'hex')

  self._debugId = self.peerId.toString('hex').substring(0, 7)

  self.destroyed = false
  self.listening = false
  self.torrentPort = opts.torrentPort || 0
  self.dhtPort = opts.dhtPort || 0
  self.tracker = opts.tracker !== undefined ? opts.tracker : {}
  self.torrents = []
  self.maxConns = Number(opts.maxConns) || 55

  self._debug(
    'new webtorrent (peerId %s, nodeId %s, port %s)',
    self.peerId, self.nodeId, self.torrentPort
  )

  if (self.tracker) {
    if (typeof self.tracker !== 'object') self.tracker = {}
    if (opts.rtcConfig) {
      // TODO: remove in v1
      console.warn('WebTorrent: opts.rtcConfig is deprecated. Use opts.tracker.rtcConfig instead')
      self.tracker.rtcConfig = opts.rtcConfig
    }
    if (opts.wrtc) {
      // TODO: remove in v1
      console.warn('WebTorrent: opts.wrtc is deprecated. Use opts.tracker.wrtc instead')
      self.tracker.wrtc = opts.wrtc
    }
    if (global.WRTC && !self.tracker.wrtc) {
      self.tracker.wrtc = global.WRTC
    }
  }

  if (typeof TCPPool === 'function') {
    self._tcpPool = new TCPPool(self)
  } else {
    process.nextTick(function () {
      self._onListening()
    })
  }

  // stats
  self._downloadSpeed = speedometer()
  self._uploadSpeed = speedometer()

  if (opts.dht !== false && typeof DHT === 'function' /* browser exclude */) {
    // use a single DHT instance for all torrents, so the routing table can be reused
    self.dht = new DHT(extend({ nodeId: self.nodeId }, opts.dht))

    self.dht.once('error', function (err) {
      self._destroy(err)
    })

    self.dht.once('listening', function () {
      var address = self.dht.address()
      if (address) self.dhtPort = address.port
    })

    // Ignore warning when there are > 10 torrents in the client
    self.dht.setMaxListeners(0)

    self.dht.listen(self.dhtPort)
  } else {
    self.dht = false
  }

  // Enable or disable BEP19 (Web Seeds). Enabled by default:
  self.enableWebSeeds = opts.webSeeds !== false

  if (typeof loadIPSet === 'function' && opts.blocklist != null) {
    loadIPSet(opts.blocklist, {
      headers: {
        'user-agent': 'WebTorrent/' + VERSION + ' (https://webtorrent.io)'
      }
    }, function (err, ipSet) {
      if (err) return self.error('Failed to load blocklist: ' + err.message)
      self.blocked = ipSet
      ready()
    })
  } else {
    process.nextTick(ready)
  }

  function ready () {
    if (self.destroyed) return
    self.ready = true
    self.emit('ready')
  }
}

WebTorrent.WEBRTC_SUPPORT = Peer.WEBRTC_SUPPORT

Object.defineProperty(WebTorrent.prototype, 'downloadSpeed', {
  get: function () { return this._downloadSpeed() }
})

Object.defineProperty(WebTorrent.prototype, 'uploadSpeed', {
  get: function () { return this._uploadSpeed() }
})

Object.defineProperty(WebTorrent.prototype, 'progress', {
  get: function () {
    var torrents = this.torrents.filter(function (torrent) {
      return torrent.progress !== 1
    })
    var downloaded = torrents.reduce(function (total, torrent) {
      return total + torrent.downloaded
    }, 0)
    var length = torrents.reduce(function (total, torrent) {
      return total + (torrent.length || 0)
    }, 0) || 1
    return downloaded / length
  }
})

Object.defineProperty(WebTorrent.prototype, 'ratio', {
  get: function () {
    var uploaded = this.torrents.reduce(function (total, torrent) {
      return total + torrent.uploaded
    }, 0)
    var received = this.torrents.reduce(function (total, torrent) {
      return total + torrent.received
    }, 0) || 1
    return uploaded / received
  }
})

/**
 * Returns the torrent with the given `torrentId`. Convenience method. Easier than
 * searching through the `client.torrents` array. Returns `null` if no matching torrent
 * found.
 *
 * @param  {string|Buffer|Object|Torrent} torrentId
 * @return {Torrent|null}
 */
WebTorrent.prototype.get = function (torrentId) {
  var self = this
  var i, torrent
  var len = self.torrents.length

  if (torrentId instanceof Torrent) {
    for (i = 0; i < len; i++) {
      torrent = self.torrents[i]
      if (torrent === torrentId) return torrent
    }
  } else {
    var parsed
    try { parsed = parseTorrent(torrentId) } catch (err) {}

    if (!parsed) return null
    if (!parsed.infoHash) throw new Error('Invalid torrent identifier')

    for (i = 0; i < len; i++) {
      torrent = self.torrents[i]
      if (torrent.infoHash === parsed.infoHash) return torrent
    }
  }
  return null
}

// TODO: remove in v1
WebTorrent.prototype.download = function (torrentId, opts, ontorrent) {
  console.warn('WebTorrent: client.download() is deprecated. Use client.add() instead')
  return this.add(torrentId, opts, ontorrent)
}

/**
 * Start downloading a new torrent. Aliased as `client.download`.
 * @param {string|Buffer|Object} torrentId
 * @param {Object} opts torrent-specific options
 * @param {function=} ontorrent called when the torrent is ready (has metadata)
 */
WebTorrent.prototype.add = function (torrentId, opts, ontorrent) {
  var self = this
  if (self.destroyed) throw new Error('client is destroyed')
  if (typeof opts === 'function') return self.add(torrentId, null, opts)

  self._debug('add')
  opts = opts ? extend(opts) : {}

  var torrent = new Torrent(torrentId, self, opts)
  self.torrents.push(torrent)

  torrent.once('_infoHash', onInfoHash)
  torrent.once('ready', onReady)
  torrent.once('close', onClose)

  function onInfoHash () {
    if (self.destroyed) return
    for (var i = 0, len = self.torrents.length; i < len; i++) {
      var t = self.torrents[i]
      if (t.infoHash === torrent.infoHash && t !== torrent) {
        torrent._destroy(new Error('Cannot add duplicate torrent ' + torrent.infoHash))
        return
      }
    }
  }

  function onReady () {
    if (self.destroyed) return
    if (typeof ontorrent === 'function') ontorrent(torrent)
    self.emit('torrent', torrent)
  }

  function onClose () {
    torrent.removeListener('_infoHash', onInfoHash)
    torrent.removeListener('ready', onReady)
    torrent.removeListener('close', onClose)
  }

  return torrent
}

/**
 * Start seeding a new file/folder.
 * @param  {string|File|FileList|Buffer|Array.<string|File|Buffer>} input
 * @param  {Object=} opts
 * @param  {function=} onseed called when torrent is seeding
 */
WebTorrent.prototype.seed = function (input, opts, onseed) {
  var self = this
  if (self.destroyed) throw new Error('client is destroyed')
  if (typeof opts === 'function') return self.seed(input, null, opts)

  self._debug('seed')
  opts = opts ? extend(opts) : {}

  // When seeding from fs path, initialize store from that path to avoid a copy
  if (typeof input === 'string') opts.path = path.dirname(input)
  if (!opts.createdBy) opts.createdBy = 'WebTorrent/' + VERSION_STR

  var torrent = self.add(null, opts, onTorrent)
  var streams

  if (isFileList(input)) input = Array.prototype.slice.call(input)
  if (!Array.isArray(input)) input = [ input ]

  parallel(input.map(function (item) {
    return function (cb) {
      if (isReadable(item)) concat(item, cb)
      else cb(null, item)
    }
  }), function (err, input) {
    if (self.destroyed) return
    if (err) return torrent._destroy(err)

    createTorrent.parseInput(input, opts, function (err, files) {
      if (self.destroyed) return
      if (err) return torrent._destroy(err)

      streams = files.map(function (file) {
        return file.getStream
      })

      createTorrent(input, opts, function (err, torrentBuf) {
        if (self.destroyed) return
        if (err) return torrent._destroy(err)

        var existingTorrent = self.get(torrentBuf)
        if (existingTorrent) {
          torrent._destroy(new Error('Cannot add duplicate torrent ' + existingTorrent.infoHash))
        } else {
          torrent._onTorrentId(torrentBuf, opts.forced_id)
        }
      })
    })
  })

  function onTorrent (torrent) {
    var tasks = [
      function (cb) {
        torrent.load(streams, cb)
      }
    ]
    if (self.dht) {
      tasks.push(function (cb) {
        torrent.once('dhtAnnounce', cb)
      })
    }
    parallel(tasks, function (err) {
      if (self.destroyed) return
      if (err) return torrent._destroy(err)
      _onseed(torrent)
    })
  }

  function _onseed (torrent) {
    self._debug('on seed')
    if (typeof onseed === 'function') onseed(torrent)
    torrent.emit('seed')
    self.emit('seed', torrent)
  }

  return torrent
}

/**
 * Remove a torrent from the client.
 * @param  {string|Buffer|Torrent}   torrentId
 * @param  {function} cb
 */
WebTorrent.prototype.remove = function (torrentId, cb) {
  this._debug('remove')
  var torrent = this.get(torrentId)
  if (!torrent) throw new Error('No torrent with id ' + torrentId)
  this._remove(torrentId, cb)
}

WebTorrent.prototype._remove = function (torrentId, cb) {
  var torrent = this.get(torrentId)
  if (!torrent) return
  this.torrents.splice(this.torrents.indexOf(torrent), 1)
  torrent.destroy(cb)
}

WebTorrent.prototype.address = function () {
  if (!this.listening) return null
  return this._tcpPool
    ? this._tcpPool.server.address()
    : { address: '0.0.0.0', family: 'IPv4', port: 0 }
}

/**
 * Destroy the client, including all torrents and connections to peers.
 * @param  {function} cb
 */
WebTorrent.prototype.destroy = function (cb) {
  if (this.destroyed) throw new Error('client already destroyed')
  this._destroy(null, cb)
}

WebTorrent.prototype._destroy = function (err, cb) {
  var self = this
  self._debug('client destroy')
  self.destroyed = true

  var tasks = self.torrents.map(function (torrent) {
    return function (cb) {
      torrent.destroy(cb)
    }
  })

  if (self._tcpPool) {
    tasks.push(function (cb) {
      self._tcpPool.destroy(cb)
    })
  }

  if (self.dht) {
    tasks.push(function (cb) {
      self.dht.destroy(cb)
    })
  }

  parallel(tasks, cb)

  if (err) self.emit('error', err)

  self.torrents = []
  self._tcpPool = null
  self.dht = null
}

WebTorrent.prototype._onListening = function () {
  this._debug('listening')
  this.listening = true

  if (this._tcpPool) {
    // Sometimes server.address() returns `null` in Docker.
    // WebTorrent issue: https://github.com/feross/bittorrent-swarm/pull/18
    var address = this._tcpPool.server.address()
    if (address) this.torrentPort = address.port
  }

  this.emit('listening')
}

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

/**
 * Check if `obj` is a node Readable stream
 * @param  {*} obj
 * @return {boolean}
 */
function isReadable (obj) {
  return typeof obj === 'object' && obj != null && typeof obj.pipe === 'function'
}

/**
 * Check if `obj` is a W3C `FileList` object
 * @param  {*} obj
 * @return {boolean}
 */
function isFileList (obj) {
  return typeof FileList !== 'undefined' && obj instanceof FileList
}