import { asyncScheduler , BehaviorSubject , Observable , OperatorFunction , throwError , timer , } from 'rxjs' ; import { filter , map , mergeMap , take } from 'rxjs/operators' ; export const genericRetryStrategy = ( { maxRetryAttempts = 3 , scalingDuration = 1000 , excludedStatusCodes = [ ] , } = { } ) => ( attempts ) => { return attempts . pipe ( mergeMap ( ( error , i ) => { const retryAttempt = i + 1 ; if ( retryAttempt > maxRetryAttempts || excludedStatusCodes . find ( ( e ) => e === error . status ) ) { return throwError ( error ) ; } return timer ( retryAttempt * scalingDuration ) ; } ) ) ; } ; export function rateLimit < T > ( count , slidingWindowTime , scheduler = asyncScheduler ) { let tokens = count ; const tokenChanged = new BehaviorSubject ( tokens ) ; const consumeToken = ( ) => tokenChanged . next ( -- tokens ) ; const renewToken = ( ) => tokenChanged . next ( ++ tokens ) ; const availableTokens = tokenChanged . pipe ( filter ( ( ) => tokens > 0 ) ) ; return function ( source ) { return source . pipe ( mergeMap < T , Observable < T > > ( ( value ) => availableTokens . pipe ( take ( 1 ) , map ( ( ) => { consumeToken ( ) ; timer ( slidingWindowTime , scheduler ) . subscribe ( renewToken ) ; return value ; } ) ) ) ) ; } ; }