module license import * as events from 'events' ; import { BackoffOptions , DEFAULT_BACKOFF_OPTIONS , getBackoffMs , } from './backoff' ; import { Stream } from './stream' ; import { sleep } from '../utils/utils' ; typeparam param getStream param backoffOptions param shouldRetry return example export function retryStream < Message > ( getStream , backoffOptions ? , shouldRetry = ( ) => true ) { return new RetryingStreamImpl < Message > ( getStream , { ... DEFAULT_BACKOFF_OPTIONS , ... backoffOptions , } , shouldRetry ) ; } /**
 * A "concatenation" of all the streams initiated by [[retryStream]].
 */ export interface RetryingStream < Message > extends Stream < Message > { start ( ) : this ; cancel ( ) : this ; /**
     * Register an event listener.
     *
     * In addition to the events from the `Stream` interface, the following
     * events are emitted:
     *
     * - `retryingError`: When an error occurred, with the following parameters:
     *   - `err`: The error that occurred
     *   - `retriesSinceLastReady`: The number of retries since a stream emitted
     * the `ready` event.
     *   - `abandoned`: Whether the retrying stream gave up because of this error.
     * - `error`: When an error occurred that made the retrying stream give up.
     */ on ( event , callback ) : this ; on ( event , callback ) : this ; on ( event , callback ) : this ; on ( event , callback ) : this ; } /**
 * All the states of a retrying stream are gathered here for better clarity.
 */ enum RetryingStreamState { /** Initial state */ initial = 'initial' , /** The `start()` function was called. */ started = 'started' , /** The `ready` event was emitted. */ ready = 'ready' , /** Either the `cancel()` function was called, or the `cancel` event was emitted. */ canceled = 'canceled' , /** The max number of retries has been reached, the RPC has been abandoned. */ abandoned = 'abandoned' , /** The RPC successfully completed. */ complete = 'complete' } class RetryingStreamImpl < Message > extends events . EventEmitter implements RetryingStream < Message > { /** The state of the retrying stream.  We use this to enforce state transitions. */ private state : RetryingStreamState = RetryingStreamState . initial ; /**
     * If a stream (provided by the stream provider) is currently open, the
     * `cancel()` function of this stream.
     */ private cancelCurrentStream : ( ( ) => void ) | undefined ; /**
     * The number of retries since a stream emitted the `ready` event.
     */ private retriesSinceLastReady : number = 0 ; constructor ( private readonly getStream , private readonly options , private readonly shouldRetry ) { super ( ) ; } private attempt = ( stream , cb ) => { stream . on ( 'ready' , ( ) => { this . state = RetryingStreamState . ready ; this . emit ( 'ready' ) ; this . retriesSinceLastReady = 0 ; } ) . on ( 'message' , message => { if ( this . state === RetryingStreamState . ready ) { this . emit ( 'message' , message ) ; } } ) . on ( 'complete' , ( ) => { this . state = RetryingStreamState . complete ; this . emit ( 'complete' ) ; cb ( ) ; } ) . on ( 'error' , async ( err ) => { if ( ( this . options . maxRetries >= 0 && this . retriesSinceLastReady >= this . options . maxRetries ) || ! this . shouldRetry ( err ) ) { this . state = RetryingStreamState . abandoned ; this . emit ( 'retryingError' , err , this . retriesSinceLastReady , true ) ; this . emit ( 'error' , err ) ; } else { this . state = RetryingStreamState . started ; this . emit ( 'retryingError' , err , this . retriesSinceLastReady , false ) ; await sleep ( getBackoffMs ( this . options , this . retriesSinceLastReady ++ ) ) ; } cb ( ) ; } ) . on ( 'canceled' , ( ) => { this . state = RetryingStreamState . canceled ; this . emit ( 'canceled' ) ; cb ( ) ; } ) . start ( ) ; } ; start ( ) { this . asyncStart ( ) ; return this ; } cancel ( ) { this . state = RetryingStreamState . canceled ; this . cancelCurrentStream && this . cancelCurrentStream ( ) ; return this ; } private async asyncStart ( ) { this . ensureState ( RetryingStreamState . initial ) ; this . state = RetryingStreamState . started ; while ( ! [ RetryingStreamState . complete , RetryingStreamState . canceled , RetryingStreamState . abandoned , ] . includes ( this . state ) ) { const stream = this . getStream ( ) ; this . cancelCurrentStream = stream . cancel . bind ( stream ) ; await new Promise < void > ( accept => { this . attempt ( stream , accept ) ; } ) ; this . cancelCurrentStream = undefined ; } } private ensureState ( state ) { if ( this . state !== state ) { throw new Error ( `invalid retrier state: actual: ${ this . state }; expected: ${ state }` ) ; } } }