262# hide manuallyClose param in close methods

Author: apietroni51

src/client.ts

@@ -156,7 +156,7 @@ export class Client {
     }
     const publisher = new StreamPublisher(this.pool, streamPublisherParams, lastPublishingId, filter)
     connection.registerForClosePublisher(publisher.extendedId, params.stream, async () => {
-      await publisher.close(false)
+      await publisher.automaticClose()
       this.publishers.delete(publisher.extendedId)
     })
     this.publishers.set(publisher.extendedId, { publisher, connection, params, filter })
@@ -176,7 +176,7 @@ export class Client {
     if (!res.ok) {
       throw new Error(`Delete Publisher command returned error with code ${res.code} - ${errorMessageOf(res.code)}`)
     }
-    await publisher?.close(true)
+    await publisher?.close()
     this.publishers.delete(extendedPublisherId)
     this.logger.info(`deleted publisher with publishing id ${publisherId}`)
     return res.ok
@@ -280,7 +280,7 @@ export class Client {
   }
 
   private async closeAllPublishers() {
-    await Promise.all([...this.publishers.values()].map((c) => c.publisher.close(true)))
+    await Promise.all([...this.publishers.values()].map((c) => c.publisher.close()))
     this.publishers = new Map<string, PublisherMappedValue>()
   }
 

src/consumer.ts

@@ -14,11 +14,8 @@ export const computeExtendedConsumerId = (consumerId: number, connectionId: stri
 export interface Consumer {
   /**
    * Close the publisher
-   *
-   * @param {boolean} manuallyClose - Weather you want to close the publisher manually or not
    */
-  // TODO - clarify the parameter
-  close(manuallyClose: boolean): Promise<void>
+  close(): Promise<void>
 
   /**
    * Store the stream offset on the server
@@ -99,7 +96,12 @@ export class StreamConsumer implements Consumer {
 
   async close(): Promise<void> {
     this.closed = true
-    await this.pool.releaseConnection(this.connection)
+    await this.pool.releaseConnection(this.connection, true)
+  }
+
+  async automaticClose(): Promise<void> {
+    this.closed = true
+    await this.pool.releaseConnection(this.connection, false)
   }
 
   public storeOffset(offsetValue: bigint): Promise<void> {

src/publisher.ts

@@ -144,11 +144,8 @@ export interface Publisher {
 
   /**
    * Close the publisher
-   *
-   * @param {boolean} manuallyClose - Weather you want to close the publisher manually or not
    */
-  // TODO - clarify the parameter
-  close(manuallyClose: boolean): Promise<void>
+  close(): Promise<void>
 
   closed: boolean
   ref: string
@@ -278,10 +275,18 @@ export class StreamPublisher implements Publisher {
     return this.publisherRef
   }
 
-  public async close(manuallyClose: boolean): Promise<void> {
+  public async close(): Promise<void> {
+    if (!this.closed) {
+      await this.flush()
+      await this.pool.releaseConnection(this.connection, true)
+    }
+    this._closed = true
+  }
+
+  public async automaticClose(): Promise<void> {
     if (!this.closed) {
       await this.flush()
-      await this.pool.releaseConnection(this.connection, manuallyClose)
+      await this.pool.releaseConnection(this.connection, false)
     }
     this._closed = true
   }

src/super_stream_consumer.ts

@@ -68,6 +68,6 @@ export class SuperStreamConsumer {
   }
 
   async close(): Promise<void> {
-    await Promise.all([...this.consumers.values()].map((c) => c.close(true)))
+    await Promise.all([...this.consumers.values()].map((c) => c.close()))
   }
 }

src/super_stream_publisher.ts

@@ -43,7 +43,7 @@ export class SuperStreamPublisher {
   }
 
   public async close(): Promise<void> {
-    await Promise.all([...this.publishers.values()].map((p) => p.close(true)))
+    await Promise.all([...this.publishers.values()].map((p) => p.close()))
     this.publishers = new Map()
   }
 

test/e2e/sub_entry_consume.test.ts

@@ -29,7 +29,7 @@ describe("consume a batch of messages", () => {
       await rabbit.deleteStream(streamName)
       await rabbit.closeAllConnections()
       await rabbit.deleteAllQueues({ match: /my-stream-/ })
-      await consumer?.close(true)
+      await consumer?.close()
     } catch (e) {}
   })