index.js 2.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. var once = require('once');
  2. var noop = function() {};
  3. var isRequest = function(stream) {
  4. return stream.setHeader && typeof stream.abort === 'function';
  5. };
  6. var isChildProcess = function(stream) {
  7. return stream.stdio && Array.isArray(stream.stdio) && stream.stdio.length === 3
  8. };
  9. var eos = function(stream, opts, callback) {
  10. if (typeof opts === 'function') return eos(stream, null, opts);
  11. if (!opts) opts = {};
  12. callback = once(callback || noop);
  13. var ws = stream._writableState;
  14. var rs = stream._readableState;
  15. var readable = opts.readable || (opts.readable !== false && stream.readable);
  16. var writable = opts.writable || (opts.writable !== false && stream.writable);
  17. var cancelled = false;
  18. var onlegacyfinish = function() {
  19. if (!stream.writable) onfinish();
  20. };
  21. var onfinish = function() {
  22. writable = false;
  23. if (!readable) callback.call(stream);
  24. };
  25. var onend = function() {
  26. readable = false;
  27. if (!writable) callback.call(stream);
  28. };
  29. var onexit = function(exitCode) {
  30. callback.call(stream, exitCode ? new Error('exited with error code: ' + exitCode) : null);
  31. };
  32. var onerror = function(err) {
  33. callback.call(stream, err);
  34. };
  35. var onclose = function() {
  36. process.nextTick(onclosenexttick);
  37. };
  38. var onclosenexttick = function() {
  39. if (cancelled) return;
  40. if (readable && !(rs && (rs.ended && !rs.destroyed))) return callback.call(stream, new Error('premature close'));
  41. if (writable && !(ws && (ws.ended && !ws.destroyed))) return callback.call(stream, new Error('premature close'));
  42. };
  43. var onrequest = function() {
  44. stream.req.on('finish', onfinish);
  45. };
  46. if (isRequest(stream)) {
  47. stream.on('complete', onfinish);
  48. stream.on('abort', onclose);
  49. if (stream.req) onrequest();
  50. else stream.on('request', onrequest);
  51. } else if (writable && !ws) { // legacy streams
  52. stream.on('end', onlegacyfinish);
  53. stream.on('close', onlegacyfinish);
  54. }
  55. if (isChildProcess(stream)) stream.on('exit', onexit);
  56. stream.on('end', onend);
  57. stream.on('finish', onfinish);
  58. if (opts.error !== false) stream.on('error', onerror);
  59. stream.on('close', onclose);
  60. return function() {
  61. cancelled = true;
  62. stream.removeListener('complete', onfinish);
  63. stream.removeListener('abort', onclose);
  64. stream.removeListener('request', onrequest);
  65. if (stream.req) stream.req.removeListener('finish', onfinish);
  66. stream.removeListener('end', onlegacyfinish);
  67. stream.removeListener('close', onlegacyfinish);
  68. stream.removeListener('finish', onfinish);
  69. stream.removeListener('exit', onexit);
  70. stream.removeListener('end', onend);
  71. stream.removeListener('error', onerror);
  72. stream.removeListener('close', onclose);
  73. };
  74. };
  75. module.exports = eos;