Feature/163 global download upload limit

index.js

@@ -18,6 +18,7 @@ var Peer = require('simple-peer')
 var randombytes = require('randombytes')
 var speedometer = require('speedometer')
 var zeroFill = require('zero-fill')
+var ThrottleGroup = require('stream-throttle').ThrottleGroup
 
 var TCPPool = require('./lib/tcp-pool') // browser exclude
 var Torrent = require('./lib/torrent')
@@ -87,6 +88,8 @@ function WebTorrent (opts) {
   self.tracker = opts.tracker !== undefined ? opts.tracker : {}
   self.torrents = []
   self.maxConns = Number(opts.maxConns) || 55
+  self.downloadLimit = Number(opts.downloadLimit) || Number.MAX_VALUE
+  self.uploadLimit = Number(opts.uploadLimit) || Number.MAX_VALUE
 
   self._debug(
     'new webtorrent (peerId %s, nodeId %s, port %s)',
@@ -110,6 +113,11 @@ function WebTorrent (opts) {
     }
   }
 
+  self.throttleGroups = {
+    down: new ThrottleGroup({rate: self.downloadLimit}),
+    up: new ThrottleGroup({rate: self.uploadLimit})
+  }
+
   if (typeof TCPPool === 'function') {
     self._tcpPool = new TCPPool(self)
   } else {
@@ -459,6 +467,26 @@ WebTorrent.prototype._debug = function () {
   debug.apply(null, args)
 }
 
+/**
+ * Set global download throttle rate
+ * @param  {Number} rate
+ */
+WebTorrent.prototype.throttleDownload = function (rate) {
+  if (!Number(rate) || Number(rate) < 0) return
+  this.throttleGroups.down.bucket.bucketSize = rate
+  this.throttleGroups.down.bucket.tokensPerInterval = rate
+}
+
+/**
+ * Set global upload throttle rate
+ * @param  {Number} rate
+ */
+WebTorrent.prototype.throttleUpload = function (rate) {
+  if (!Number(rate) || Number(rate) < 0) return
+  this.throttleGroups.up.bucket.bucketSize = rate
+  this.throttleGroups.up.bucket.tokensPerInterval = rate
+}
+
 /**
  * Check if `obj` is a node Readable stream
  * @param  {*} obj

lib/peer.js

@@ -143,10 +143,27 @@ Peer.prototype.onConnect = function () {
   })
   self.startHandshakeTimeout()
 
-  conn.pipe(wire).pipe(conn)
+  if ((self.type === 'tcpOutgoing') || (self.type === 'webrtc')) {
+    self.setThrottlePipes()
+  } else {
+    conn.pipe(wire).pipe(conn)
+  }
+
   if (self.swarm && !self.sentHandshake) self.handshake()
 }
 
+Peer.prototype.clearPipes = function () {
+  this.wire.unpipe()
+  this.conn.unpipe()
+}
+
+Peer.prototype.setThrottlePipes = function () {
+  this.conn.pipe(this.swarm.client.throttleGroups.down.throttle())
+    .pipe(this.wire)
+    .pipe(this.swarm.client.throttleGroups.up.throttle())
+    .pipe(this.conn)
+}
+
 /**
  * Called when handshake is received from remote peer.
  * @param {string} infoHash

lib/tcp-pool.js

@@ -106,6 +106,8 @@ TCPPool.prototype._onConnection = function (conn) {
     var torrent = self._client.get(infoHash)
     if (torrent) {
       peer.swarm = torrent
+      peer.clearPipes()
+      peer.setThrottlePipes()
       torrent._addIncomingPeer(peer)
       peer.onHandshake(infoHash, peerId)
     } else {

package.json

@@ -57,6 +57,7 @@
     "simple-peer": "^8.0.0",
     "simple-sha1": "^2.0.8",
     "speedometer": "^1.0.0",
+    "stream-throttle": "^0.1.3",
     "stream-to-blob": "^1.0.0",
     "stream-to-blob-url": "^2.1.0",
     "stream-with-known-length-to-buffer": "^1.0.0",