import { ExecuteStatementResult , FetchPageResult , Page , } from "aws-sdk/clients/qldbsession" ; import { dom } from "ion-js" ; import { Readable } from "stream" ; import { Communicator } from "./Communicator" ; import { Result } from "./Result" ; import { IOUsage } from "./stats/IOUsage" ; import { TimingInformation } from "./stats/TimingInformation" ; /**
 * A class representing the result of a statement returned from QLDB as a stream.
 * Extends Readable from the Node.JS Stream API interface.
 * The stream will always operate in object mode.
 */ export class ResultReadable extends Readable { private _communicator : Communicator ; private _cachedPage : Page ; private _txnId : string ; private _shouldPushCachedPage : boolean ; private _retrieveIndex : number ; private _isPushingData : boolean ; private _readIOs : number ; private _processingTime : number ; param txnId param executeResult param communicator internal constructor ( txnId , executeResult , communicator ) { super ( { objectMode : true } ) ; this . _communicator = communicator ; this . _cachedPage = executeResult . FirstPage ; this . _txnId = txnId ; this . _shouldPushCachedPage = true ; this . _retrieveIndex = 0 ; this . _isPushingData = false ; this . _readIOs = executeResult . ConsumedIOs == null ? null : executeResult . ConsumedIOs . ReadIOs ; this . _processingTime = executeResult . TimingInformation == null ? null : executeResult . TimingInformation . ProcessingTimeMilliseconds ; } returns getConsumedIOs ( ) { return this . _readIOs == null ? null : new IOUsage ( this . _readIOs ) ; } returns getTimingInformation ( ) { return this . _processingTime == null ? null : new TimingInformation ( this . _processingTime ) ; } param size internal _read ( size ? ) { if ( this . _isPushingData ) { return ; } this . _isPushingData = true ; this . _pushPageValues ( ) ; } returns private async _pushPageValues ( ) { let canPush = true ; try { if ( this . _shouldPushCachedPage ) { this . _shouldPushCachedPage = false ; } else if ( this . _cachedPage . NextPageToken ) { try { const fetchPageResult = await this . _communicator . fetchPage ( this . _txnId , this . _cachedPage . NextPageToken ) ; this . _cachedPage = fetchPageResult . Page ; if ( fetchPageResult . ConsumedIOs != null ) { this . _readIOs += fetchPageResult . ConsumedIOs . ReadIOs ; } if ( fetchPageResult . TimingInformation != null ) { this . _processingTime += fetchPageResult . TimingInformation . ProcessingTimeMilliseconds ; } this . _retrieveIndex = 0 ; } catch ( e ) { this . destroy ( e ) ; canPush = false ; return ; } } while ( this . _retrieveIndex < this . _cachedPage . Values . length ) { const ionValue = dom . load ( Result . _handleBlob ( this . _cachedPage . Values [ this . _retrieveIndex ++ ] . IonBinary ) ) ; canPush = this . push ( ionValue ) ; if ( ! canPush ) { this . _shouldPushCachedPage = this . _retrieveIndex < this . _cachedPage . Values . length ; return ; } } if ( ! this . _cachedPage . NextPageToken ) { this . push ( null ) ; canPush = false ; } } finally { this . _isPushingData = false ; if ( canPush ) { this . _read ( ) ; } } } }