index.js 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153
  1. 'use strict'
  2. var PassThrough = require('readable-stream').PassThrough
  3. var inherits = require('inherits')
  4. var p = require('process-nextick-args')
  5. function Cloneable (stream, opts) {
  6. if (!(this instanceof Cloneable)) {
  7. return new Cloneable(stream, opts)
  8. }
  9. var objectMode = stream._readableState.objectMode
  10. this._original = stream
  11. this._clonesCount = 1
  12. opts = opts || {}
  13. opts.objectMode = objectMode
  14. PassThrough.call(this, opts)
  15. forwardDestroy(stream, this)
  16. this.on('newListener', onData)
  17. this.once('resume', onResume)
  18. this._hasListener = true
  19. }
  20. inherits(Cloneable, PassThrough)
  21. function onData (event, listener) {
  22. if (event === 'data' || event === 'readable') {
  23. this._hasListener = false
  24. this.removeListener('newListener', onData)
  25. this.removeListener('resume', onResume)
  26. p.nextTick(clonePiped, this)
  27. }
  28. }
  29. function onResume () {
  30. this._hasListener = false
  31. this.removeListener('newListener', onData)
  32. p.nextTick(clonePiped, this)
  33. }
  34. Cloneable.prototype.clone = function () {
  35. if (!this._original) {
  36. throw new Error('already started')
  37. }
  38. this._clonesCount++
  39. // the events added by the clone should not count
  40. // for starting the flow
  41. this.removeListener('newListener', onData)
  42. var clone = new Clone(this)
  43. if (this._hasListener) {
  44. this.on('newListener', onData)
  45. }
  46. return clone
  47. }
  48. Cloneable.prototype._destroy = function (err, cb) {
  49. if (!err) {
  50. this.push(null)
  51. this.end()
  52. this.emit('close')
  53. }
  54. p.nextTick(cb, err)
  55. }
  56. function forwardDestroy (src, dest) {
  57. src.on('error', destroy)
  58. src.on('close', onClose)
  59. function destroy (err) {
  60. src.removeListener('close', onClose)
  61. dest.destroy(err)
  62. }
  63. function onClose () {
  64. dest.end()
  65. }
  66. }
  67. function clonePiped (that) {
  68. if (--that._clonesCount === 0 && !that._readableState.destroyed) {
  69. that._original.pipe(that)
  70. that._original = undefined
  71. }
  72. }
  73. function Clone (parent, opts) {
  74. if (!(this instanceof Clone)) {
  75. return new Clone(parent, opts)
  76. }
  77. var objectMode = parent._readableState.objectMode
  78. opts = opts || {}
  79. opts.objectMode = objectMode
  80. this.parent = parent
  81. PassThrough.call(this, opts)
  82. forwardDestroy(parent, this)
  83. parent.pipe(this)
  84. // the events added by the clone should not count
  85. // for starting the flow
  86. // so we add the newListener handle after we are done
  87. this.on('newListener', onDataClone)
  88. this.on('resume', onResumeClone)
  89. }
  90. function onDataClone (event, listener) {
  91. // We start the flow once all clones are piped or destroyed
  92. if (event === 'data' || event === 'readable' || event === 'close') {
  93. p.nextTick(clonePiped, this.parent)
  94. this.removeListener('newListener', onDataClone)
  95. }
  96. }
  97. function onResumeClone () {
  98. this.removeListener('newListener', onDataClone)
  99. p.nextTick(clonePiped, this.parent)
  100. }
  101. inherits(Clone, PassThrough)
  102. Clone.prototype.clone = function () {
  103. return this.parent.clone()
  104. }
  105. Cloneable.isCloneable = function (stream) {
  106. return stream instanceof Cloneable || stream instanceof Clone
  107. }
  108. Clone.prototype._destroy = function (err, cb) {
  109. if (!err) {
  110. this.push(null)
  111. this.end()
  112. this.emit('close')
  113. }
  114. p.nextTick(cb, err)
  115. }
  116. module.exports = Cloneable