94 lines
2.3 KiB
JavaScript
94 lines
2.3 KiB
JavaScript
|
'use strict';
|
||
|
|
const {PassThrough} = require('stream');
|
||
|
|
const duplexer3 = require('duplexer3');
|
||
|
|
const requestAsEventEmitter = require('./request-as-event-emitter');
|
||
|
|
const {HTTPError, ReadError} = require('./errors');
|
||
|
|
|
||
|
|
module.exports = options => {
|
||
|
|
const input = new PassThrough();
|
||
|
|
const output = new PassThrough();
|
||
|
|
const proxy = duplexer3(input, output);
|
||
|
|
const piped = new Set();
|
||
|
|
let isFinished = false;
|
||
|
|
|
||
|
|
options.retry.retries = () => 0;
|
||
|
|
|
||
|
|
if (options.body) {
|
||
|
|
proxy.write = () => {
|
||
|
|
throw new Error('Got\'s stream is not writable when the `body` option is used');
|
||
|
|
};
|
||
|
|
}
|
||
|
|
|
||
|
|
const emitter = requestAsEventEmitter(options, input);
|
||
|
|
|
||
|
|
// Cancels the request
|
||
|
|
proxy._destroy = emitter.abort;
|
||
|
|
|
||
|
|
emitter.on('response', response => {
|
||
|
|
const {statusCode} = response;
|
||
|
|
|
||
|
|
response.on('error', error => {
|
||
|
|
proxy.emit('error', new ReadError(error, options));
|
||
|
|
});
|
||
|
|
|
||
|
|
if (options.throwHttpErrors && statusCode !== 304 && (statusCode < 200 || statusCode > 299)) {
|
||
|
|
proxy.emit('error', new HTTPError(response, options), null, response);
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
|
||
|
|
isFinished = true;
|
||
|
|
|
||
|
|
response.pipe(output);
|
||
|
|
|
||
|
|
for (const destination of piped) {
|
||
|
|
if (destination.headersSent) {
|
||
|
|
continue;
|
||
|
|
}
|
||
|
|
|
||
|
|
for (const [key, value] of Object.entries(response.headers)) {
|
||
|
|
// Got gives *decompressed* data. Overriding `content-encoding` header would result in an error.
|
||
|
|
// It's not possible to decompress already decompressed data, is it?
|
||
|
|
const allowed = options.decompress ? key !== 'content-encoding' : true;
|
||
|
|
if (allowed) {
|
||
|
|
destination.setHeader(key, value);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
destination.statusCode = response.statusCode;
|
||
|
|
}
|
||
|
|
|
||
|
|
proxy.emit('response', response);
|
||
|
|
});
|
||
|
|
|
||
|
|
[
|
||
|
|
'error',
|
||
|
|
'request',
|
||
|
|
'redirect',
|
||
|
|
'uploadProgress',
|
||
|
|
'downloadProgress'
|
||
|
|
].forEach(event => emitter.on(event, (...args) => proxy.emit(event, ...args)));
|
||
|
|
|
||
|
|
const pipe = proxy.pipe.bind(proxy);
|
||
|
|
const unpipe = proxy.unpipe.bind(proxy);
|
||
|
|
proxy.pipe = (destination, options) => {
|
||
|
|
if (isFinished) {
|
||
|
|
throw new Error('Failed to pipe. The response has been emitted already.');
|
||
|
|
}
|
||
|
|
|
||
|
|
const result = pipe(destination, options);
|
||
|
|
|
||
|
|
if (Reflect.has(destination, 'setHeader')) {
|
||
|
|
piped.add(destination);
|
||
|
|
}
|
||
|
|
|
||
|
|
return result;
|
||
|
|
};
|
||
|
|
|
||
|
|
proxy.unpipe = stream => {
|
||
|
|
piped.delete(stream);
|
||
|
|
return unpipe(stream);
|
||
|
|
};
|
||
|
|
|
||
|
|
return proxy;
|
||
|
|
};
|