collect.js 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970
  1. module.exports = collect
  2. function collect (stream) {
  3. if (stream._collected) return
  4. if (stream._paused) return stream.on('resume', collect.bind(null, stream))
  5. stream._collected = true
  6. stream.pause()
  7. stream.on('data', save)
  8. stream.on('end', save)
  9. var buf = []
  10. function save (b) {
  11. if (typeof b === 'string') b = new Buffer(b)
  12. if (Buffer.isBuffer(b) && !b.length) return
  13. buf.push(b)
  14. }
  15. stream.on('entry', saveEntry)
  16. var entryBuffer = []
  17. function saveEntry (e) {
  18. collect(e)
  19. entryBuffer.push(e)
  20. }
  21. stream.on('proxy', proxyPause)
  22. function proxyPause (p) {
  23. p.pause()
  24. }
  25. // replace the pipe method with a new version that will
  26. // unlock the buffered stuff. if you just call .pipe()
  27. // without a destination, then it'll re-play the events.
  28. stream.pipe = (function (orig) {
  29. return function (dest) {
  30. // console.error(' === open the pipes', dest && dest.path)
  31. // let the entries flow through one at a time.
  32. // Once they're all done, then we can resume completely.
  33. var e = 0
  34. ;(function unblockEntry () {
  35. var entry = entryBuffer[e++]
  36. // console.error(" ==== unblock entry", entry && entry.path)
  37. if (!entry) return resume()
  38. entry.on('end', unblockEntry)
  39. if (dest) dest.add(entry)
  40. else stream.emit('entry', entry)
  41. })()
  42. function resume () {
  43. stream.removeListener('entry', saveEntry)
  44. stream.removeListener('data', save)
  45. stream.removeListener('end', save)
  46. stream.pipe = orig
  47. if (dest) stream.pipe(dest)
  48. buf.forEach(function (b) {
  49. if (b) stream.emit('data', b)
  50. else stream.emit('end')
  51. })
  52. stream.resume()
  53. }
  54. return dest
  55. }
  56. })(stream.pipe)
  57. }