import { Observable } from 'rxjs' ; import { JobNumbered } from '../../jobTypes' ; type JobWithRetries = JobNumbered & { retryNumber ? : number ; } ; type BackoffMethod = ( retryNumber : number ) => number ; const waitedJob = ( job , timeToWait ) => new Promise ( res => setTimeout ( ( ) => res ( job ) , timeToWait ) ) ; const backoffStream = ( backoffMethod ) => ( subject ) => { return subject . map ( job => { const jobWithRetries = < JobWithRetries > job ; return jobWithRetries . retryNumber ? { ... jobWithRetries , retryNumber : jobWithRetries . retryNumber + 1 } : { ... jobWithRetries , retryNumber : 1 } ; } ) . flatMap ( async ( job ) => waitedJob ( job , backoffMethod ( job . retryNumber - 1 ) ) ) ; } ; const exponentialBackoffMethod = ( initialWaitTime , maxWaitTime ? ) => ( retryNumber ) => { const waitTime = initialWaitTime * Math . pow ( 2 , retryNumber ) ; return maxWaitTime && maxWaitTime < waitTime ? maxWaitTime : waitTime ; } ; const fibonacci = ( index ) : number => { if ( index === 0 ) return 1 ; if ( index === 1 ) return 1 ; return fibonacci ( index - 1 ) + fibonacci ( index - 2 ) ; } ; const fibonacciBackoffMethod = ( initialWaitTime , maxWaitTime ? ) => ( retryNumber ) => { const waitTime = initialWaitTime * fibonacci ( retryNumber + 1 ) ; return maxWaitTime && maxWaitTime < waitTime ? maxWaitTime : waitTime ; } ; const withBackoff = { exponential : ( initialWaitTime , maxWaitTime ? ) => backoffStream ( exponentialBackoffMethod ( initialWaitTime , maxWaitTime ) ) , fibonacci : ( initialWaitTime , maxWaitTime ? ) => backoffStream ( fibonacciBackoffMethod ( initialWaitTime , maxWaitTime ) ) } ; export default withBackoff ;