1// Ported from https://github.com/mafintosh/pump with 2// permission from the author, Mathias Buus (@mafintosh). 3'use strict'; 4 5var eos; 6 7function once(callback) { 8 var called = false; 9 return function () { 10 if (called) return; 11 called = true; 12 callback.apply(void 0, arguments); 13 }; 14} 15 16var _require$codes = require('../../../errors').codes, 17 ERR_MISSING_ARGS = _require$codes.ERR_MISSING_ARGS, 18 ERR_STREAM_DESTROYED = _require$codes.ERR_STREAM_DESTROYED; 19 20function noop(err) { 21 // Rethrow the error if it exists to avoid swallowing it 22 if (err) throw err; 23} 24 25function isRequest(stream) { 26 return stream.setHeader && typeof stream.abort === 'function'; 27} 28 29function destroyer(stream, reading, writing, callback) { 30 callback = once(callback); 31 var closed = false; 32 stream.on('close', function () { 33 closed = true; 34 }); 35 if (eos === undefined) eos = require('./end-of-stream'); 36 eos(stream, { 37 readable: reading, 38 writable: writing 39 }, function (err) { 40 if (err) return callback(err); 41 closed = true; 42 callback(); 43 }); 44 var destroyed = false; 45 return function (err) { 46 if (closed) return; 47 if (destroyed) return; 48 destroyed = true; // request.destroy just do .end - .abort is what we want 49 50 if (isRequest(stream)) return stream.abort(); 51 if (typeof stream.destroy === 'function') return stream.destroy(); 52 callback(err || new ERR_STREAM_DESTROYED('pipe')); 53 }; 54} 55 56function call(fn) { 57 fn(); 58} 59 60function pipe(from, to) { 61 return from.pipe(to); 62} 63 64function popCallback(streams) { 65 if (!streams.length) return noop; 66 if (typeof streams[streams.length - 1] !== 'function') return noop; 67 return streams.pop(); 68} 69 70function pipeline() { 71 for (var _len = arguments.length, streams = new Array(_len), _key = 0; _key < _len; _key++) { 72 streams[_key] = arguments[_key]; 73 } 74 75 var callback = popCallback(streams); 76 if (Array.isArray(streams[0])) streams = streams[0]; 77 78 if (streams.length < 2) { 79 throw new ERR_MISSING_ARGS('streams'); 80 } 81 82 var error; 83 var destroys = streams.map(function (stream, i) { 84 var reading = i < streams.length - 1; 85 var writing = i > 0; 86 return destroyer(stream, reading, writing, function (err) { 87 if (!error) error = err; 88 if (err) destroys.forEach(call); 89 if (reading) return; 90 destroys.forEach(call); 91 callback(error); 92 }); 93 }); 94 return streams.reduce(pipe); 95} 96 97module.exports = pipeline;