implementation.js 2.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106
  1. 'use strict';
  2. // Ported from https://github.com/nodejs/node/blob/master/lib/internal/streams/pipeline.js
  3. // which was ported from https://github.com/mafintosh/pump with
  4. // permission from the author, Mathias Buus (@mafintosh).
  5. var eos = require('stream.finished');
  6. function once(callback) {
  7. var called = false;
  8. return function(err) {
  9. if (called) return;
  10. called = true;
  11. callback(err);
  12. };
  13. }
  14. function noop(err) {
  15. // Rethrow the error if it exists to avoid swallowing it
  16. if (err) throw err;
  17. }
  18. function isRequest(stream) {
  19. return stream.setHeader && typeof stream.abort === 'function';
  20. }
  21. function destroyer(stream, reading, writing, callback) {
  22. callback = once(callback);
  23. var closed = false;
  24. stream.on('close', function() {
  25. closed = true;
  26. });
  27. eos(stream, {readable: reading, writable: writing}, function(err) {
  28. if (err) return callback(err);
  29. closed = true;
  30. callback();
  31. });
  32. var destroyed = false;
  33. return function(err) {
  34. if (closed) return;
  35. if (destroyed) return;
  36. destroyed = true;
  37. // request.destroy just do .end - .abort is what we want
  38. if (isRequest(stream)) return stream.abort();
  39. if (typeof stream.destroy === 'function') return stream.destroy();
  40. if (!err) {
  41. err = new Error('Cannot call pipe after a stream was destroyed');
  42. err.code = 'ERR_STREAM_DESTROYED';
  43. err.name = 'Error [ERR_STREAM_DESTROYED]';
  44. }
  45. callback(err);
  46. };
  47. }
  48. function call(fn) {
  49. fn();
  50. }
  51. function pipe(from, to) {
  52. return from.pipe(to);
  53. }
  54. function popCallback(streams) {
  55. if (!streams.length) return noop;
  56. if (typeof streams[streams.length - 1] !== 'function') return noop;
  57. return streams.pop();
  58. }
  59. function pipeline() {
  60. for (var len = arguments.length, streams = Array(len), key = 0; key < len; key++) {
  61. streams[key] = arguments[key];
  62. }
  63. var callback = popCallback(streams);
  64. if (Array.isArray(streams[0])) streams = streams[0];
  65. if (streams.length < 2) {
  66. var argsError = new Error('The "streams" argument must be specified');
  67. argsError.code = 'ERR_MISSING_ARGS';
  68. argsError.name = 'TypeError [ERR_MISSING_ARGS]';
  69. throw argsError;
  70. }
  71. var error;
  72. var destroys = streams.map(function(stream, i) {
  73. var reading = i < streams.length - 1;
  74. var writing = i > 0;
  75. return destroyer(stream, reading, writing, function(err) {
  76. if (!error) error = err;
  77. if (err) destroys.forEach(call);
  78. if (reading) return;
  79. destroys.forEach(call);
  80. callback(error);
  81. });
  82. });
  83. return streams.reduce(pipe);
  84. }
  85. module.exports = pipeline;