From bbb5658279e935d12c6abfa3d3c00783f54be39b Mon Sep 17 00:00:00 2001
From: James Elias Sigurdarson <jamiees2@gmail.com>
Date: Thu, 7 Oct 2021 10:22:05 +0000
Subject: [PATCH] Fix EventEmitter leak

Signed-off-by: James Elias Sigurdarson <jamiees2@gmail.com>
---
 src/socket.ts       | 17 +++++++++++++----
 test/test.socket.ts | 13 +++++++++++++
 2 files changed, 26 insertions(+), 4 deletions(-)

diff --git a/src/socket.ts b/src/socket.ts
index fcb4c84..c1bf743 100644
--- a/src/socket.ts
+++ b/src/socket.ts
@@ -453,9 +453,18 @@ export class FluentSocket extends EventEmitter {
     this.processMessages(protocol.decodeServerStream(this.passThroughStream));
 
     return new Promise<void>((resolve, reject) => {
-      // This may call both resolve and reject, but the ES standard says this is OK
-      this.once(FluentSocketEvent.CONNECTED, () => resolve());
-      this.once(FluentSocketEvent.ERROR, err => reject(err));
+      const onConnected = () => {
+        resolve();
+        // Avoid a memory leak and remove the other listener
+        this.removeListener(FluentSocketEvent.ERROR, onError);
+      };
+      const onError = (err: Error) => {
+        reject(err);
+        // Avoid a memory leak and remove the other listener
+        this.removeListener(FluentSocketEvent.CONNECTED, onConnected);
+      };
+      this.once(FluentSocketEvent.CONNECTED, onConnected);
+      this.once(FluentSocketEvent.ERROR, onError);
     });
   }
 
@@ -483,7 +492,7 @@ export class FluentSocket extends EventEmitter {
         this.onMessage(message);
       }
     } catch (e) {
-      this.close(CloseState.RECONNECT, e);
+      this.close(CloseState.RECONNECT, e as Error);
     }
   }
 
diff --git a/test/test.socket.ts b/test/test.socket.ts
index 00733d0..b098ea5 100644
--- a/test/test.socket.ts
+++ b/test/test.socket.ts
@@ -60,6 +60,19 @@ describe("FluentSocket", () => {
     sinon.assert.calledOnce(connectStub);
   });
 
+  it("should not preserve error handlers after connect", done => {
+    const {socket, connectStub} = createFluentSocket({disableReconnect: true});
+
+    socket.on(FluentSocketEvent.WRITABLE, () => {
+      expect(socket.listenerCount(FluentSocketEvent.ERROR)).to.equal(0);
+      done();
+    });
+
+    socket.connect();
+
+    sinon.assert.calledOnce(connectStub);
+  });
+
   it("should not block for draining on write by default", done => {
     const {socket, stream, connectStub} = createFluentSocket({
       disableReconnect: true,