1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
const stream = require('stream')
const debugFactory = require('debug')
const eos = require('end-of-stream')
const debug = debugFactory('webtorrent:file-stream')
/**
* Readable stream of a torrent file
*
* @param {File} file
* @param {Object} opts
* @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 {
constructor (file, opts) {
super(opts)
this._torrent = file._torrent
const start = (opts && opts.start) || 0
const end = (opts && opts.end && opts.end < file.length)
? opts.end
: file.length - 1
const pieceLength = file._torrent.pieceLength
this._startPiece = (start + file.offset) / pieceLength | 0
this._endPiece = (end + file.offset) / pieceLength | 0
this._piece = this._startPiece
this._offset = (start + file.offset) - (this._startPiece * pieceLength)
this._missing = end - start + 1
this._reading = false
this._notifying = false
this._criticalLength = Math.min((1024 * 1024 / pieceLength) | 0, 2)
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 () {
if (this._reading) return
this._reading = true
this._notify()
}
_notify () {
if (!this._reading || this._missing === 0) return
if (!this._torrent.bitfield.get(this._piece)) {
return this._torrent.critical(this._piece, this._piece + this._criticalLength)
}
if (this._notifying) return
this._notifying = true
if (this._torrent.destroyed) return this.destroy(new Error('Torrent removed'))
const p = this._piece
const getOpts = {}
// Specify length for the last piece in case it is zero-padded
if (p === this._torrent.pieces.length - 1) {
getOpts.length = this._torrent.lastPieceLength
}
this._torrent.store.get(p, getOpts, (err, buffer) => {
this._notifying = false
if (this.destroyed) return
debug('read %s (length %s) (err %s)', p, buffer && buffer.length, err && err.message)
if (err) return this.destroy(err)
if (this._offset) {
buffer = buffer.slice(this._offset)
this._offset = 0
}
if (this._missing < buffer.length) {
buffer = buffer.slice(0, this._missing)
}
this._missing -= buffer.length
debug('pushing buffer of length %s', buffer.length)
this._reading = false
this.push(buffer)
if (this._missing === 0) this.push(null)
})
this._piece += 1
}
_destroy (err, cb) {
if (!this._torrent.destroyed) {
this._torrent.deselect(this._startPiece, this._endPiece, true)
}
cb(err)
}
}
module.exports = FileStream
|
