added pooling

Author: daimor

src/IRISDriver.ts

@@ -23,7 +23,7 @@ import { TableForeignKey } from "typeorm/schema-builder/table/TableForeignKey"
 import { InstanceChecker } from "typeorm/util/InstanceChecker"
 import { UpsertType } from "typeorm/driver/types/UpsertType"
 import { TypeORMError } from "typeorm/error/TypeORMError"
-import { IRISNative } from "./IRISNative"
+import { IRISConnectionPool } from "./IRISNative"
 import { ConnectionIsNotSetError } from "typeorm"
 
 /**
@@ -41,7 +41,7 @@ export class IRISDriver implements Driver {
 
     iris: any
 
-    master: any
+    master: IRISConnectionPool
 
     // -------------------------------------------------------------------------
     // Public Implemented Properties
@@ -302,8 +302,7 @@ export class IRISDriver implements Driver {
         options: IRISConnectionOptions,
         credentials: IRISConnectionCredentialsOptions,
     ): Promise<any> {
-        const { logger } = this.connection
-        logger.log("info", "Creating connection...")
+        // const { logger } = this.connection
         credentials = Object.assign({}, credentials)
         const connectionOptions = Object.assign(
             {},
@@ -318,7 +317,10 @@ export class IRISDriver implements Driver {
             },
             options.extra || {},
         )
-        const conn = IRISNative.createConnection(connectionOptions)
+        const conn = new IRISConnectionPool(
+            connectionOptions,
+            options.poolSize || 5,
+        )
         return Promise.resolve(conn)
     }
 
@@ -337,8 +339,8 @@ export class IRISDriver implements Driver {
             throw new ConnectionIsNotSetError("iris")
         }
 
-        this.master.close()
-        this.master = undefined
+        // this.master.close()
+        // this.master = undefined
     }
 
     /**
@@ -470,13 +472,6 @@ export class IRISDriver implements Driver {
      * Prepares given value to a value to be persisted, based on its column type and metadata.
      */
     preparePersistentValue(value: any, columnMetadata: ColumnMetadata): any {
-        // console.log(
-        //   "IRISDriver.preparePersistentValue:",[
-        //   columnMetadata.databaseName,
-        //   columnMetadata.type,
-        //   typeof value,
-        //   value,
-        // ])
         if (columnMetadata.transformer)
             value = ApplyValueTransformers.transformTo(
                 columnMetadata.transformer,
@@ -513,12 +508,6 @@ export class IRISDriver implements Driver {
      * Prepares given value to a value to be persisted, based on its column type or metadata.
      */
     prepareHydratedValue(value: any, columnMetadata: ColumnMetadata): any {
-        // console.log("IRISDriver.prepareHydratedValue:", [
-        //   columnMetadata.databaseName,
-        //   columnMetadata.type,
-        //   typeof value,
-        //   value,
-        // ])
         if (value === null || value === undefined)
             return columnMetadata.transformer
                 ? ApplyValueTransformers.transformFrom(
@@ -696,12 +685,11 @@ export class IRISDriver implements Driver {
      * Used for replication.
      * If replication is not setup then returns default connection's database connection.
      */
-    obtainMasterConnection(): Promise<any> {
+    async obtainMasterConnection(): Promise<any> {
         if (!this.master) {
             throw new TypeORMError("Driver not Connected")
         }
-        // return this.createConnection(this.options, this.options)
-        return Promise.resolve(this.master)
+        return this.master.getConnection()
     }
 
     /**

src/IRISNative.ts

@@ -2,7 +2,9 @@ const IRISNative = require("@intersystems/intersystems-iris-native")
 import { Connection } from "@intersystems/intersystems-iris-native"
 
 export interface IRISConnection extends Connection {
-    query: (queryString: string, argSets?: any[]) => any
+    connectionId: string
+    query: (queryString: string, argSets?: any[]) => Promise<any>
+    release: () => Promise<void>
 }
 
 type IRISConnectionOptions = {
@@ -228,7 +230,6 @@ class ResultSet {
             })
             this.rows.push(row)
         }
-        // console.log("ResultSet rows fetched:", this.rows.length);
     }
 }
 
@@ -271,7 +272,7 @@ function normalizeArgs(args: any[]) {
 
 function createQuery(connection: any) {
     const iris = connection.createIris()
-    return function (queryString: string, argSets: any[]): any {
+    return async function (queryString: string, argSets: any[]): Promise<any> {
         argSets =
             argSets && argSets.length && Array.isArray(argSets[0])
                 ? argSets
@@ -304,10 +305,10 @@ function createQuery(connection: any) {
             }
             resultSet = st.execute(...args)
             if (resultSet.Message) {
-                return resultSet
+                return Promise.resolve(resultSet)
             }
         }
-        return resultSet
+        return Promise.resolve(resultSet)
     }
 }
 
@@ -322,4 +323,114 @@ IRISNative.createConnection = (
 }
 IRISNative.connect = IRISNative.createConnection
 
-export { IRISNative }
+class IRISConnectionPool {
+    private options: IRISConnectionOptions
+    private maxConnections: number
+    private pool: IRISConnection[] = []
+    private activeConnections: Set<IRISConnection> = new Set()
+    private connectionCount: number = 0
+
+    constructor(options: IRISConnectionOptions, maxConnections: number = 5) {
+        this.options = options
+        this.maxConnections = maxConnections
+    }
+
+    async getConnection(): Promise<IRISConnection> {
+        // Try to reuse an existing connection from pool
+        if (this.pool.length > 0) {
+            const connection = this.pool.pop()!
+            this.activeConnections.add(connection)
+            return connection
+        }
+
+        // Create new connection if under limit
+        if (this.connectionCount < this.maxConnections) {
+            try {
+                const connection = await new Promise<IRISConnection>(
+                    (resolve, reject) => {
+                        try {
+                            const conn = IRISNative.createConnection(
+                                this.options,
+                            )
+                            this.connectionCount++
+                            conn.connectionId = `conn-${this.connectionCount}`
+                            conn.release = () => {
+                                this.releaseConnection(conn)
+                                return Promise.resolve()
+                            }
+                            resolve(conn)
+                        } catch (error) {
+                            reject(error)
+                        }
+                    },
+                )
+                this.activeConnections.add(connection)
+                return connection
+            } catch (error) {
+                console.error("Failed to create connection:", error)
+                throw error
+            }
+        }
+
+        // Wait for a connection to become available
+        // throw new Error("Max connections reached, please try again later.")
+        return new Promise<IRISConnection>((resolve) => {
+            const checkPool = (): void => {
+                if (this.pool.length > 0) {
+                    const connection = this.pool.pop()!
+                    this.activeConnections.add(connection)
+                    resolve(connection)
+                } else {
+                    setTimeout(checkPool, 100)
+                }
+            }
+            checkPool()
+        })
+    }
+
+    releaseConnection(connection: IRISConnection): void {
+        if (this.activeConnections.has(connection)) {
+            this.activeConnections.delete(connection)
+            this.pool.push(connection)
+        }
+    }
+
+    async closeAll(): Promise<void> {
+        // Close active connections
+        for (const connection of this.activeConnections) {
+            try {
+                connection.close()
+            } catch (error) {
+                console.error("Error closing active connection:", error)
+            }
+        }
+        this.activeConnections.clear()
+
+        // Close pooled connections
+        for (const connection of this.pool) {
+            try {
+                connection.close()
+            } catch (error) {
+                console.error("Error closing pooled connection:", error)
+            }
+        }
+        this.pool.length = 0
+        this.connectionCount = 0
+
+        // Force garbage collection if available
+        if ((global as any).gc) {
+            ;(global as any).gc()
+            console.log("Forced garbage collection")
+        }
+    }
+
+    getPoolStats(): { active: number; pooled: number; total: number } {
+        return {
+            active: this.activeConnections.size,
+            pooled: this.pool.length,
+            total: this.connectionCount,
+        }
+    }
+}
+
+export { IRISNative, IRISConnectionPool }

src/IRISQueryRunner.ts

@@ -40,14 +40,16 @@ export class IRISQueryRunner extends BaseQueryRunner implements QueryRunner {
      */
     driver: IRISDriver
 
+    protected releaseCallback?: () => void
+
     // -------------------------------------------------------------------------
     // Protected Properties
     // -------------------------------------------------------------------------
 
     /**
      * Promise used to obtain a database connection from a pool for a first time.
      */
-    // protected databaseConnectionPromise: Promise<any>
+    protected databaseConnectionPromise: Promise<any>
 
     // -------------------------------------------------------------------------
     // Constructor
@@ -70,15 +72,31 @@ export class IRISQueryRunner extends BaseQueryRunner implements QueryRunner {
      * Returns obtained database connection.
      */
     async connect(): Promise<any> {
-        return this.driver.obtainMasterConnection()
+        if (this.databaseConnection) {
+            return Promise.resolve(this.databaseConnection)
+        }
+
+        if (this.databaseConnectionPromise)
+            return this.databaseConnectionPromise
+
+        this.databaseConnectionPromise = this.driver
+            .obtainMasterConnection()
+            .then((connection) => {
+                this.databaseConnection = connection
+                return this.databaseConnection
+            })
+
+        return this.databaseConnectionPromise
     }
 
     /**
      * Releases used database connection.
      * You cannot use query runner methods once its released.
      */
     async release(): Promise<void> {
-        // releasing connection are not supported by mongodb driver, so simply don't do anything here
+        this.isReleased = true
+        if (this.databaseConnection) this.databaseConnection.release()
+        return Promise.resolve()
     }
 
     /**
@@ -160,7 +178,6 @@ export class IRISQueryRunner extends BaseQueryRunner implements QueryRunner {
         useStructuredResult = false,
     ): Promise<any> {
         if (this.isReleased) throw new QueryRunnerAlreadyReleasedError()
-        // const isInsertQuery = query.startsWith("INSERT INTO ")
 
         const databaseConnection = await this.connect()
 
@@ -208,6 +225,7 @@ export class IRISQueryRunner extends BaseQueryRunner implements QueryRunner {
             return Promise.reject(err)
         } finally {
             await broadcasterResult.wait()
+            databaseConnection.release()
         }
     }
 

test/iris.spec.ts

@@ -15,13 +15,13 @@ describe("IRISNative test", () => {
         const res = await connection.query(
             "SELECT 1 AS test1, '2' AS test2",
             [],
-        ).rows
-        res.should.be.an("array")
-        res.should.have.lengthOf(1)
-        res[0].should.be.an("object")
-        res[0].should.have.property("test1")
-        res[0].should.have.property("test2")
-        res[0].should.have.property("test1", 1)
-        res[0].should.have.property("test2", "2")
+        )
+        res.rows.should.be.an("array")
+        res.rows.should.have.lengthOf(1)
+        res.rows[0].should.be.an("object")
+        res.rows[0].should.have.property("test1")
+        res.rows[0].should.have.property("test2")
+        res.rows[0].should.have.property("test1", 1)
+        res.rows[0].should.have.property("test2", "2")
     })
 })

test/utils/test-utils.ts

@@ -20,6 +20,7 @@ export function getTypeOrmConfig(): IRISConnectionOptions[] {
             namespace: globalOptions.ns,
             username: globalOptions.user,
             password: globalOptions.pwd,
+            poolSize: 5,
         },
     ]
 }