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