fix cannot close client after consumer (#272)

* fix cannot close client after consumer

* lint

* added log

Author: apietroni51

src/client.ts

@@ -68,7 +68,7 @@ export class Client {
   private publishers = new Map<string, PublisherMappedValue>()
   private compressions = new Map<CompressionType, Compression>()
   private locatorConnection: Connection
-  private pool: ConnectionPool = new ConnectionPool()
+  private pool: ConnectionPool
 
   private constructor(
     private readonly logger: Logger,
@@ -77,6 +77,7 @@ export class Client {
     this.compressions.set(CompressionType.None, NoneCompression.create())
     this.compressions.set(CompressionType.Gzip, GzipCompression.create())
     this.locatorConnection = this.getLocatorConnection()
+    this.pool = new ConnectionPool(logger)
   }
 
   getCompression(compressionType: CompressionType) {

src/connection_pool.ts

@@ -1,4 +1,6 @@
+import { inspect } from "util"
 import { Connection } from "./connection"
+import { Logger } from "./logger"
 import { getMaxSharedConnectionInstances } from "./util"
 
 type InstanceKey = string
@@ -7,6 +9,8 @@ export type ConnectionPurpose = "consumer" | "publisher"
 export class ConnectionPool {
   private connectionsMap: Map<InstanceKey, Connection[]> = new Map<InstanceKey, Connection[]>()
 
+  constructor(private readonly log: Logger) {}
+
   public async getConnection(
     entityType: ConnectionPurpose,
     streamName: string,
@@ -32,8 +36,14 @@ export class ConnectionPool {
 
   public async releaseConnection(connection: Connection, manuallyClose = true): Promise<void> {
     connection.decrRefCount()
-    if (connection.refCount <= 0) {
-      await connection.close({ closingCode: 0, closingReason: "", manuallyClose })
+    if (connection.refCount <= 0 && connection.ready) {
+      try {
+        await connection.close({ closingCode: 0, closingReason: "", manuallyClose })
+      } catch (e) {
+        // in case the client is closed immediately after a consumer, its connection has still not
+        // reset the ready flag, so we get an "Error: write after end"
+        this.log.warn(`Could not close connection: ${inspect(e)}`)
+      }
       this.removeCachedConnection(connection)
     }
   }

test/e2e/client_close.test.ts

@@ -0,0 +1,40 @@
+import { Client, Offset } from "../../src"
+import { createClient, createStreamName } from "../support/fake_data"
+import { Rabbit } from "../support/rabbit"
+import { username, password } from "../support/util"
+
+describe("close client", () => {
+  const rabbit = new Rabbit(username, password)
+  let streamName: string
+  let client: Client
+
+  beforeEach(async () => {
+    client = await createClient(username, password)
+    streamName = createStreamName()
+    await rabbit.createStream(streamName)
+  })
+
+  afterEach(async () => {
+    try {
+      await rabbit.deleteStream(streamName)
+      await rabbit.closeAllConnections()
+      await rabbit.deleteAllQueues({ match: /my-stream-/ })
+    } catch (e) {}
+  })
+
+  it("can close client after closing publisher", async () => {
+    const publisher = await client.declarePublisher({ stream: streamName })
+
+    await publisher.close(true)
+    await client.close()
+  })
+
+  it("can close client after closing consumer", async () => {
+    const consumer = await client.declareConsumer({ stream: streamName, offset: Offset.first() }, (_msg) => {
+      /* nothing */
+    })
+
+    await consumer.close(true)
+    await client.close()
+  })
+})