Feat: Agent connect/disconnect events (#558)

ognjenjevremovic · web-flow · commit 138ad619c840 · 2021-02-19T19:38:08.000+01:00
* feat: 🎸 Add connect/disconnect events to Agent Extend EventEmitter class from within Agent and define support for the connect and disconnect events. ✅ Closes: #522 * test: 💍 Add tests for Agent connect/disconnect events Provide minimum set of tests for connect/disconnect events for the Agent class. ✅ Closes: #522 * test: 💍 Extend tests for Agent connect/disconnect events Disconnect event listener should be removed from the pool/client (acquired through the Agent instance) when instance of the Agent class is destroyed. ✅ Closes: #522 * test: 💍 Increase Agent test coverage Provide the tests for the Agent get method, checking weather the pool is instance of Client or Pool class. ✅ Closes: #522
diff --git a/lib/agent.js b/lib/agent.js
@@ -4,11 +4,32 @@ const Pool = require('./pool')
 const Client = require('./core/client')
 const util = require('./core/util')
 const { kAgentOpts, kAgentCache } = require('./core/symbols')
+const EventEmitter = require('events')
 
-class Agent {
+const kOnConnect = Symbol('onConnect')
+const kOnDisconnect = Symbol('onDisconnect')
+
+class Agent extends EventEmitter {
   constructor (opts) {
+    super()
+
     this[kAgentOpts] = opts
     this[kAgentCache] = new Map()
+
+    const agent = this
+
+    this[kOnConnect] = function onConnect (client) {
+      agent.emit('connect', client)
+    }
+
+    this[kOnDisconnect] = function onDestroy (client, err) {
+      if (this.connected === 0 && this.size === 0) {
+        this.off('disconnect', agent[kOnDisconnect])
+        agent[kAgentCache].delete(this.origin)
+      }
+
+      agent.emit('disconnect', client, err)
+    }
   }
 
   get (origin) {
@@ -19,18 +40,15 @@ class Agent {
     const self = this
     let pool = self[kAgentCache].get(origin)
 
-    function onDisconnect () {
-      if (this.connected === 0 && this.size === 0) {
-        this.off('disconnect', onDisconnect)
-        self[kAgentCache].delete(origin)
-      }
-    }
-
     if (!pool) {
       pool = self[kAgentOpts] && self[kAgentOpts].connections === 1
         ? new Client(origin, self[kAgentOpts])
         : new Pool(origin, self[kAgentOpts])
-      pool.on('disconnect', onDisconnect)
+
+      pool
+        .on('connect', this[kOnConnect])
+        .on('disconnect', this[kOnDisconnect])
+
       self[kAgentCache].set(origin, pool)
     }
 
diff --git a/test/agent.js b/test/agent.js
@@ -5,10 +5,11 @@ const http = require('http')
 const { Agent, request, stream, pipeline, setGlobalAgent } = require('../lib/agent')
 const { PassThrough } = require('stream')
 const { InvalidArgumentError, InvalidReturnValueError } = require('../lib/core/errors')
-const { errors } = require('..')
+const { Client, Pool, errors } = require('../index')
+const { promisify } = require('util')
 
 tap.test('Agent', t => {
-  t.plan(6)
+  t.plan(9)
 
   t.test('setGlobalAgent', t => {
     t.plan(2)
@@ -122,6 +123,118 @@ tap.test('Agent', t => {
     })
   })
 
+  t.test('Agent connect/disconnect event(s)', t => {
+    t.plan(2)
+
+    t.test('multiple connections', t => {
+      const connections = 3
+      t.plan(6 * connections)
+
+      const server = http.createServer((req, res) => {
+        res.writeHead(200, {
+          Connection: 'keep-alive',
+          'Keep-Alive': 'timeout=1s'
+        })
+        res.end('ok')
+      })
+      t.tearDown(server.close.bind(server))
+
+      server.listen(0, async () => {
+        const origin = `http://localhost:${server.address().port}`
+        const agent = new Agent({ connections })
+
+        t.tearDown(agent.close.bind(agent))
+
+        agent.on('connect', (client) => {
+          t.ok(client)
+        })
+        agent.on('disconnect', (client, error) => {
+          t.ok(client)
+          t.true(error instanceof errors.InformationalError)
+          t.strictEqual(error.code, 'UND_ERR_INFO')
+          t.strictEqual(error.message, 'reset')
+        })
+
+        for (let i = 0; i < connections; i++) {
+          await request(origin, { agent })
+            .then(() => {
+              t.pass('should pass')
+            })
+            .catch(err => {
+              t.fail(err)
+            })
+        }
+      })
+    })
+
+    t.test('remove disconnect listeners when destroyed', t => {
+      t.plan(3)
+
+      const server = http.createServer((req, res) => {
+        res.writeHead(200, {
+          Connection: 'keep-alive',
+          'Keep-Alive': 'timeout=1s'
+        })
+        res.end('ok')
+      })
+      t.tearDown(server.close.bind(server))
+
+      server.listen(0, async () => {
+        const origin = `http://localhost:${server.address().port}`
+        const agent = new Agent()
+
+        t.tearDown(agent.close.bind(agent))
+
+        const pool = agent.get(origin)
+        t.true(pool.listeners('disconnect').length === 1)
+
+        agent.on('disconnect', () => {
+          t.true(pool.listeners('disconnect').length === 0)
+        })
+
+        await request(origin, { agent })
+          .then(() => {
+            t.pass('should pass')
+          })
+          .catch(err => {
+            t.fail(err)
+          })
+      })
+    })
+  })
+
+  t.test('check if pool', async t => {
+    t.plan(1)
+
+    const server = http.createServer()
+    t.tearDown(server.close.bind(server))
+    await promisify(server.listen.bind(server))(0)
+
+    const origin = `http://localhost:${server.address().port}`
+    const agent = new Agent()
+
+    t.tearDown(agent.close.bind(agent))
+
+    const pool = agent.get(origin)
+    t.true(pool instanceof Pool)
+  })
+
+  t.test('check if client', async t => {
+    t.plan(1)
+
+    const server = http.createServer()
+    t.tearDown(server.close.bind(server))
+    await promisify(server.listen.bind(server))(0)
+
+    const origin = `http://localhost:${server.address().port}`
+    const agent = new Agent({ connections: 1 })
+
+    t.tearDown(agent.close.bind(agent))
+
+    const pool = agent.get(origin)
+    t.true(pool instanceof Client)
+  })
+
   t.test('request a resource', t => {
     t.plan(5)
 
