151 lines
3.9 KiB
JavaScript
151 lines
3.9 KiB
JavaScript
// Basically just a wrapper around an fs.ReadStream
|
|
|
|
module.exports = FileReader
|
|
|
|
var fs = require('graceful-fs')
|
|
var inherits = require('inherits')
|
|
var Reader = require('./reader.js')
|
|
var EOF = {EOF: true}
|
|
var CLOSE = {CLOSE: true}
|
|
|
|
inherits(FileReader, Reader)
|
|
|
|
function FileReader (props) {
|
|
// console.error(" FR create", props.path, props.size, new Error().stack)
|
|
var self = this
|
|
if (!(self instanceof FileReader)) {
|
|
throw new Error('FileReader must be called as constructor.')
|
|
}
|
|
|
|
// should already be established as a File type
|
|
// XXX Todo: preserve hardlinks by tracking dev+inode+nlink,
|
|
// with a HardLinkReader class.
|
|
if (!((props.type === 'Link' && props.Link) ||
|
|
(props.type === 'File' && props.File))) {
|
|
throw new Error('Non-file type ' + props.type)
|
|
}
|
|
|
|
self._buffer = []
|
|
self._bytesEmitted = 0
|
|
Reader.call(self, props)
|
|
}
|
|
|
|
FileReader.prototype._getStream = function () {
|
|
var self = this
|
|
var stream = self._stream = fs.createReadStream(self._path, self.props)
|
|
|
|
if (self.props.blksize) {
|
|
stream.bufferSize = self.props.blksize
|
|
}
|
|
|
|
stream.on('open', self.emit.bind(self, 'open'))
|
|
|
|
stream.on('data', function (c) {
|
|
// console.error('\t\t%d %s', c.length, self.basename)
|
|
self._bytesEmitted += c.length
|
|
// no point saving empty chunks
|
|
if (!c.length) {
|
|
return
|
|
} else if (self._paused || self._buffer.length) {
|
|
self._buffer.push(c)
|
|
self._read()
|
|
} else self.emit('data', c)
|
|
})
|
|
|
|
stream.on('end', function () {
|
|
if (self._paused || self._buffer.length) {
|
|
// console.error('FR Buffering End', self._path)
|
|
self._buffer.push(EOF)
|
|
self._read()
|
|
} else {
|
|
self.emit('end')
|
|
}
|
|
|
|
if (self._bytesEmitted !== self.props.size) {
|
|
self.error("Didn't get expected byte count\n" +
|
|
'expect: ' + self.props.size + '\n' +
|
|
'actual: ' + self._bytesEmitted)
|
|
}
|
|
})
|
|
|
|
stream.on('close', function () {
|
|
if (self._paused || self._buffer.length) {
|
|
// console.error('FR Buffering Close', self._path)
|
|
self._buffer.push(CLOSE)
|
|
self._read()
|
|
} else {
|
|
// console.error('FR close 1', self._path)
|
|
self.emit('close')
|
|
}
|
|
})
|
|
|
|
stream.on('error', function (e) {
|
|
self.emit('error', e)
|
|
})
|
|
|
|
self._read()
|
|
}
|
|
|
|
FileReader.prototype._read = function () {
|
|
var self = this
|
|
// console.error('FR _read', self._path)
|
|
if (self._paused) {
|
|
// console.error('FR _read paused', self._path)
|
|
return
|
|
}
|
|
|
|
if (!self._stream) {
|
|
// console.error('FR _getStream calling', self._path)
|
|
return self._getStream()
|
|
}
|
|
|
|
// clear out the buffer, if there is one.
|
|
if (self._buffer.length) {
|
|
// console.error('FR _read has buffer', self._buffer.length, self._path)
|
|
var buf = self._buffer
|
|
for (var i = 0, l = buf.length; i < l; i++) {
|
|
var c = buf[i]
|
|
if (c === EOF) {
|
|
// console.error('FR Read emitting buffered end', self._path)
|
|
self.emit('end')
|
|
} else if (c === CLOSE) {
|
|
// console.error('FR Read emitting buffered close', self._path)
|
|
self.emit('close')
|
|
} else {
|
|
// console.error('FR Read emitting buffered data', self._path)
|
|
self.emit('data', c)
|
|
}
|
|
|
|
if (self._paused) {
|
|
// console.error('FR Read Re-pausing at '+i, self._path)
|
|
self._buffer = buf.slice(i)
|
|
return
|
|
}
|
|
}
|
|
self._buffer.length = 0
|
|
}
|
|
// console.error("FR _read done")
|
|
// that's about all there is to it.
|
|
}
|
|
|
|
FileReader.prototype.pause = function (who) {
|
|
var self = this
|
|
// console.error('FR Pause', self._path)
|
|
if (self._paused) return
|
|
who = who || self
|
|
self._paused = true
|
|
if (self._stream) self._stream.pause()
|
|
self.emit('pause', who)
|
|
}
|
|
|
|
FileReader.prototype.resume = function (who) {
|
|
var self = this
|
|
// console.error('FR Resume', self._path)
|
|
if (!self._paused) return
|
|
who = who || self
|
|
self.emit('resume', who)
|
|
self._paused = false
|
|
if (self._stream) self._stream.resume()
|
|
self._read()
|
|
}
|