AK-21/Graphite-Industrial-Intelligence

1var once = require('once')2var eos = require('end-of-stream')3var fs4 5try {6  fs = require('fs') // we only need fs to get the ReadStream and WriteStream prototypes7} catch (e) {}8 9var noop = function () {}10var ancient = typeof process === 'undefined' ? false : /^v?\.0/.test(process.version)11 12var isFn = function (fn) {13  return typeof fn === 'function'14}15 16var isFS = function (stream) {17  if (!ancient) return false // newer node version do not need to care about fs is a special way18  if (!fs) return false // browser19  return (stream instanceof (fs.ReadStream || noop) || stream instanceof (fs.WriteStream || noop)) && isFn(stream.close)20}21 22var isRequest = function (stream) {23  return stream.setHeader && isFn(stream.abort)24}25 26var destroyer = function (stream, reading, writing, callback) {27  callback = once(callback)28 29  var closed = false30  stream.on('close', function () {31    closed = true32  })33 34  eos(stream, {readable: reading, writable: writing}, function (err) {35    if (err) return callback(err)36    closed = true37    callback()38  })39 40  var destroyed = false41  return function (err) {42    if (closed) return43    if (destroyed) return44    destroyed = true45 46    if (isFS(stream)) return stream.close(noop) // use close for fs streams to avoid fd leaks47    if (isRequest(stream)) return stream.abort() // request.destroy just do .end - .abort is what we want48 49    if (isFn(stream.destroy)) return stream.destroy()50 51    callback(err || new Error('stream was destroyed'))52  }53}54 55var call = function (fn) {56  fn()57}58 59var pipe = function (from, to) {60  return from.pipe(to)61}62 63var pump = function () {64  var streams = Array.prototype.slice.call(arguments)65  var callback = isFn(streams[streams.length - 1] || noop) && streams.pop() || noop66 67  if (Array.isArray(streams[0])) streams = streams[0]68  if (streams.length < 2) throw new Error('pump requires two streams per minimum')69 70  var error71  var destroys = streams.map(function (stream, i) {72    var reading = i < streams.length - 173    var writing = i > 074    return destroyer(stream, reading, writing, function (err) {75      if (!error) error = err76      if (err) destroys.forEach(call)77      if (reading) return78      destroys.forEach(call)79      callback(error)80    })81  })82 83  return streams.reduce(pipe)84}85 86module.exports = pump87