index.js 2.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. 'use strict'
  2. var through2 = require('through2')
  3. var inherits = require('inherits')
  4. var nextTick = require('process-nextick-args')
  5. var Ctor = through2.ctor()
  6. function Cloneable (stream, opts) {
  7. if (!(this instanceof Cloneable)) {
  8. return new Cloneable(stream, opts)
  9. }
  10. var objectMode = stream._readableState.objectMode
  11. this._original = stream
  12. this._clonesCount = 1
  13. opts = opts || {}
  14. opts.objectMode = objectMode
  15. Ctor.call(this, opts)
  16. forwardDestroy(stream, this)
  17. this.on('newListener', onData)
  18. }
  19. inherits(Cloneable, Ctor)
  20. function onData (event, listener) {
  21. if (event === 'data' || event === 'readable') {
  22. this.removeListener('newListener', onData)
  23. nextTick(clonePiped, this)
  24. }
  25. }
  26. Cloneable.prototype.clone = function () {
  27. if (!this._original) {
  28. throw new Error('already started')
  29. }
  30. this._clonesCount++
  31. // the events added by the clone should not count
  32. // for starting the flow
  33. this.removeListener('newListener', onData)
  34. var clone = new Clone(this)
  35. this.on('newListener', onData)
  36. return clone
  37. }
  38. function forwardDestroy (src, dest) {
  39. src.on('error', destroy)
  40. src.on('close', destroy)
  41. function destroy (err) {
  42. dest.destroy(err)
  43. }
  44. }
  45. function clonePiped (that) {
  46. if (--that._clonesCount === 0 && !that._destroyed) {
  47. that._original.pipe(that)
  48. that._original = undefined
  49. }
  50. }
  51. function Clone (parent, opts) {
  52. if (!(this instanceof Clone)) {
  53. return new Clone(parent, opts)
  54. }
  55. var objectMode = parent._readableState.objectMode
  56. opts = opts || {}
  57. opts.objectMode = objectMode
  58. this.parent = parent
  59. Ctor.call(this, opts)
  60. forwardDestroy(this.parent, this)
  61. parent.pipe(this)
  62. // the events added by the clone should not count
  63. // for starting the flow
  64. // so we add the newListener handle after we are done
  65. this.on('newListener', onDataClone)
  66. }
  67. function onDataClone (event, listener) {
  68. // We start the flow once all clones are piped or destroyed
  69. if (event === 'data' || event === 'readable' || event === 'close') {
  70. nextTick(clonePiped, this.parent)
  71. this.removeListener('newListener', onDataClone)
  72. }
  73. }
  74. inherits(Clone, Ctor)
  75. Clone.prototype.clone = function () {
  76. return this.parent.clone()
  77. }
  78. Cloneable.isCloneable = function (stream) {
  79. return stream instanceof Cloneable || stream instanceof Clone
  80. }
  81. module.exports = Cloneable