strong-tie/inbound-calls

1var stream = require('readable-stream')2var eos = require('end-of-stream')3var inherits = require('inherits')4var shift = require('stream-shift')5 6var SIGNAL_FLUSH = (Buffer.from && Buffer.from !== Uint8Array.from)7  ? Buffer.from([0])8  : new Buffer([0])9 10var onuncork = function(self, fn) {11  if (self._corked) self.once('uncork', fn)12  else fn()13}14 15var autoDestroy = function (self, err) {16  if (self._autoDestroy) self.destroy(err)17}18 19var destroyer = function(self, end) {20  return function(err) {21    if (err) autoDestroy(self, err.message === 'premature close' ? null : err)22    else if (end && !self._ended) self.end()23  }24}25 26var end = function(ws, fn) {27  if (!ws) return fn()28  if (ws._writableState && ws._writableState.finished) return fn()29  if (ws._writableState) return ws.end(fn)30  ws.end()31  fn()32}33 34var noop = function() {}35 36var toStreams2 = function(rs) {37  return new (stream.Readable)({objectMode:true, highWaterMark:16}).wrap(rs)38}39 40var Duplexify = function(writable, readable, opts) {41  if (!(this instanceof Duplexify)) return new Duplexify(writable, readable, opts)42  stream.Duplex.call(this, opts)43 44  this._writable = null45  this._readable = null46  this._readable2 = null47 48  this._autoDestroy = !opts || opts.autoDestroy !== false49  this._forwardDestroy = !opts || opts.destroy !== false50  this._forwardEnd = !opts || opts.end !== false51  this._corked = 1 // start corked52  this._ondrain = null53  this._drained = false54  this._forwarding = false55  this._unwrite = null56  this._unread = null57  this._ended = false58 59  this.destroyed = false60 61  if (writable) this.setWritable(writable)62  if (readable) this.setReadable(readable)63}64 65inherits(Duplexify, stream.Duplex)66 67Duplexify.obj = function(writable, readable, opts) {68  if (!opts) opts = {}69  opts.objectMode = true70  opts.highWaterMark = 1671  return new Duplexify(writable, readable, opts)72}73 74Duplexify.prototype.cork = function() {75  if (++this._corked === 1) this.emit('cork')76}77 78Duplexify.prototype.uncork = function() {79  if (this._corked && --this._corked === 0) this.emit('uncork')80}81 82Duplexify.prototype.setWritable = function(writable) {83  if (this._unwrite) this._unwrite()84 85  if (this.destroyed) {86    if (writable && writable.destroy) writable.destroy()87    return88  }89 90  if (writable === null || writable === false) {91    this.end()92    return93  }94 95  var self = this96  var unend = eos(writable, {writable:true, readable:false}, destroyer(this, this._forwardEnd))97 98  var ondrain = function() {99    var ondrain = self._ondrain100    self._ondrain = null101    if (ondrain) ondrain()102  }103 104  var clear = function() {105    self._writable.removeListener('drain', ondrain)106    unend()107  }108 109  if (this._unwrite) process.nextTick(ondrain) // force a drain on stream reset to avoid livelocks110 111  this._writable = writable112  this._writable.on('drain', ondrain)113  this._unwrite = clear114 115  this.uncork() // always uncork setWritable116}117 118Duplexify.prototype.setReadable = function(readable) {119  if (this._unread) this._unread()120 121  if (this.destroyed) {122    if (readable && readable.destroy) readable.destroy()123    return124  }125 126  if (readable === null || readable === false) {127    this.push(null)128    this.resume()129    return130  }131 132  var self = this133  var unend = eos(readable, {writable:false, readable:true}, destroyer(this))134 135  var onreadable = function() {136    self._forward()137  }138 139  var onend = function() {140    self.push(null)141  }142 143  var clear = function() {144    self._readable2.removeListener('readable', onreadable)145    self._readable2.removeListener('end', onend)146    unend()147  }148 149  this._drained = true150  this._readable = readable151  this._readable2 = readable._readableState ? readable : toStreams2(readable)152  this._readable2.on('readable', onreadable)153  this._readable2.on('end', onend)154  this._unread = clear155 156  this._forward()157}158 159Duplexify.prototype._read = function() {160  this._drained = true161  this._forward()162}163 164Duplexify.prototype._forward = function() {165  if (this._forwarding || !this._readable2 || !this._drained) return166  this._forwarding = true167 168  var data169 170  while (this._drained && (data = shift(this._readable2)) !== null) {171    if (this.destroyed) continue172    this._drained = this.push(data)173  }174 175  this._forwarding = false176}177 178Duplexify.prototype.destroy = function(err, cb) {179  if (!cb) cb = noop180  if (this.destroyed) return cb(null)181  this.destroyed = true182 183  var self = this184  process.nextTick(function() {185    self._destroy(err)186    cb(null)187  })188}189 190Duplexify.prototype._destroy = function(err) {191  if (err) {192    var ondrain = this._ondrain193    this._ondrain = null194    if (ondrain) ondrain(err)195    else this.emit('error', err)196  }197 198  if (this._forwardDestroy) {199    if (this._readable && this._readable.destroy) this._readable.destroy()200    if (this._writable && this._writable.destroy) this._writable.destroy()201  }202 203  this.emit('close')204}205 206Duplexify.prototype._write = function(data, enc, cb) {207  if (this.destroyed) return208  if (this._corked) return onuncork(this, this._write.bind(this, data, enc, cb))209  if (data === SIGNAL_FLUSH) return this._finish(cb)210  if (!this._writable) return cb()211 212  if (this._writable.write(data) === false) this._ondrain = cb213  else if (!this.destroyed) cb()214}215 216Duplexify.prototype._finish = function(cb) {217  var self = this218  this.emit('preend')219  onuncork(this, function() {220    end(self._forwardEnd && self._writable, function() {221      // haxx to not emit prefinish twice222      if (self._writableState.prefinished === false) self._writableState.prefinished = true223      self.emit('prefinish')224      onuncork(self, cb)225    })226  })227}228 229Duplexify.prototype.end = function(data, enc, cb) {230  if (typeof data === 'function') return this.end(null, null, data)231  if (typeof enc === 'function') return this.end(data, null, enc)232  this._ended = true233  if (data) this.write(data)234  if (!this._writableState.ending && !this._writableState.destroyed) this.write(SIGNAL_FLUSH)235  return stream.Writable.prototype.end.call(this, cb)236}237 238module.exports = Duplexify239