2 import { JobState, JobType } from '../../models'
3 import { wait } from '../miscs'
4 import { PeerTubeServer } from './server'
6 async function waitJobs (serversArg: PeerTubeServer[] | PeerTubeServer, skipDelayed = false) {
7 const pendingJobWait = process.env.NODE_PENDING_JOB_WAIT
8 ? parseInt(process.env.NODE_PENDING_JOB_WAIT, 10)
11 let servers: PeerTubeServer[]
13 if (Array.isArray(serversArg) === false) servers = [ serversArg as PeerTubeServer ]
14 else servers = serversArg as PeerTubeServer[]
16 const states: JobState[] = [ 'waiting', 'active' ]
17 if (!skipDelayed) states.push('delayed')
19 const repeatableJobs: JobType[] = [ 'videos-views-stats', 'activitypub-cleaner' ]
20 let pendingRequests: boolean
22 function tasksBuilder () {
23 const tasks: Promise<any>[] = []
25 // Check if each server has pending request
26 for (const server of servers) {
27 for (const state of states) {
28 const p = server.jobs.list({
33 }).then(body => body.data)
34 .then(jobs => jobs.filter(j => !repeatableJobs.includes(j.type)))
36 if (jobs.length !== 0) {
37 pendingRequests = true
44 const p = server.debug.getDebug()
46 if (obj.activityPubMessagesWaiting !== 0) {
47 pendingRequests = true
58 pendingRequests = false
59 await Promise.all(tasksBuilder())
61 // Retry, in case of new jobs were created
62 if (pendingRequests === false) {
63 await wait(pendingJobWait)
64 await Promise.all(tasksBuilder())
67 if (pendingRequests) {
68 await wait(pendingJobWait)
70 } while (pendingRequests)
73 // ---------------------------------------------------------------------------