import * as mysql from 'mysql' ; import { MysqlError , Pool as MySQLPool , PoolConnection } from 'mysql' ; import { Pool as PgPool , PoolClient } from 'pg' ; import { log , generateError } from './FsOps' ; import Conversion from './Conversion' ; import DBVendors from './DBVendors' ; import DBAccessQueryResult from './DBAccessQueryResult' ; import IDBAccessQueryParams from './IDBAccessQueryParams' ; export default class DBAccess { /**
     * Ensures MySQL connection pool existence.
     */ private static async _getMysqlConnection ( conversion ) { if ( ! conversion . _mysql ) { conversion . _sourceConString . connectionLimit = conversion . _maxEachDbConnectionPoolSize ; conversion . _sourceConString . multipleStatements = true ; const pool = mysql . createPool ( conversion . _sourceConString ) ; if ( ! pool ) { await generateError ( conversion , '\t--[getMysqlConnection] Cannot connect to MySQL server...' ) ; process . exit ( 1 ) ; } conversion . _mysql = pool ; } } /**
     * Ensures PostgreSQL connection pool existence.
     */ private static async _getPgConnection ( conversion ) { if ( ! conversion . _pg ) { conversion . _targetConString . max = conversion . _maxEachDbConnectionPoolSize ; const pool = new PgPool ( conversion . _targetConString ) ; if ( ! pool ) { await generateError ( conversion , '\t--[getPgConnection] Cannot connect to PostgreSQL server...' ) ; process . exit ( 1 ) ; } conversion . _pg = pool ; conversion . _pg . on ( 'error' , async ( error ) => { const message = `Cannot connect to PostgreSQL server...\n' ${ error . message }\n${ error . stack }` ; await generateError ( conversion , message ) ; } ) ; } } /**
     * Closes both connection-pools.
     */ public static async closeConnectionPools ( conversion ) { const closeMySqlConnections = ( ) => { return new Promise ( resolve => { if ( conversion . _mysql ) { conversion . _mysql . end ( async ( error ) => { if ( error ) { await generateError ( conversion , `\t--[DBAccess::closeConnectionPools] ${ error }` ) ; } return resolve ( ) ; } ) ; } resolve ( ) ; } ) ; } ; const closePgConnections = async ( ) => { if ( conversion . _pg ) { try { await conversion . _pg . end ( ) ; } catch ( error ) { await generateError ( conversion , `\t--[DBAccess::closeConnectionPools] ${ error }` ) ; } } } ; await Promise . all ( [ closeMySqlConnections , closePgConnections ] ) ; log ( conversion , `\t--[DBAccess::closeConnectionPools] Closed all DB connections.` ) ; return conversion ; } /**
     * Obtains PoolConnection instance.
     */ public static getMysqlClient ( conversion ) { return new Promise < PoolConnection > ( async ( resolve , reject ) => { await DBAccess . _getMysqlConnection ( conversion ) ; ( < MySQLPool > conversion . _mysql ) . getConnection ( ( err , connection ) => { return err ? reject ( err ) : resolve ( connection ) ; } ) ; } ) ; } /**
     * Obtains PoolClient instance.
     */ public static async getPgClient ( conversion ) { await DBAccess . _getPgConnection ( conversion ) ; return ( < PgPool > conversion . _pg ) . connect ( ) ; } /**
     * Releases MySQL or PostgreSQL connection back to appropriate pool.
     */ public static async releaseDbClient ( conversion , dbClient ? ) { try { ( < PoolConnection | PoolClient > dbClient ) . release ( ) ; dbClient = undefined ; } catch ( error ) { await generateError ( conversion , `\t--[DBAccess::releaseDbClient] ${ error }` ) ; } } /**
     * Checks if there are no more queries to be sent using current client.
     * In such case the client should be released.
     */ private static async _releaseDbClientIfNecessary ( conversion , client , shouldHoldClient ) { if ( ! shouldHoldClient ) { await this . releaseDbClient ( conversion , client ) ; } } /**
     * Sends given SQL query to specified DB.
     * Performs appropriate actions (requesting/releasing client) against target connections pool.
     */ public static async query ( queryParams ) { let { conversion , caller , sql , vendor , processExitOnError , shouldReturnClient , client , bindings } = queryParams ; if ( ! client ) { try { client = vendor === DBVendors . PG ? await DBAccess . getPgClient ( conversion ) : await DBAccess . getMysqlClient ( conversion ) ; } catch ( error ) { await generateError ( conversion , `\t--[${ caller }] ${ error }` , sql ) ; return processExitOnError ? process . exit ( 1 ) : new DBAccessQueryResult ( client , undefined , error ) ; } } return vendor === DBVendors . PG ? DBAccess . _queryPG ( conversion , caller , sql , processExitOnError , shouldReturnClient , ( < PoolClient > client ) , bindings ) : DBAccess . _queryMySQL ( conversion , caller , sql , processExitOnError , shouldReturnClient , ( < PoolConnection > client ) , bindings ) ; } /**
     * Sends given SQL query to MySQL.
     */ private static _queryMySQL ( conversion , caller , sql , processExitOnError , shouldReturnClient , client ? , bindings ? ) { return new Promise < DBAccessQueryResult > ( ( resolve , reject ) => { if ( Array . isArray ( bindings ) ) { sql = ( < PoolConnection > client ) . format ( sql , bindings ) ; } ( < PoolConnection > client ) . query ( sql , async ( error , data ) => { await DBAccess . _releaseDbClientIfNecessary ( conversion , ( < PoolConnection > client ) , shouldReturnClient ) ; if ( error ) { await generateError ( conversion , `\t--[${ caller }] ${ error }` , sql ) ; return processExitOnError ? process . exit ( 1 ) : reject ( new DBAccessQueryResult ( client , undefined , error ) ) ; } return resolve ( new DBAccessQueryResult ( client , data , undefined ) ) ; } ) ; } ) ; } /**
     * Sends given SQL query to PostgreSQL.
     */ private static async _queryPG ( conversion , caller , sql , processExitOnError , shouldReturnClient , client ? , bindings ? ) { try { const data = Array . isArray ( bindings ) ? await ( < PoolClient > client ) . query ( sql , bindings ) : await ( < PoolClient > client ) . query ( sql ) ; return new DBAccessQueryResult ( client , data , undefined ) ; } catch ( error ) { await generateError ( conversion , `\t--[${ caller }] ${ error }` , sql ) ; return processExitOnError ? process . exit ( 1 ) : new DBAccessQueryResult ( client , undefined , error ) ; } finally { await DBAccess . _releaseDbClientIfNecessary ( conversion , ( < PoolClient > client ) , shouldReturnClient ) ; } } }