259 As a developer, I want the connection pool as an instance (#263)

* 259 As a developer, I want the connection pool as an instance

* fixed name

Author: apietroni51

example/package-lock.json

@@ -68,6 +68,7 @@ export class Client {
   private publishers = new Map<string, PublisherMappedValue>()
   private compressions = new Map<CompressionType, Compression>()
   private connection: Connection
+  private pool: ConnectionPool = new ConnectionPool()
 
   private constructor(
     private readonly logger: Logger,
@@ -115,7 +116,7 @@ export class Client {
 
   private async closeConnectionIfUnused(connection: Connection, params: ClosingParams) {
     if (connection.refCount <= 0) {
-      ConnectionPool.removeCachedConnection(this.connection)
+      this.pool.removeCachedConnection(this.connection)
       await this.connection.close({ ...params, manuallyClose: true })
     }
   }
@@ -162,7 +163,7 @@ export class Client {
         publisherRef: streamPublisherParams.publisherRef,
       })
     }
-    const publisher = new StreamPublisher(streamPublisherParams, lastPublishingId, filter)
+    const publisher = new StreamPublisher(this.pool, streamPublisherParams, lastPublishingId, filter)
     connection.registerForClosePublisher(publisher.extendedId, params.stream, async () => {
       await publisher.close(false)
       this.publishers.delete(publisher.extendedId)
@@ -203,6 +204,7 @@ export class Client {
     }
 
     const consumer = new StreamConsumer(
+      this.pool,
       handle,
       {
         connection,
@@ -631,12 +633,7 @@ export class Client {
     if (!chosenNode) {
       throw new Error(`Stream was not found on any node`)
     }
-    const cachedConnection = ConnectionPool.getUsableCachedConnection(
-      purpose,
-      streamName,
-      this.connection.vhost,
-      chosenNode.host
-    )
+    const cachedConnection = this.pool.getCachedConnection(purpose, streamName, this.connection.vhost, chosenNode.host)
     if (cachedConnection) return cachedConnection
 
     const newConnection = await this.getConnectionOnChosenNode(
@@ -647,7 +644,7 @@ export class Client {
       connectionClosedListener
     )
 
-    ConnectionPool.cacheConnection(purpose, streamName, this.connection.vhost, newConnection.hostname, newConnection)
+    this.pool.cacheConnection(purpose, streamName, this.connection.vhost, newConnection.hostname, newConnection)
     return newConnection
   }
 

src/client.ts

@@ -5,55 +5,53 @@ type InstanceKey = string
 export type ConnectionPurpose = "consumer" | "publisher"
 
 export class ConnectionPool {
-  private static consumerConnectionProxies = new Map<InstanceKey, Connection[]>()
-  private static publisherConnectionProxies = new Map<InstanceKey, Connection[]>()
+  private consumerConnectionProxies: Map<InstanceKey, Connection[]> = new Map<InstanceKey, Connection[]>()
+  private publisherConnectionProxies: Map<InstanceKey, Connection[]> = new Map<InstanceKey, Connection[]>()
 
-  public static getUsableCachedConnection(purpose: ConnectionPurpose, streamName: string, vhost: string, host: string) {
-    const map =
-      purpose === "publisher" ? ConnectionPool.publisherConnectionProxies : ConnectionPool.consumerConnectionProxies
-    const key = ConnectionPool.getCacheKey(streamName, vhost, host)
+  public getCachedConnection(purpose: ConnectionPurpose, streamName: string, vhost: string, host: string) {
+    const map = purpose === "publisher" ? this.publisherConnectionProxies : this.consumerConnectionProxies
+    const key = this.getCacheKey(streamName, vhost, host)
     const proxies = map.get(key) || []
     const connection = proxies.at(-1)
     const refCount = connection?.refCount
     return refCount !== undefined && refCount < getMaxSharedConnectionInstances() ? connection : undefined
   }
 
-  public static cacheConnection(
+  public cacheConnection(
     purpose: ConnectionPurpose,
     streamName: string,
     vhost: string,
     host: string,
     client: Connection
   ) {
-    const map =
-      purpose === "publisher" ? ConnectionPool.publisherConnectionProxies : ConnectionPool.consumerConnectionProxies
-    const key = ConnectionPool.getCacheKey(streamName, vhost, host)
+    const map = purpose === "publisher" ? this.publisherConnectionProxies : this.consumerConnectionProxies
+    const key = this.getCacheKey(streamName, vhost, host)
     const currentlyCached = map.get(key) || []
     currentlyCached.push(client)
     map.set(key, currentlyCached)
   }
 
-  public static removeIfUnused(connection: Connection) {
+  public removeIfUnused(connection: Connection) {
     if (connection.refCount <= 0) {
-      ConnectionPool.removeCachedConnection(connection)
+      this.removeCachedConnection(connection)
       return true
     }
     return false
   }
 
-  public static removeCachedConnection(connection: Connection) {
+  public removeCachedConnection(connection: Connection) {
     const { leader, streamName, hostname: host, vhost } = connection
     if (streamName === undefined) return
-    const m = leader ? ConnectionPool.publisherConnectionProxies : ConnectionPool.consumerConnectionProxies
-    const k = ConnectionPool.getCacheKey(streamName, vhost, host)
+    const m = leader ? this.publisherConnectionProxies : this.consumerConnectionProxies
+    const k = this.getCacheKey(streamName, vhost, host)
     const mappedClientList = m.get(k)
     if (mappedClientList) {
       const filtered = mappedClientList.filter((c) => c !== connection)
       m.set(k, filtered)
     }
   }
 
-  private static getCacheKey(streamName: string, vhost: string, host: string) {
+  private getCacheKey(streamName: string, vhost: string, host: string) {
     return `${streamName}@${vhost}@${host}`
   }
 }

src/connection_pool.ts

@@ -68,6 +68,7 @@ export class StreamConsumer implements Consumer {
   private singleActive: boolean = false
 
   constructor(
+    private pool: ConnectionPool,
     handle: ConsumerFunc,
     params: {
       connection: Connection
@@ -99,7 +100,7 @@ export class StreamConsumer implements Consumer {
   async close(manuallyClose: boolean): Promise<void> {
     this.closed = true
     this.connection.decrRefCount()
-    if (ConnectionPool.removeIfUnused(this.connection)) {
+    if (this.pool.removeIfUnused(this.connection)) {
       await this.connection.close({ closingCode: 0, closingReason: "", manuallyClose })
     }
   }

src/consumer.ts

@@ -173,6 +173,7 @@ export class StreamPublisher implements Publisher {
   private _closed = false
 
   constructor(
+    private pool: ConnectionPool,
     params: {
       connection: Connection
       stream: string
@@ -281,7 +282,7 @@ export class StreamPublisher implements Publisher {
     if (!this.closed) {
       await this.flush()
       this.connection.decrRefCount()
-      if (ConnectionPool.removeIfUnused(this.connection)) {
+      if (this.pool.removeIfUnused(this.connection)) {
         await this.connection.close({ closingCode: 0, closingReason: "", manuallyClose })
       }
     }