Remove listeners on close() (#34)

lib/connection.ts

@@ -244,7 +244,8 @@ export class Connection extends Entity {
   }
 
   /**
-   * Removes the provided session from the internal map.
+   * Removes the provided session from the internal map in rhea.
+   * Also removes all the event handlers added in the rhea-promise library on the provided session.
    * @param {Session} session The session to be removed.
    */
   removeSession(session: Session): void {
@@ -377,7 +378,9 @@ export class Connection extends Entity {
   }
 
   /**
-   * Remove all the sessions from the internal map.
+   * Clears all the amqp sessions from the internal map maintained in rhea. This does not remove any
+   * of the event handlers added in the rhea-promise library. To clear such event handlers, either
+   * call remove() or close() on each session
    */
   removeAllSessions(): void {
     if (this._connection) {

lib/link.ts

@@ -191,7 +191,8 @@ export abstract class Link extends Entity {
   }
 
   /**
-   * Removes the sender/receiver and it's underlying session from the internal map.
+   * Removes the underlying amqp link and it's session from the internal map in rhea. Also removes
+   * all the event handlers added in the rhea-promise library on the link and it's session.
    * @returns {void} void
    */
   remove(): void {
@@ -207,13 +208,15 @@ export abstract class Link extends Entity {
   }
 
   /**
-   * Closes the amqp link.
+   * Closes the underlying amqp link and session in rhea if open. Also removes all the event
+   * handlers added in the rhea-promise library on the link and it's session
    * @return {Promise<void>} Promise<void>
    * - **Resolves** the promise when rhea emits the "sender_close" | "receiver_close" event.
    * - **Rejects** the promise with an AmqpError when rhea emits the
    * "sender_error" | "receiver_error" event while trying to close the amqp link.
    */
   async close(): Promise<void> {
+    this.removeAllListeners();
     await new Promise<void>((resolve, reject) => {
       log.error("[%s] The %s is open ? -> %s", this.connection.id, this.type, this.isOpen());
       if (this.isOpen()) {

lib/session.ts

@@ -84,6 +84,10 @@ export class Session extends Entity {
     return this._session.is_itself_closed();
   }
 
+  /**
+   * Removes the underlying amqp session from the internal map in rhea.
+   * Also removes all the event handlers added in the rhea-promise library on the session.
+   */
   remove(): void {
     if (this._session) {
       // Remove our listeners and listeners from rhea's 'session' object.
@@ -100,13 +104,15 @@ export class Session extends Entity {
   }
 
   /**
-   * Closes the amqp session.
+   * Closes the underlying amqp session in rhea if open. Also removes all the event
+   * handlers added in the rhea-promise library on the session
    * @return {Promise<void>} Promise<void>
    * - **Resolves** the promise when rhea emits the "session_close" event.
    * - **Rejects** the promise with an AmqpError when rhea emits the "session_error" event while trying
    * to close an amqp session.
    */
   close(): Promise<void> {
+    this.removeAllListeners();
     return new Promise<void>((resolve, reject) => {
       log.error("[%s] The session is open ? -> %s", this.connection.id, this.isOpen());
       if (this.isOpen()) {