| // 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() |
| } |