as-stream.js 2.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293
  1. 'use strict';
  2. const {PassThrough} = require('stream');
  3. const duplexer3 = require('duplexer3');
  4. const requestAsEventEmitter = require('./request-as-event-emitter');
  5. const {HTTPError, ReadError} = require('./errors');
  6. module.exports = options => {
  7. const input = new PassThrough();
  8. const output = new PassThrough();
  9. const proxy = duplexer3(input, output);
  10. const piped = new Set();
  11. let isFinished = false;
  12. options.retry.retries = () => 0;
  13. if (options.body) {
  14. proxy.write = () => {
  15. throw new Error('Got\'s stream is not writable when the `body` option is used');
  16. };
  17. }
  18. const emitter = requestAsEventEmitter(options, input);
  19. // Cancels the request
  20. proxy._destroy = emitter.abort;
  21. emitter.on('response', response => {
  22. const {statusCode} = response;
  23. response.on('error', error => {
  24. proxy.emit('error', new ReadError(error, options));
  25. });
  26. if (options.throwHttpErrors && statusCode !== 304 && (statusCode < 200 || statusCode > 299)) {
  27. proxy.emit('error', new HTTPError(response, options), null, response);
  28. return;
  29. }
  30. isFinished = true;
  31. response.pipe(output);
  32. for (const destination of piped) {
  33. if (destination.headersSent) {
  34. continue;
  35. }
  36. for (const [key, value] of Object.entries(response.headers)) {
  37. // Got gives *decompressed* data. Overriding `content-encoding` header would result in an error.
  38. // It's not possible to decompress already decompressed data, is it?
  39. const allowed = options.decompress ? key !== 'content-encoding' : true;
  40. if (allowed) {
  41. destination.setHeader(key, value);
  42. }
  43. }
  44. destination.statusCode = response.statusCode;
  45. }
  46. proxy.emit('response', response);
  47. });
  48. [
  49. 'error',
  50. 'request',
  51. 'redirect',
  52. 'uploadProgress',
  53. 'downloadProgress'
  54. ].forEach(event => emitter.on(event, (...args) => proxy.emit(event, ...args)));
  55. const pipe = proxy.pipe.bind(proxy);
  56. const unpipe = proxy.unpipe.bind(proxy);
  57. proxy.pipe = (destination, options) => {
  58. if (isFinished) {
  59. throw new Error('Failed to pipe. The response has been emitted already.');
  60. }
  61. const result = pipe(destination, options);
  62. if (Reflect.has(destination, 'setHeader')) {
  63. piped.add(destination);
  64. }
  65. return result;
  66. };
  67. proxy.unpipe = stream => {
  68. piped.delete(stream);
  69. return unpipe(stream);
  70. };
  71. return proxy;
  72. };