import { Observable } from 'rxjs' ; import { concatMap , last } from 'rxjs/operators' ; import { ClaimOwnershipResult } from '../queries/task_claiming' ; import { ConcreteTaskInstance } from '../task' ; import { WithTaskTiming , startTaskTimer } from '../task_events' ; import { TaskPoolRunResult } from '../task_pool' ; import { TaskManagerRunner } from '../task_running' ; import { Result , map as mapResult , asErr , asOk } from './result_type' ; export enum FillPoolResult { Failed = 'Failed' , NoAvailableWorkers = 'NoAvailableWorkers' , NoTasksClaimed = 'NoTasksClaimed' , RunningAtCapacity = 'RunningAtCapacity' , RanOutOfCapacity = 'RanOutOfCapacity' , PoolFilled = 'PoolFilled' } type FillPoolAndRunResult = Result < { result : TaskPoolRunResult ; stats ? : ClaimOwnershipResult [ 'stats' ] ; } , { result : FillPoolResult ; stats ? : ClaimOwnershipResult [ 'stats' ] ; } > ; export type ClaimAndFillPoolResult = Partial < Pick < ClaimOwnershipResult , 'stats' > > & { result : FillPoolResult ; } ; export type TimedFillPoolResult = WithTaskTiming < ClaimAndFillPoolResult > ; param run param fetchAvailableTasks param converter export async function fillPool ( fetchAvailableTasks , converter , run ) { return new Promise ( ( resolve , reject ) => { const stopTaskTimer = startTaskTimer ( ) ; const augmentTimingTo = ( result , stats ? ) : TimedFillPoolResult => ( { result , stats , timing : stopTaskTimer ( ) , } ) ; fetchAvailableTasks ( ) . pipe ( concatMap ( async ( res ) => mapResult < ClaimOwnershipResult , FillPoolResult , Promise < FillPoolAndRunResult > > ( res , async ( { docs , stats } ) => { if ( ! docs . length ) { return asOk ( { result : TaskPoolRunResult . NoTaskWereRan , stats } ) ; } return asOk ( await run ( docs . map ( converter ) ) . then ( ( runResult ) => ( { result : runResult , stats , } ) ) ) ; } , async ( fillPoolResult ) => asErr ( { result : fillPoolResult } ) ) ) , last ( ) ) . subscribe ( ( claimResults ) => { resolve ( mapResult ( claimResults , ( { result , stats } ) => { switch ( result ) { case TaskPoolRunResult . RanOutOfCapacity : return augmentTimingTo ( FillPoolResult . RanOutOfCapacity , stats ) ; case TaskPoolRunResult . RunningAtCapacity : return augmentTimingTo ( FillPoolResult . RunningAtCapacity , stats ) ; case TaskPoolRunResult . NoTaskWereRan : return augmentTimingTo ( FillPoolResult . NoTasksClaimed , stats ) ; default : return augmentTimingTo ( FillPoolResult . PoolFilled , stats ) ; } } , ( { result , stats } ) => augmentTimingTo ( result , stats ) ) ) ; } , ( err ) => reject ( err ) ) ; } ) ; }