Merge pull request #2339 from ThaUnknown/streamx

fix: use streamx instead of stream

3 files changed, 13 insertions, 16 deletions

diff --git a/lib/file-stream.js b/lib/file-stream.js
index a3682b1..c344913 100644
--- a/lib/file-stream.js
+++ b/lib/file-stream.js
@@ -1,6 +1,5 @@
-const stream = require('stream')
+const { Readable } = require('streamx')
 const debugFactory = require('debug')
-const eos = require('end-of-stream')
 
 const debug = debugFactory('webtorrent:file-stream')
 
@@ -12,9 +11,9 @@ const debug = debugFactory('webtorrent:file-stream')
  * @param {number} opts.start stream slice of file, starting from this byte (inclusive)
  * @param {number} opts.end stream slice of file, ending with this byte (inclusive)
  */
-class FileStream extends stream.Readable {
+class FileStream extends Readable {
   constructor (file, opts) {
-    super(opts)
+    super(opts ?? {})
 
     this._torrent = file._torrent
 
@@ -39,26 +38,22 @@ class FileStream extends stream.Readable {
     this._torrent.select(this._startPiece, this._endPiece, true, () => {
       this._notify()
     })
-
-    // Ensure that cleanup happens even if destroy() is never called (readable-stream v3 currently doesn't call it automaticallly)
-    eos(this, (err) => {
-      this.destroy(err)
-    })
   }
 
-  _read () {
+  _read (cb) {
     if (this._reading) return
     this._reading = true
-    this._notify()
+    this._notify(cb)
   }
 
-  _notify () {
-    if (!this._reading || this._missing === 0) return
+  _notify (cb = () => {}) {
+    if (!this._reading || this._missing === 0) return cb()
     if (!this._torrent.bitfield.get(this._piece)) {
+      cb()
       return this._torrent.critical(this._piece, this._piece + this._criticalLength)
     }
 
-    if (this._notifying) return
+    if (this._notifying) return cb()
     this._notifying = true
 
     if (this._torrent.destroyed) return this.destroy(new Error('Torrent removed'))
@@ -92,11 +87,12 @@ class FileStream extends stream.Readable {
       this.push(buffer)
 
       if (this._missing === 0) this.push(null)
+      cb()
     })
     this._piece += 1
   }
 
-  _destroy (err, cb) {
+  _destroy (cb, err) {
     if (!this._torrent.destroyed) {
       this._torrent.deselect(this._startPiece, this._endPiece, true)
     }
diff --git a/lib/file.js b/lib/file.js
index 5e37581..e0574d0 100644
--- a/lib/file.js
+++ b/lib/file.js
@@ -1,5 +1,5 @@
 const EventEmitter = require('events')
-const { PassThrough } = require('stream')
+const { PassThrough } = require('streamx')
 const path = require('path')
 const render = require('render-media')
 const { BlobWriteStream } = require('fast-blob-stream')
diff --git a/package.json b/package.json
index fd9fa03..e05f5d6 100644
--- a/package.json
+++ b/package.json
@@ -73,6 +73,7 @@
     "simple-sha1": "^3.1.0",
     "speed-limiter": "^1.0.2",
     "stream-with-known-length-to-buffer": "^1.0.4",
+    "streamx": "^2.12.4",
     "throughput": "^1.0.1",
     "torrent-discovery": "^9.4.13",
     "torrent-piece": "^2.0.1",