file-reader.js 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150
  1. // Basically just a wrapper around an fs.ReadStream
  2. module.exports = FileReader
  3. var fs = require('graceful-fs')
  4. var inherits = require('inherits')
  5. var Reader = require('./reader.js')
  6. var EOF = {EOF: true}
  7. var CLOSE = {CLOSE: true}
  8. inherits(FileReader, Reader)
  9. function FileReader (props) {
  10. // console.error(" FR create", props.path, props.size, new Error().stack)
  11. var self = this
  12. if (!(self instanceof FileReader)) {
  13. throw new Error('FileReader must be called as constructor.')
  14. }
  15. // should already be established as a File type
  16. // XXX Todo: preserve hardlinks by tracking dev+inode+nlink,
  17. // with a HardLinkReader class.
  18. if (!((props.type === 'Link' && props.Link) ||
  19. (props.type === 'File' && props.File))) {
  20. throw new Error('Non-file type ' + props.type)
  21. }
  22. self._buffer = []
  23. self._bytesEmitted = 0
  24. Reader.call(self, props)
  25. }
  26. FileReader.prototype._getStream = function () {
  27. var self = this
  28. var stream = self._stream = fs.createReadStream(self._path, self.props)
  29. if (self.props.blksize) {
  30. stream.bufferSize = self.props.blksize
  31. }
  32. stream.on('open', self.emit.bind(self, 'open'))
  33. stream.on('data', function (c) {
  34. // console.error('\t\t%d %s', c.length, self.basename)
  35. self._bytesEmitted += c.length
  36. // no point saving empty chunks
  37. if (!c.length) {
  38. return
  39. } else if (self._paused || self._buffer.length) {
  40. self._buffer.push(c)
  41. self._read()
  42. } else self.emit('data', c)
  43. })
  44. stream.on('end', function () {
  45. if (self._paused || self._buffer.length) {
  46. // console.error('FR Buffering End', self._path)
  47. self._buffer.push(EOF)
  48. self._read()
  49. } else {
  50. self.emit('end')
  51. }
  52. if (self._bytesEmitted !== self.props.size) {
  53. self.error("Didn't get expected byte count\n" +
  54. 'expect: ' + self.props.size + '\n' +
  55. 'actual: ' + self._bytesEmitted)
  56. }
  57. })
  58. stream.on('close', function () {
  59. if (self._paused || self._buffer.length) {
  60. // console.error('FR Buffering Close', self._path)
  61. self._buffer.push(CLOSE)
  62. self._read()
  63. } else {
  64. // console.error('FR close 1', self._path)
  65. self.emit('close')
  66. }
  67. })
  68. stream.on('error', function (e) {
  69. self.emit('error', e)
  70. })
  71. self._read()
  72. }
  73. FileReader.prototype._read = function () {
  74. var self = this
  75. // console.error('FR _read', self._path)
  76. if (self._paused) {
  77. // console.error('FR _read paused', self._path)
  78. return
  79. }
  80. if (!self._stream) {
  81. // console.error('FR _getStream calling', self._path)
  82. return self._getStream()
  83. }
  84. // clear out the buffer, if there is one.
  85. if (self._buffer.length) {
  86. // console.error('FR _read has buffer', self._buffer.length, self._path)
  87. var buf = self._buffer
  88. for (var i = 0, l = buf.length; i < l; i++) {
  89. var c = buf[i]
  90. if (c === EOF) {
  91. // console.error('FR Read emitting buffered end', self._path)
  92. self.emit('end')
  93. } else if (c === CLOSE) {
  94. // console.error('FR Read emitting buffered close', self._path)
  95. self.emit('close')
  96. } else {
  97. // console.error('FR Read emitting buffered data', self._path)
  98. self.emit('data', c)
  99. }
  100. if (self._paused) {
  101. // console.error('FR Read Re-pausing at '+i, self._path)
  102. self._buffer = buf.slice(i)
  103. return
  104. }
  105. }
  106. self._buffer.length = 0
  107. }
  108. // console.error("FR _read done")
  109. // that's about all there is to it.
  110. }
  111. FileReader.prototype.pause = function (who) {
  112. var self = this
  113. // console.error('FR Pause', self._path)
  114. if (self._paused) return
  115. who = who || self
  116. self._paused = true
  117. if (self._stream) self._stream.pause()
  118. self.emit('pause', who)
  119. }
  120. FileReader.prototype.resume = function (who) {
  121. var self = this
  122. // console.error('FR Resume', self._path)
  123. if (!self._paused) return
  124. who = who || self
  125. self.emit('resume', who)
  126. self._paused = false
  127. if (self._stream) self._stream.resume()
  128. self._read()
  129. }