feat: use agent-base

This allows a single agent to serve as both the http and https agents.

This PR also refactors some things based on feedback in #57.

Author: lukekarrys

lib/agents.js

@@ -1,199 +1,203 @@
 '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 { urlify, appendPort } = require('./util')
 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
 
-  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 = {
+        proxy: urlify(proxy),
+        noProxy,
+        Agent: getProxyAgent(proxy),
+      }
     }
+  }
 
-    get proxy () {
-      return this.#proxy ? { url: this.#proxy.proxy } : {}
-    }
+  get proxy () {
+    return this.#proxy ? { url: this.#proxy.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(appendPort(`${options.protocol}//${options.host}`, options.port), {
+      proxy: this.#proxy.proxy,
+      noProxy: this.#proxy.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,
+    })
+
+    if (proxyCache.has(cacheKey)) {
+      return proxyCache.get(cacheKey)
     }
 
-    #setIdleTimeout (socket, options) {
-      if (this.#timeouts.idle) {
-        socket.setTimeout(this.#timeouts.idle, () => {
-          socket.destroy(new Errors.IdleTimeoutError(options))
-        })
-      }
+    let { Agent: ProxyAgent } = this.#proxy
+    if (Array.isArray(ProxyAgent)) {
+      ProxyAgent = options.secureEndpoint ? ProxyAgent[1] : ProxyAgent[0]
     }
 
-    async #proxyConnect (proxy, request, options) {
-      // socks-proxy-agent accepts a dns lookup function
-      options.lookup ??= this.#options.lookup
+    const proxyAgent = new ProxyAgent(proxy, this.#options)
+    proxyCache.set(cacheKey, proxyAgent)
 
-      // 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
+    return proxyAgent
+  }
 
-      const socket = await abortRace([
-        (ac) => createTimeout(this.#timeouts.connection, ac).catch(() => {
+  // 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)
-        }),
-        (ac) => proxy.connect(request, options).then((s) => {
-          this.#setKeepAlive(s)
-
-          const connectEvent = SECURE ? 'secureConnect' : 'connect'
-          const connectingEvent = SECURE ? 'secureConnecting' : 'connecting'
-
-          if (!s[connectingEvent]) {
-            return s
+        }).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))
+      })
     }
 
-    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?.url))
+        }, 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?.url))
+        }, 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)
+  }
 }