Global download and upload limit

index.js

@@ -13,6 +13,7 @@ const path = require('path')
 const Peer = require('simple-peer')
 const randombytes = require('randombytes')
 const speedometer = require('speedometer')
+var ThrottleGroup = require('stream-throttle').ThrottleGroup
 
 const TCPPool = require('./lib/tcp-pool') // browser exclude
 const Torrent = require('./lib/torrent')
@@ -72,6 +73,8 @@ class WebTorrent extends EventEmitter {
     this.tracker = opts.tracker !== undefined ? opts.tracker : {}
     this.torrents = []
     this.maxConns = Number(opts.maxConns) || 55
+    this.downloadLimit = Number(opts.downloadLimit) || Number.MAX_VALUE
+    this.uploadLimit = Number(opts.uploadLimit) || Number.MAX_VALUE
 
     this._debug(
       'new webtorrent (peerId %s, nodeId %s, port %s)',
@@ -95,6 +98,11 @@ class WebTorrent extends EventEmitter {
       }
     }
 
+    this.throttleGroups = {
+      down: new ThrottleGroup({rate: this.downloadLimit}),
+      up: new ThrottleGroup({rate: this.uploadLimit})
+    }
+
     if (typeof TCPPool === 'function') {
       this._tcpPool = new TCPPool(this)
     } else {
@@ -409,6 +417,26 @@ class WebTorrent extends EventEmitter {
 WebTorrent.WEBRTC_SUPPORT = Peer.WEBRTC_SUPPORT
 WebTorrent.VERSION = VERSION
 
+/**
+ * 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

@@ -92,10 +92,22 @@ class Peer {
     this.destroyed = false
     this.timeout = null // handshake timeout
     this.retries = 0 // outgoing TCP connection retry count
-
+    
     this.sentHandshake = false
   }
 
+  clearPipes() {
+    this.wire.unpipe()
+    this.conn.unpipe()
+  }
+
+  setThrottlePipes() {
+    this.conn.pipe(this.swarm.client.throttleGroups.down.throttle())
+      .pipe(this.wire)
+      .pipe(this.swarm.client.throttleGroups.up.throttle())
+      .pipe(this.conn)
+  }
+
   /**
    * Called once the peer is connected (i.e. fired 'connect' event)
    * @param {Socket} conn
@@ -142,7 +154,11 @@ class Peer {
     })
     this.startHandshakeTimeout()
 
-    conn.pipe(wire).pipe(conn)
+    if ((this.type === 'tcpOutgoing') || (this.type === 'webrtc')) {
+      this.setThrottlePipes()
+    } else {
+      conn.pipe(wire).pipe(conn)
+    }
     if (this.swarm && !this.sentHandshake) this.handshake()
   }
 

lib/tcp-pool.js

@@ -104,6 +104,8 @@ class TCPPool {
       if (torrent) {
         peer.swarm = torrent
         torrent._addIncomingPeer(peer)
+        peer.clearPipes()
+        peer.setThrottlePipes()
         peer.onHandshake(infoHash, peerId)
       } else {
         const err = new Error(

package.json

@@ -57,6 +57,7 @@
     "simple-peer": "^9.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",