import { QLDBSession } from "aws-sdk" ; import { ClientConfiguration } from "aws-sdk/clients/qldbsession" ; import { globalAgent } from "http" ; import { dom } from "ion-js" ; import Semaphore from "semaphore-async-await" ; import { version } from "../package.json" ; import { Communicator } from "./Communicator" ; import { DriverClosedError , ExecuteError , SessionPoolEmptyError , } from "./errors/Errors" ; import { debug , info } from "./LogUtil" ; import { QldbSession } from "./QldbSession" ; import { Result } from "./Result" ; import { BackoffFunction } from "./retry/BackoffFunction" ; import { defaultRetryConfig } from "./retry/DefaultRetryConfig" ; import { RetryConfig } from "./retry/RetryConfig" ; import { TransactionExecutor } from "./TransactionExecutor" ; /**
  * This is the entry point for all interactions with Amazon QLDB.
  *
  * In order to start using the driver, you need to instantiate it with a ledger name:
  *
  * ```
  * let qldbDriver: QldbDriver = new QldbDriver(your-ledger-name);
  * ```
  * You can pass more parameters to the constructor of the driver which allow you to control certain limits
  * to improve the performance. Check the  QldbDriver . constructor to see all the available parameters.
  *
  * A single instance of the QldbDriver is attached to only one ledger. All transactions will be executed against
  * the ledger specified.
  *
  * The driver exposes  QldbDriver . executeLambda method which should be used to execute the transactions.
  * Check the  QldbDriver . executeLambda method for more details on how to execute the Transaction.
  export class QldbDriver { private _maxConcurrentTransactions : number ; private _availablePermits : number ; private _sessionPool : QldbSession [ ] ; private _semaphore : Semaphore ; private _qldbClient : QLDBSession ; private _ledgerName : string ; private _isClosed : boolean ; private _retryConfig : RetryConfig ; param ledgerName param qldbClientOptions The object containing options for configuring the low level client.
     *                          See  https .
     *  param maxConcurrentTransactions The driver internally uses a pool of sessions to execute the transactions.
     *                                  The maxConcurrentTransactions parameter specifies the number of sessions that the driver can hold in the pool.
     *                                  The default is set to maximum number of sockets specified in the globalAgent.
     *                                  See  https for more details.
     *  param retryConfig throws constructor ( ledgerName , qldbClientOptions = { } , maxConcurrentTransactions = 0 , retryConfig = defaultRetryConfig ) { qldbClientOptions . customUserAgent = `QLDB Driver for Node.js v${ version }` ; qldbClientOptions . maxRetries = 0 ; this . _qldbClient = new QLDBSession ( qldbClientOptions ) ; this . _ledgerName = ledgerName ; this . _isClosed = false ; this . _retryConfig = retryConfig ; if ( maxConcurrentTransactions < 0 ) { throw new RangeError ( "Value for maxConcurrentTransactions cannot be negative." ) ; } let maxSockets ; if ( qldbClientOptions . httpOptions && qldbClientOptions . httpOptions . agent ) { maxSockets = qldbClientOptions . httpOptions . agent . maxSockets ; } else { maxSockets = globalAgent . maxSockets ; } if ( 0 === maxConcurrentTransactions ) { this . _maxConcurrentTransactions = maxSockets ; } else { this . _maxConcurrentTransactions = maxConcurrentTransactions ; } if ( this . _maxConcurrentTransactions > maxSockets ) { throw new RangeError ( `The session pool limit given, ${ this . _maxConcurrentTransactions }, exceeds the limit set by the client,
                 ${ maxSockets }. Please lower the limit and retry.` ) ; } this . _availablePermits = this . _maxConcurrentTransactions ; this . _sessionPool = [ ] ; this . _semaphore = new Semaphore ( this . _maxConcurrentTransactions ) ; } /**
     * This is a driver shutdown method which closes all the sessions and marks the driver as closed.
     * Once the driver is closed, no transactions can be executed on that driver instance.
     *
     * Note: There is no corresponding `open` method and the only option is to instantiate another driver.
     */ close ( ) { this . _isClosed = true ; while ( this . _sessionPool . length > 0 ) { const session = this . _sessionPool . pop ( ) ; if ( session != undefined ) { session . endSession ( ) ; } } } param transactionLambda param retryConfig throws throws throws throws throws async executeLambda < T > ( transactionLambda , retryConfig ? ) { if ( this . _isClosed ) { throw new DriverClosedError ( ) ; } const getSession = async function ( thisDriver , startNewSession ) : Promise < QldbSession > { debug ( `Getting session. Current free session count: ${ thisDriver . _sessionPool . length }. ` + `Currently available permit count: ${ thisDriver . _availablePermits }.` ) ; if ( thisDriver . _semaphore . tryAcquire ( ) ) { thisDriver . _availablePermits -- ; try { let session ; if ( ! startNewSession ) { session = thisDriver . _sessionPool . pop ( ) ; } if ( startNewSession || session == undefined ) { debug ( "Creating a new session." ) ; const communicator = await Communicator . create ( thisDriver . _qldbClient , thisDriver . _ledgerName ) ; session = new QldbSession ( communicator ) ; } return session ; } catch ( e ) { thisDriver . _semaphore . release ( ) ; thisDriver . _availablePermits ++ ; throw new ExecuteError ( e , true , true ) ; } } else { throw new SessionPoolEmptyError ( ) ; } } ; const releaseSession = function ( thisDriver , session ) : boolean { if ( session != null ) { thisDriver . _semaphore . release ( ) ; thisDriver . _availablePermits ++ ; if ( session . isAlive ( ) ) { thisDriver . _sessionPool . push ( session ) ; return true ; } } return false ; } ; retryConfig = ( retryConfig == null ) ? this . _retryConfig : retryConfig ; let session ; let startNewSession = false ; for ( let retryAttempt = 1 ; true ; retryAttempt ++ ) { try { session = null ; session = await getSession ( this , startNewSession ) ; return await session . executeLambda ( transactionLambda ) ; } catch ( e ) { if ( e instanceof ExecuteError ) { if ( e . isRetryable ) { if ( retryAttempt == 1 && e . isInvalidSessionException ) { debug ( "Initial session received from pool is invalid. Retrying..." ) ; continue ; } if ( retryAttempt > retryConfig . getRetryLimit ( ) ) { throw e . cause ; } const backoffFunction = retryConfig . getBackoffFunction ( ) ; let backoffDelay = backoffFunction ( retryAttempt , e . cause , e . transactionId ) ; if ( backoffDelay == null || backoffDelay < 0 ) { backoffDelay = 0 ; } await new Promise ( resolve => setTimeout ( resolve , backoffDelay ) ) ; info ( `A recoverable error has occurred. Attempting retry #${ retryAttempt }.` ) ; debug ( `Error cause: ${ e . cause }` ) ; continue ; } else { throw e . cause ; } } else { throw e ; } } finally { startNewSession = ! releaseSession ( this , session ) ; } } } returns async getTableNames ( ) { const statement = "SELECT name FROM information_schema.user_tables WHERE status = 'ACTIVE'" ; return await this . executeLambda ( async ( transactionExecutor ) : Promise < string [ ] > => { const result = await transactionExecutor . execute ( statement ) ; const resultStructs = result . getResultList ( ) ; const listOfTableNames = resultStructs . map ( tableNameStruct => tableNameStruct . get ( "name" ) . stringValue ( ) ) ; return listOfTableNames ; } ) ; } }