fix this issue #90. Closes underlying socket and dangling timeout (#91)

src/producer/services/close.js

@@ -8,11 +8,12 @@ const close = async ({ producerId, client, connected, requestId, responseMediato
       message: 'Cannot close not connected producer',
     });
 
-  const { sendSimpleCommandRequest } = client.getCnx();
+  const { sendSimpleCommandRequest, close } = client.getCnx();
   const closeProducer = commands.closeProducer({ producerId, requestId });
   const { command } = await sendSimpleCommandRequest({ command: closeProducer }, responseMediator);
   if (!utils.isNil(command.error))
     throw new errors.PulsarFlexProducerCloseError({ message: command.message });
+  close(); //close the underlying socket to ensure there are no dangling resources
 };
 
 module.exports = close;

src/responseMediators/abstract/responseMediator.js

@@ -6,6 +6,7 @@ class ResponseMediator {
     this._commands = [];
     this._responseEvents = client.getResponseEvents();
     this._timeout = timeout;
+    this._pendingTimeout = null;
   }
 
   _startToMediate() {
@@ -28,13 +29,14 @@ class ResponseMediator {
 
   purgeRequests({ error }) {
     Object.values(this._requests).forEach(({ reject }) => reject(new error({})));
+    clearTimeout(this._pendingTimeout); //ensure there are no dangling resources
   }
 
   response({ data, autoResolve }) {
     const id = this._idFunc(this._parseCommand(data));
     return new Promise((resolve, reject) => {
       autoResolve && resolve();
-      setTimeout(() => reject(new errors.PulsarFlexResponseTimeoutError({ id })), this._timeout);
+      this._pendingTimeout = setTimeout(() => reject(new errors.PulsarFlexResponseTimeoutError({ id })), this._timeout);
       this._requests[id] = { resolve, reject };
     });
   }