262# hide manuallyClose param in close methods

Author: apietroni51

src/client.ts

@@ -4,10 +4,17 @@ import { inspect } from "util"
 import { Compression, CompressionType, GzipCompression, NoneCompression } from "./compression"
 import { Connection, ConnectionInfo, ConnectionParams, errorMessageOf } from "./connection"
 import { ConnectionPool, ConnectionPurpose } from "./connection_pool"
-import { Consumer, ConsumerFunc, ConsumerUpdateListener, StreamConsumer, computeExtendedConsumerId } from "./consumer"
+import {
+  Consumer,
+  ConsumerFunc,
+  ConsumerUpdateListener,
+  StreamConsumer,
+  PrivateConsumer,
+  computeExtendedConsumerId,
+} from "./consumer"
 import { STREAM_ALREADY_EXISTS_ERROR_CODE } from "./error_codes"
 import { Logger, NullLogger } from "./logger"
-import { FilterFunc, Message, Publisher, StreamPublisher } from "./publisher"
+import { FilterFunc, Message, Publisher, PrivatePublisher, StreamPublisher } from "./publisher"
 import { ConsumerUpdateResponse } from "./requests/consumer_update_response"
 import { CreateStreamArguments, CreateStreamRequest } from "./requests/create_stream_request"
 import { CreateSuperStreamRequest } from "./requests/create_super_stream_request"
@@ -48,10 +55,10 @@ export type ConnectionClosedListener = (hadError: boolean) => void
 
 export type ClosingParams = { closingCode: number; closingReason: string; manuallyClose?: boolean }
 
-type ConsumerMappedValue = { connection: Connection; consumer: StreamConsumer; params: DeclareConsumerParams }
+type ConsumerMappedValue = { connection: Connection; consumer: PrivateConsumer; params: DeclareConsumerParams }
 type PublisherMappedValue = {
   connection: Connection
-  publisher: StreamPublisher
+  publisher: PrivatePublisher
   params: DeclarePublisherParams
   filter: FilterFunc | undefined
 }
@@ -154,7 +161,7 @@ export class Client {
         publisherRef: streamPublisherParams.publisherRef,
       })
     }
