source: trip-planner-front/node_modules/readable-stream/lib/internal/streams/pipeline.js

Last change on this file was 6a3a178, checked in by Ema <ema_spirova@…>, 3 years ago

initial commit

  • Property mode set to 100644
File size: 2.4 KB
RevLine 
[6a3a178]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;
Note: See TracBrowser for help on using the repository browser.