add stream back preasure

lib/file-stream.js

@@ -37,12 +37,13 @@ function FileStream (file, opts) {
   this._reading = false
   this._notifying = false
   this._criticalLength = Math.min((1024 * 1024 / pieceLength) | 0, 2)
+  this._backPreasure = false;
 }
 
 FileStream.prototype._read = function () {
   if (this._reading) return
   this._reading = true
-  this._notify()
+  this.emit('piece', this._piece);
 }
 
 FileStream.prototype._notify = function () {
@@ -77,8 +78,8 @@ FileStream.prototype._notify = function () {
 
     debug('pushing buffer of length %s', buffer.length)
     self._reading = false
-    self.push(buffer)
-
+    self._backPreasure = self.push(buffer)
+    if (!self._backPreasure && self._missing !== 0) self.emit('piece', self._piece);
     if (self._missing === 0) self.push(null)
   })
   self._piece += 1
@@ -93,7 +94,7 @@ FileStream.prototype._destroy = function (err, onclose) {
   this.destroyed = true
 
   if (!this._torrent.destroyed) {
-    this._torrent.deselect(this._startPiece, this._endPiece, true)
+    this._torrent.deselect(this._startPiece, this._endPiece, false)
   }
 
   if (err) this.emit('error', err)

lib/file.js

@@ -80,14 +80,30 @@ File.prototype.createReadStream = function (opts) {
     return empty
   }
 
+  var pieceLength = this._torrent.pieceLength
+
+  var selectLength =  Math.min((5 * 1024 * 1024 / pieceLength) | 0, 2);
+
   var fileStream = new FileStream(self, opts)
-  self._torrent.select(fileStream._startPiece, fileStream._endPiece, true, function () {
-    fileStream._notify()
-  })
+  function select (start) {
+    if (start > fileStream._endPiece) return;
+    var end = (start + selectLength < fileStream._endPiece)
+      ? start + selectLength
+      : fileStream._endPiece
+    self._torrent.deselect(fileStream._startPiece, fileStream._endPiece, false)
+    self._torrent.select(start, end, true, function () {
+      fileStream._notify()
+    })
+  }
+  fileStream.on('piece', function (p) {
+    select(p);
+  });
+  select(fileStream._startPiece)
+
   eos(fileStream, function () {
     if (self._destroyed) return
     if (!self._torrent.destroyed) {
-      self._torrent.deselect(fileStream._startPiece, fileStream._endPiece, true)
+      self._torrent.deselect(fileStream._startPiece, fileStream._endPiece, false)
     }
   })
   return fileStream