-
Notifications
You must be signed in to change notification settings - Fork 3.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Showing
9 changed files
with
263 additions
and
327 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,199 +1,202 @@ | ||
| 'use strict' | ||
|
|
||
| const http = require('http') | ||
| const https = require('https') | ||
| const net = require('net') | ||
| const tls = require('tls') | ||
| const { once } = require('events') | ||
| const { createTimeout, abortRace, urlify, appendPort, cacheAgent } = require('./util') | ||
| const timers = require('timers/promises') | ||
| const { normalizeOptions, cacheOptions } = require('./options') | ||
| const { getProxy, getProxyType, proxyCache } = require('./proxy.js') | ||
| const { getProxy, getProxyAgent, proxyCache } = require('./proxy.js') | ||
| const Errors = require('./errors.js') | ||
| const { Agent: AgentBase } = require('agent-base') | ||
|
|
||
| const createAgent = (base, name) => { | ||
| const SECURE = base === https | ||
| const SOCKET_TYPE = SECURE ? tls : net | ||
| module.exports = class Agent extends AgentBase { | ||
| #options | ||
| #timeouts | ||
| #proxy | ||
| #noProxy | ||
| #ProxyAgent | ||
|
|
||
| const agent = class extends base.Agent { | ||
| #options | ||
| #timeouts | ||
| #proxy | ||
| #socket | ||
| constructor (options = {}) { | ||
| const { timeouts, proxy, noProxy, ...normalizedOptions } = normalizeOptions(options) | ||
|
|
||
| constructor (_options) { | ||
| const { timeouts, proxy, noProxy, ...options } = normalizeOptions(_options) | ||
| super(normalizedOptions) | ||
|
|
||
| super(options) | ||
| this.#options = normalizedOptions | ||
| this.#timeouts = timeouts | ||
|
|
||
| this.#options = options | ||
| this.#timeouts = timeouts | ||
| this.#proxy = proxy ? { proxies: getProxyType(proxy), proxy: urlify(proxy), noProxy } : null | ||
| if (proxy) { | ||
| this.#proxy = new URL(proxy) | ||
| this.#noProxy = noProxy | ||
| this.#ProxyAgent = getProxyAgent(proxy) | ||
| } | ||
| } | ||
|
|
||
| get proxy () { | ||
| return this.#proxy ? { url: this.#proxy.proxy } : {} | ||
| } | ||
| get proxy () { | ||
| return this.#proxy ? { url: this.#proxy } : {} | ||
| } | ||
|
|
||
| #getProxy (options) { | ||
| const proxy = this.#proxy | ||
| ? getProxy(appendPort(`${options.protocol}//${options.host}`, options.port), this.#proxy) | ||
| : null | ||
| #getProxy (options) { | ||
| if (!this.#proxy) { | ||
| return | ||
| } | ||
|
|
||
| if (!proxy) { | ||
| return | ||
| } | ||
| const proxy = getProxy(`${options.protocol}//${options.host}:${options.port}`, { | ||
| proxy: this.#proxy, | ||
| noProxy: this.#noProxy, | ||
| }) | ||
|
|
||
| return cacheAgent({ | ||
| key: cacheOptions({ | ||
| ...options, | ||
| ...this.#options, | ||
| secure: SECURE, | ||
| timeouts: this.#timeouts, | ||
| proxy, | ||
| }), | ||
| cache: proxyCache, | ||
| secure: SECURE, | ||
| proxies: this.#proxy.proxies, | ||
| }, proxy, this.#options) | ||
| if (!proxy) { | ||
| return | ||
| } | ||
|
|
||
| #setKeepAlive (socket) { | ||
| socket.setKeepAlive(this.keepAlive, this.keepAliveMsecs) | ||
| socket.setNoDelay(this.keepAlive) | ||
| } | ||
| const cacheKey = cacheOptions({ | ||
| ...options, | ||
| ...this.#options, | ||
| timeouts: this.#timeouts, | ||
| proxy, | ||
| }) | ||
|
|
||
| #setIdleTimeout (socket, options) { | ||
| if (this.#timeouts.idle) { | ||
| socket.setTimeout(this.#timeouts.idle, () => { | ||
| socket.destroy(new Errors.IdleTimeoutError(options)) | ||
| }) | ||
| } | ||
| if (proxyCache.has(cacheKey)) { | ||
| return proxyCache.get(cacheKey) | ||
| } | ||
|
|
||
| async #proxyConnect (proxy, request, options) { | ||
| // socks-proxy-agent accepts a dns lookup function | ||
| options.lookup ??= this.#options.lookup | ||
|
|
||
| // all the proxy agents use this secureEndpoint option to determine | ||
| // if the proxy should connect over tls or not. we can set it based | ||
| // on if the HttpAgent or HttpsAgent is used. | ||
| options.secureEndpoint = SECURE | ||
| let ProxyAgent = this.#ProxyAgent | ||
| if (Array.isArray(ProxyAgent)) { | ||
| ProxyAgent = options.secureEndpoint ? ProxyAgent[1] : ProxyAgent[0] | ||
| } | ||
|
|
||
| const socket = await abortRace([ | ||
| (ac) => createTimeout(this.#timeouts.connection, ac).catch(() => { | ||
| throw new Errors.ConnectionTimeoutError(options) | ||
| }), | ||
| (ac) => proxy.connect(request, options).then((s) => { | ||
| this.#setKeepAlive(s) | ||
| const proxyAgent = new ProxyAgent(proxy, this.#options) | ||
| proxyCache.set(cacheKey, proxyAgent) | ||
|
|
||
| const connectEvent = SECURE ? 'secureConnect' : 'connect' | ||
| const connectingEvent = SECURE ? 'secureConnecting' : 'connecting' | ||
| return proxyAgent | ||
| } | ||
|
|
||
| if (!s[connectingEvent]) { | ||
| return s | ||
| // takes an array of promises and races them against the connection timeout | ||
| // which will throw the necessary error if it is hit. This will return the | ||
| // result of the promise race. | ||
| async #timeoutConnection ({ promises, options, timeout }, ac = new AbortController()) { | ||
| if (timeout) { | ||
| const connectionTimeout = timers.setTimeout(timeout, null, { signal: ac.signal }) | ||
| .then(() => { | ||
| throw new Errors.ConnectionTimeoutError(`${options.host}:${options.port}`) | ||
| }).catch((err) => { | ||
| if (err.name === 'AbortError') { | ||
| return | ||
| } | ||
| throw err | ||
| }) | ||
| promises.push(connectionTimeout) | ||
| } | ||
|
|
||
| return abortRace([ | ||
| () => once(s, 'error', ac).then((err) => { | ||
| throw err | ||
| }), | ||
| () => once(s, connectEvent, ac).then(() => s), | ||
| ], ac) | ||
| }), | ||
| ]) | ||
|
|
||
| this.#setIdleTimeout(socket, options) | ||
|
|
||
| return socket | ||
| let result | ||
| try { | ||
| result = await Promise.race(promises) | ||
| ac.abort() | ||
| } catch (err) { | ||
| ac.abort() | ||
| throw err | ||
| } | ||
| return result | ||
| } | ||
|
|
||
| async connect (request, options) { | ||
| const proxy = this.#getProxy(options) | ||
| if (proxy) { | ||
| return this.#proxyConnect(proxy, request, options) | ||
| async connect (request, options) { | ||
| // if the connection does not have its own lookup function | ||
| // set, then use the one from our options | ||
| options.lookup ??= this.#options.lookup | ||
|
|
||
| let socket | ||
| let timeout = this.#timeouts.connection | ||
|
|
||
| const proxy = this.#getProxy(options) | ||
| if (proxy) { | ||
| // some of the proxies will wait for the socket to fully connect before | ||
| // returning so we have to await this while also racing it against the | ||
| // connection timeout. | ||
| const start = Date.now() | ||
| socket = await this.#timeoutConnection({ | ||
| options, | ||
| timeout, | ||
| promises: [proxy.connect(request, options)], | ||
| }) | ||
| // see how much time proxy.connect took and subtract it from | ||
| // the timeout | ||
| if (timeout) { | ||
| timeout = timeout - (Date.now() - start) | ||
| } | ||
| } else { | ||
| socket = (options.secureEndpoint ? tls : net).connect(options) | ||
| } | ||
|
|
||
| const socket = SOCKET_TYPE.connect(options) | ||
| socket.setKeepAlive(this.keepAlive, this.keepAliveMsecs) | ||
| socket.setNoDelay(this.keepAlive) | ||
|
|
||
| this.#setKeepAlive(socket) | ||
| const abortController = new AbortController() | ||
| const { signal } = abortController | ||
|
|
||
| await abortRace([ | ||
| (s) => createTimeout(this.#timeouts.connection, s).catch(() => { | ||
| throw new Errors.ConnectionTimeoutError(options) | ||
| }), | ||
| (s) => once(socket, 'error', s).then((err) => { | ||
| throw err | ||
| }), | ||
| (s) => once(socket, 'connect', s), | ||
| ]) | ||
| const connectPromise = socket[options.secureEndpoint ? 'secureConnecting' : 'connecting'] | ||
| ? once(socket, options.secureEndpoint ? 'secureConnect' : 'connect', { signal }) | ||
| : Promise.resolve() | ||
|
|
||
| this.#setIdleTimeout(socket, options) | ||
| await this.#timeoutConnection({ | ||
| options, | ||
| timeout, | ||
| promises: [ | ||
| connectPromise, | ||
| once(socket, 'error', { signal }).then((err) => { | ||
| throw err[0] | ||
| }), | ||
| ], | ||
| }, abortController) | ||
|
|
||
| return socket | ||
| if (this.#timeouts.idle) { | ||
| socket.setTimeout(this.#timeouts.idle, () => { | ||
| socket.destroy(new Errors.IdleTimeoutError(`${options.host}:${options.port}`)) | ||
| }) | ||
| } | ||
|
|
||
| addRequest (request, options) { | ||
| const proxy = this.#getProxy(options) | ||
| // it would be better to call proxy.addRequest here but this causes the | ||
| // http-proxy-agent to call its super.addRequest which causes the request | ||
| // to be added to the agent twice. since we only support 3 agents | ||
| // currently (see the required agents in proxy.js) we have manually | ||
| // checked that the only public methods we need to call are called in the | ||
| // next block. this could change in the future and presumably we would get | ||
| // failing tests until we have properly called the necessary methods on | ||
| // each of our proxy agents | ||
| if (proxy?.setRequestProps) { | ||
| proxy.setRequestProps(request, options) | ||
| } | ||
|
|
||
| request.setHeader('connection', this.keepAlive ? 'keep-alive' : 'close') | ||
|
|
||
| const responseTimeout = createTimeout(this.#timeouts.response) | ||
| if (responseTimeout) { | ||
| request.once('finish', () => { | ||
| responseTimeout.start(() => { | ||
| request.destroy(new Errors.ResponseTimeoutError(request, this.proxy?.url)) | ||
| }) | ||
| }) | ||
| request.once('response', () => { | ||
| responseTimeout.clear() | ||
| }) | ||
| } | ||
|
|
||
| const transferTimeout = createTimeout(this.#timeouts.transfer) | ||
| if (transferTimeout) { | ||
| request.once('response', (res) => { | ||
| transferTimeout.start(() => { | ||
| res.destroy(new Errors.TransferTimeoutError(request, this.proxy?.url)) | ||
| }) | ||
| res.once('close', () => { | ||
| transferTimeout.clear() | ||
| }) | ||
| }) | ||
| } | ||
| return socket | ||
| } | ||
|
|
||
| return super.addRequest(request, options) | ||
| addRequest (request, options) { | ||
| const proxy = this.#getProxy(options) | ||
| // it would be better to call proxy.addRequest here but this causes the | ||
| // http-proxy-agent to call its super.addRequest which causes the request | ||
| // to be added to the agent twice. since we only support 3 agents | ||
| // currently (see the required agents in proxy.js) we have manually | ||
| // checked that the only public methods we need to call are called in the | ||
| // next block. this could change in the future and presumably we would get | ||
| // failing tests until we have properly called the necessary methods on | ||
| // each of our proxy agents | ||
| if (proxy?.setRequestProps) { | ||
| proxy.setRequestProps(request, options) | ||
| } | ||
|
|
||
| createSocket (req, options, cb) { | ||
| return Promise.resolve() | ||
| .then(() => this.connect(req, options)) | ||
| .then((socket) => { | ||
| this.#socket = socket | ||
| return super.createSocket(req, options, cb) | ||
| }, cb) | ||
| request.setHeader('connection', this.keepAlive ? 'keep-alive' : 'close') | ||
|
|
||
| if (this.#timeouts.response) { | ||
| let responseTimeout | ||
| request.once('finish', () => { | ||
| setTimeout(() => { | ||
| request.destroy(new Errors.ResponseTimeoutError(request, this.#proxy)) | ||
| }, this.#timeouts.response) | ||
| }) | ||
| request.once('response', () => { | ||
| clearTimeout(responseTimeout) | ||
| }) | ||
| } | ||
|
|
||
| createConnection () { | ||
| return this.#socket | ||
| if (this.#timeouts.transfer) { | ||
| let transferTimeout | ||
| request.once('response', (res) => { | ||
| setTimeout(() => { | ||
| res.destroy(new Errors.TransferTimeoutError(request, this.#proxy)) | ||
| }, this.#timeouts.transfer) | ||
| res.once('close', () => { | ||
| clearTimeout(transferTimeout) | ||
| }) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| Object.defineProperty(agent, 'name', { value: name }) | ||
| return agent | ||
| } | ||
|
|
||
| module.exports = { | ||
| HttpAgent: createAgent(http, 'HttpAgent'), | ||
| HttpsAgent: createAgent(https, 'HttpsAgent'), | ||
| return super.addRequest(request, options) | ||
| } | ||
| } |
Oops, something went wrong.