import { JestProcess } from './JestProcess' ; import { TaskArrayFunctions , JestProcessRequest , QueueType , Task , JestProcessInfo } from './types' ; import { Logging } from '../logging' ; import { createTaskQueue , TaskQueue } from './task-queue' ; import { isDup , requestString } from './helper' ; import { JestExtContext } from '../JestExt' ; export class JestProcessManager implements TaskArrayFunctions < JestProcess > { private extContext : JestExtContext ; private queues : Map < QueueType , TaskQueue < JestProcess > > ; private logging : Logging ; constructor ( extContext ) { this . extContext = extContext ; this . logging = extContext . loggingFactory . create ( 'JestProcessManager' ) ; this . queues = new Map ( [ [ 'blocking' , createTaskQueue ( 'blocking-queue' , 1 ) ] , [ 'blocking-2' , createTaskQueue ( 'blocking-queue-2' , 1 ) ] , [ 'non-blocking' , createTaskQueue ( 'non-blocking-queue' , 3 ) ] , ] ) ; } private getQueue ( type ) { return this . queues . get ( type ) ! ; } private foundDup ( request ) { if ( ! request . schedule . dedup ) { return false ; } const queue = this . getQueue ( request . schedule . queue ) ; const dupTasks = queue . filter ( ( p ) => isDup ( p , request ) ) ; if ( dupTasks . length > 0 ) { this . logging ( 'debug' , `found ${ dupTasks . length } duplicate processes, will not schedule request:` , request ) ; return true ; } return false ; } param request returns public scheduleJestProcess ( request ) { if ( this . foundDup ( request ) ) { this . logging ( 'debug' , `duplicate request found, process is not scheduled: ${ requestString ( request ) }` ) ; return ; } const queue = this . getQueue ( request . schedule . queue ) ; const process = new JestProcess ( this . extContext , request ) ; queue . add ( process ) ; this . run ( queue ) ; return process ; } private async run ( queue ) { const task = queue . getRunnableTask ( ) ; if ( ! task ) { return ; } const process = task . data ; try { this . logging ( 'debug' , 'starting process:' , process ) ; await process . start ( ) ; this . logging ( 'debug' , 'process ended:' , process ) ; } catch ( e ) { this . logging ( 'error' , `${ queue . name }: process failed:` , process , e ) ; } finally { queue . remove ( task ) ; } return this . run ( queue ) ; } /** stop and remove all process matching the queue type, if no queue type specified, stop all queues */ public async stopAll ( queueType ? ) { let promises ; if ( ! queueType ) { promises = Array . from ( this . queues . keys ( ) ) . map ( ( q ) => this . stopAll ( q ) ) ; } else { const queue = this . getQueue ( queueType ) ; promises = queue . map ( ( t ) => t . data . stop ( ) ) ; queue . reset ( ) ; } await Promise . allSettled ( promises ) ; return ; } public numberOfProcesses ( queueType ? ) { if ( queueType ) { return this . getQueue ( queueType ) . size ( ) ; } return Array . from ( this . queues . values ( ) ) . reduce ( ( pCount , q ) => { pCount += q . size ( ) ; return pCount ; } , 0 ) ; } private getQueues ( queueType ? ) { return queueType ? [ this . getQueue ( queueType ) ] : Array . from ( this . queues . values ( ) ) ; } public map < M > ( f , queueType ? ) { const queues = this . getQueues ( queueType ) ; return queues . reduce ( ( list , q ) => { list . push ( ... q . map ( f ) ) ; return list ; } , [ ] as M [ ] ) ; } public filter ( f , queueType ? ) { const queues = this . getQueues ( queueType ) ; return queues . reduce ( ( list , q ) => { list . push ( ... q . filter ( f ) ) ; return list ; } , [ ] as Task < JestProcess > [ ] ) ; } public find ( f , queueType ? ) { const queues = this . getQueues ( queueType ) ; for ( const q of queues ) { const t = q . find ( f ) ; if ( t ) { return t ; } } } }