feat(dispatcher/proxy-agent): new diagnostics event 'undici:proxy:connected' (#4659)

Co-authored-by: o.drapeza <[email protected]>

Author: SuperOleg39

docs/docs/api/DiagnosticsChannel.md

@@ -254,3 +254,17 @@ diagnosticsChannel.channel('undici:websocket:pong').subscribe(({ payload, websoc
   console.log(websocket) // the WebSocket instance
 })
 ```
+
+## `undici:proxy:connected`
+
+This message is published after the `ProxyAgent` establishes a connection to the proxy server.
+
+```js
+import diagnosticsChannel from 'diagnostics_channel'
+
+diagnosticsChannel.channel('undici:proxy:connected').subscribe(({ socket, connectParams }) => {
+  console.log(socket)
+  console.log(connectParams)
+  // const { origin, port, path, signal, headers, servername } = connectParams
+})
+```

lib/core/diagnostics.js

@@ -26,7 +26,9 @@ const channels = {
   close: diagnosticsChannel.channel('undici:websocket:close'),
   socketError: diagnosticsChannel.channel('undici:websocket:socket_error'),
   ping: diagnosticsChannel.channel('undici:websocket:ping'),
-  pong: diagnosticsChannel.channel('undici:websocket:pong')
+  pong: diagnosticsChannel.channel('undici:websocket:pong'),
+  // ProxyAgent
+  proxyConnected: diagnosticsChannel.channel('undici:proxy:connected')
 }
 
 let isTrackingClientEvents = false

lib/dispatcher/proxy-agent.js

@@ -7,6 +7,7 @@ const DispatcherBase = require('./dispatcher-base')
 const { InvalidArgumentError, RequestAbortedError, SecureProxyConnectionError } = require('../core/errors')
 const buildConnector = require('../core/connect')
 const Client = require('./client')
+const { channels } = require('../core/diagnostics')
 
 const kAgent = Symbol('proxy agent')
 const kClient = Symbol('proxy client')
@@ -150,7 +151,7 @@ class ProxyAgent extends DispatcherBase {
           requestedPath += `:${defaultProtocolPort(opts.protocol)}`
         }
         try {
-          const { socket, statusCode } = await this[kClient].connect({
+          const connectParams = {
             origin,
             port,
             path: requestedPath,
@@ -161,12 +162,21 @@ class ProxyAgent extends DispatcherBase {
               ...(opts.connections == null || opts.connections > 0 ? { 'proxy-connection': 'keep-alive' } : {})
             },
             servername: this[kProxyTls]?.servername || proxyHostname
-          })
+          }
+          const { socket, statusCode } = await this[kClient].connect(connectParams)
           if (statusCode !== 200) {
             socket.on('error', noop).destroy()
             callback(new RequestAbortedError(`Proxy response (${statusCode}) !== 200 when HTTP Tunneling`))
             return
           }
+
+          if (channels.proxyConnected.hasSubscribers) {
+            channels.proxyConnected.publish({
+              socket,
+              connectParams
+            })
+          }
+
           if (opts.protocol !== 'https:') {
             callback(null, socket)
             return

test/proxy-agent.js

@@ -2,12 +2,14 @@
 
 const { tspl } = require('@matteo.collina/tspl')
 const { test, after } = require('node:test')
+const diagnosticsChannel = require('node:diagnostics_channel')
 const { request, fetch, setGlobalDispatcher, getGlobalDispatcher } = require('..')
 const { InvalidArgumentError, SecureProxyConnectionError } = require('../lib/core/errors')
 const ProxyAgent = require('../lib/dispatcher/proxy-agent')
 const Pool = require('../lib/dispatcher/pool')
 const { createServer } = require('node:http')
 const https = require('node:https')
+const { Socket } = require('node:net')
 const { createProxy } = require('proxy')
 
 const certs = (() => {
@@ -120,11 +122,17 @@ test('should accept string, URL and object as options', (t) => {
 })
 
 test('use proxy-agent to connect through proxy (keep alive)', async (t) => {
-  t = tspl(t, { plan: 6 })
+  t = tspl(t, { plan: 10 })
   const server = await buildServer()
   const proxy = await buildProxy()
   delete proxy.authenticate
 
+  let _socket, _connectParams
+  diagnosticsChannel.channel('undici:proxy:connected').subscribe(({ socket, connectParams }) => {
+    _socket = socket
+    _connectParams = connectParams
+  })
+
   const serverUrl = `http://localhost:${server.address().port}`
   const proxyUrl = `http://localhost:${proxy.address().port}`
   const proxyAgent = new ProxyAgent({
@@ -154,6 +162,10 @@ test('use proxy-agent to connect through proxy (keep alive)', async (t) => {
   t.strictEqual(statusCode, 200)
   t.deepStrictEqual(json, { hello: 'world' })
   t.strictEqual(headers.connection, 'keep-alive', 'should remain the connection open')
+  t.ok(_socket instanceof Socket)
+  t.equal(_connectParams.origin, proxyUrl)
+  t.equal(_connectParams.path, serverUrl.replace('http://', ''))
+  t.equal(_connectParams.headers['proxy-connection'], 'keep-alive')
 
   server.close()
   proxy.close()