import { Subject , merge , of , Observable , combineLatest , timer } from 'rxjs' ; import { mapTo , filter , scan , concatMap , tap , catchError , switchMap } from 'rxjs/operators' ; import { pipe } from 'fp-ts/lib/pipeable' ; import { Option , none , map as mapOptional , getOrElse } from 'fp-ts/lib/Option' ; import { Logger } from '../../../../../src/core/server' ; import { pullFromSet } from '../lib/pull_from_set' ; import { Result , Err , isErr , map as mapResult , asOk , asErr , promiseResult , } from '../lib/result_type' ; import { timeoutPromiseAfter } from './timeout_promise_after' ; type WorkFn < T , H > = ( ... params : T [ ] ) => Promise < H > ; interface Opts < T , H > { logger : Logger ; pollInterval$ : Observable < number > ; pollIntervalDelay$ : Observable < number > ; bufferCapacity : number ; getCapacity : ( ) => number ; pollRequests$ : Observable < Option < T > > ; work : WorkFn < T , H > ; workTimeout : number ; } param opts prop prop prop prop prop returns { Observable < Set < T > > } export function createTaskPoller < T , H > ( { logger , pollInterval$ , pollIntervalDelay$ , getCapacity , pollRequests$ , bufferCapacity , work , workTimeout , } ) { const hasCapacity = ( ) => getCapacity ( ) > 0 ; const errors$ = new Subject < Err < PollingError < T > > > ( ) ; const requestWorkProcessing$ = merge ( pollRequests$ , combineLatest ( [ pollInterval$ . pipe ( tap ( ( period ) => { logger . debug ( `Task poller now using interval of ${ period }ms` ) ; } ) ) , pollIntervalDelay$ . pipe ( tap ( ( pollDelay ) => { logger . debug ( `Task poller now delaying emission by ${ pollDelay }ms` ) ; } ) ) , ] ) . pipe ( switchMap ( ( [ period , pollDelay ] ) => timer ( period + ( pollDelay % period ) , period ) ) , mapTo ( none ) ) ) . pipe ( scan < Option < T > , Set < T > > ( ( queue , request ) => { if ( isErr ( pushOptionalIntoSet ( queue , bufferCapacity , request ) ) ) { errors$ . next ( asPollingError < T > ( `request capacity reached` , PollingErrorType . RequestCapacityReached , request ) ) ; } return queue ; } , new Set < T > ( ) ) , filter ( hasCapacity ) , concatMap ( async ( set ) => { return mapResult < H , Error , Result < H , PollingError < T > > > ( await promiseResult < H , Error > ( timeoutPromiseAfter < H , Error > ( work ( ... pullFromSet ( set , getCapacity ( ) ) ) , workTimeout , ( ) => new Error ( `work has timed out` ) ) ) , ( workResult ) => asOk ( workResult ) , ( err ) => asPollingError < T > ( err , PollingErrorType . WorkError ) ) ; } ) , catchError ( ( err ) => of ( asPollingError < T > ( err , PollingErrorType . WorkError ) ) ) ) ; return merge ( requestWorkProcessing$ , errors$ ) ; } param set param maxCapacity param value function pushOptionalIntoSet < T > ( set , maxCapacity , value ) { return pipe ( value , mapOptional < T , Result < Set < T > , Set < T > > > ( ( req ) => { if ( set . size >= maxCapacity ) { return asErr ( set ) ; } set . add ( req ) ; return asOk ( set ) ; } ) , getOrElse ( ( ) => asOk ( set ) as Result < Set < T > , Set < T > > ) ) ; } export enum PollingErrorType { WorkError , WorkTimeout , RequestCapacityReached } function asPollingError < T > ( err , type , data = none ) { return asErr ( new PollingError < T > ( `Failed to poll for work: ${ err }` , type , data ) ) ; } export class PollingError < T > extends Error { public readonly type : PollingErrorType ; public readonly data : Option < T > ; constructor ( message , type , data ) { super ( message ) ; Object . setPrototypeOf ( this , new . target . prototype ) ; this . type = type ; this . data = data ; } }