-    const publisher = new StreamPublisher(this.pool, streamPublisherParams, lastPublishingId, filter)
+    const publisher = new PrivatePublisher(this.pool, streamPublisherParams, lastPublishingId, filter)
     connection.registerForClosePublisher(publisher.extendedId, params.stream, async () => {
       await publisher.close(false)
       this.publishers.delete(publisher.extendedId)
@@ -163,7 +170,7 @@ export class Client {
     this.logger.info(
       `New publisher created with stream name ${params.stream}, publisher id ${publisherId} and publisher reference ${params.publisherRef}`
     )
-    return publisher
+    return new StreamPublisher(publisher)
   }
 
   public async deletePublisher(extendedPublisherId: string) {
@@ -194,7 +201,7 @@ export class Client {
       throw new Error(`Broker does not support message filtering.`)
     }
 
-    const consumer = new StreamConsumer(
+    const consumer = new PrivateConsumer(
       this.pool,
       handle,
       {
@@ -211,20 +218,24 @@ export class Client {
       params.filter
     )
     connection.registerForCloseConsumer(consumer.extendedId, params.stream, async () => {
-      if (params.connectionClosedListener) {
-        params.connectionClosedListener(false)
-      }
-      await this.closeConsumer(consumer.extendedId)
+      const activeConsumer = await this.prepareCloseConsumer(consumer.extendedId)
+      await this.closing(activeConsumer.consumer, consumer.extendedId, false)
     })
     this.consumers.set(consumer.extendedId, { connection, consumer, params })
     await this.declareConsumerOnConnection(params, consumerId, connection, superStreamConsumer?.superStream)
     this.logger.info(
       `New consumer created with stream name ${params.stream}, consumer id ${consumerId} and offset ${params.offset.type}`
     )
-    return consumer
+    return new StreamConsumer(consumer)
   }
 
   public async closeConsumer(extendedConsumerId: string) {
+    const activeConsumer = await this.prepareCloseConsumer(extendedConsumerId)
+    await this.closing(activeConsumer.consumer, extendedConsumerId, true)
+    return true
+  }
+
+  private async prepareCloseConsumer(extendedConsumerId: string) {
     const activeConsumer = this.consumers.get(extendedConsumerId)
     if (!activeConsumer) {
       this.logger.error("Consumer does not exist")
@@ -238,8 +249,7 @@ export class Client {
     if (streamInfos.length > 0 && streamExists(streamInfos[0])) {
       await this.unsubscribe(activeConsumer.connection, consumerId)
     }
-    await this.closing(activeConsumer.consumer, extendedConsumerId)
-    return true
+    return activeConsumer
   }
 
   public async declareSuperStreamConsumer(
@@ -275,7 +285,7 @@ export class Client {
   }
 
   private async closeAllConsumers() {
-    await Promise.all([...this.consumers.values()].map(({ consumer }) => consumer.close()))
+    await Promise.all([...this.consumers.values()].map(({ consumer }) => consumer.close(true)))
     this.consumers = new Map<string, ConsumerMappedValue>()
   }
 
@@ -593,7 +603,7 @@ export class Client {
     }
   }
 
-  private async getConsumerOrServerSavedOffset(consumer: StreamConsumer) {
+  private async getConsumerOrServerSavedOffset(consumer: PrivateConsumer) {
     if (consumer.isSingleActive && consumer.consumerRef && consumer.consumerUpdateListener) {
       try {
         const offset = await consumer.consumerUpdateListener(consumer.consumerRef, consumer.streamName)
@@ -721,8 +731,8 @@ export class Client {
     return res
   }
 
-  private async closing(consumer: StreamConsumer, extendedConsumerId: string) {
-    await consumer.close()
+  private async closing(consumer: PrivateConsumer, extendedConsumerId: string, manuallyClose: boolean) {
+    await consumer.close(manuallyClose)
     this.consumers.delete(extendedConsumerId)
     this.logger.info(`Closed consumer with id: ${extendedConsumerId}`)
   }

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
@@ -54,6 +51,34 @@ export interface Consumer {
 }
 
 export class StreamConsumer implements Consumer {
+  constructor(private readonly c: PrivateConsumer) {}
+  close(): Promise<void> {
+    return this.c.close(true)
+  }
+  storeOffset(offsetValue: bigint): Promise<void> {
+    return this.c.storeOffset(offsetValue)
+  }
+  queryOffset(): Promise<bigint> {
+    return this.c.queryOffset()
+  }
+  getConnectionInfo(): ConnectionInfo {
+    return this.c.getConnectionInfo()
+  }
+  updateConsumerOffset(offset: Offset): void {
+    this.c.updateConsumerOffset(offset)
+  }
+  public get consumerId(): number {
+    return this.c.consumerId
+  }
+  public get consumerRef(): string | undefined {
+    return this.c.consumerRef
+  }
+  public get extendedId(): string {
+    return this.c.extendedId
+  }
+}
+
+export class PrivateConsumer {
   private connection: Connection
   private stream: string
   public consumerId: number
@@ -97,9 +122,9 @@ export class StreamConsumer implements Consumer {
     this.singleActive = params.singleActive ?? false
   }
 
-  async close(): Promise<void> {
+  async close(manuallyClose: boolean): Promise<void> {
     this.closed = true
-    await this.pool.releaseConnection(this.connection)
+    await this.pool.releaseConnection(this.connection, manuallyClose)
   }
 
   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
@@ -159,7 +156,60 @@ export interface Publisher {
 export type FilterFunc = (msg: Message) => string | undefined
 type PublishConfirmCallback = (err: number | null, publishingIds: bigint[]) => void
 export type SendResult = { sent: boolean; publishingId: bigint }
+
 export class StreamPublisher implements Publisher {
+  constructor(private readonly p: PrivatePublisher) {}
+
+  send(message: Buffer, opts?: MessageOptions): Promise<SendResult> {
+    return this.p.send(message, opts)
+  }
+  basicSend(publishingId: bigint, content: Buffer, opts?: MessageOptions): Promise<SendResult> {
+    return this.p.basicSend(publishingId, content, opts)
+  }
+  flush(): Promise<boolean> {
+    return this.p.flush()
+  }
+  sendSubEntries(messages: Message[], compressionType?: CompressionType): Promise<void> {
+    return this.p.sendSubEntries(messages, compressionType)
+  }
+  on(event: "metadata_update", listener: MetadataUpdateListener): void
+  on(event: "publish_confirm", listener: PublishConfirmCallback): void
+  on(event: "metadata_update" | "publish_confirm", listener: MetadataUpdateListener | PublishConfirmCallback): void {
+    switch (event) {
+      case "metadata_update":
+        this.p.on(event, listener as MetadataUpdateListener)
+        break
+      case "publish_confirm":
+        this.p.on(event, listener as PublishConfirmCallback)
+        break
+      default:
+        break
+    }
+  }
+  getLastPublishingId(): Promise<bigint> {
+    return this.p.getLastPublishingId()
+  }
+  getConnectionInfo(): ConnectionInfo {
+    return this.p.getConnectionInfo()
+  }
+  close(): Promise<void> {
+    return this.p.close(true)
+  }
+  public get closed(): boolean {
+    return this.p.closed
+  }
+  public get ref(): string {
+    return this.p.ref
+  }
+  public get publisherId(): number {
+    return this.p.publisherId
+  }
+  public get extendedId(): string {
+    return this.p.extendedId
+  }
+}
+
+export class PrivatePublisher {
   private connection: Connection
   private stream: string
   readonly publisherId: number

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/declare_consumer.test.ts

@@ -85,7 +85,7 @@ describe("declare consumer", () => {
     await eventually(() => expect(messages).eql([Buffer.from("hello")]))
   }).timeout(10000)
 
-  it.skip("closing a consumer on an existing stream - raises connectionClosedListener", async () => {
+  it("closing a consumer on an existing stream - does not raise connectionClosedListener", async () => {
     const messages: Buffer[] = []
     await publisher.send(Buffer.from("hello"))
     let called = false
@@ -97,9 +97,9 @@ describe("declare consumer", () => {
     )
     await eventually(() => expect(messages).eql([Buffer.from("hello")]))
 
-    await client.close({ closingCode: 0, closingReason: "", manuallyClose: false })
+    await client.close({ closingCode: 0, closingReason: "" })
 
-    await eventually(() => expect(called).true)
+    await eventually(() => expect(called).false)
   }).timeout(10000)
 
   it("declaring a consumer on an existing stream with identifiers", async () => {

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) {}
   })