shared/extra-utils/server/jobs.ts

   1 import * as request from 'supertest'
   2 import { Job, JobState } from '../../models'
   3 import { wait } from '../miscs/miscs'
   4 import { ServerInfo } from './servers'
   5 import { inspect } from 'util'
   6
   7 function getJobsList (url: string, accessToken: string, state: JobState) {
   8   const path = '/api/v1/jobs/' + state
   9
  10   return request(url)
  11           .get(path)
  12           .set('Accept', 'application/json')
  13           .set('Authorization', 'Bearer ' + accessToken)
  14           .expect(200)
  15           .expect('Content-Type', /json/)
  16 }
  17
  18 function getJobsListPaginationAndSort (url: string, accessToken: string, state: JobState, start: number, count: number, sort: string) {
  19   const path = '/api/v1/jobs/' + state
  20
  21   return request(url)
  22           .get(path)
  23           .query({ start })
  24           .query({ count })
  25           .query({ sort })
  26           .set('Accept', 'application/json')
  27           .set('Authorization', 'Bearer ' + accessToken)
  28           .expect(200)
  29           .expect('Content-Type', /json/)
  30 }
  31
  32 async function waitJobs (serversArg: ServerInfo[] | ServerInfo) {
  33   const pendingJobWait = process.env.NODE_PENDING_JOB_WAIT ? parseInt(process.env.NODE_PENDING_JOB_WAIT, 10) : 2000
  34   let servers: ServerInfo[]
  35
  36   if (Array.isArray(serversArg) === false) servers = [ serversArg as ServerInfo ]
  37   else servers = serversArg as ServerInfo[]
  38
  39   const states: JobState[] = [ 'waiting', 'active', 'delayed' ]
  40   let pendingRequests = false
  41
  42   function tasksBuilder () {
  43     const tasks: Promise<any>[] = []
  44     pendingRequests = false
  45
  46     // Check if each server has pending request
  47     for (const server of servers) {
  48       for (const state of states) {
  49         const p = getJobsListPaginationAndSort(server.url, server.accessToken, state, 0, 10, '-createdAt')
  50           .then(res => res.body.data)
  51           .then((jobs: Job[]) => jobs.filter(j => j.type !== 'videos-views'))
  52           .then(jobs => {
  53             if (jobs.length !== 0) {
  54               pendingRequests = true
  55             }
  56           })
  57         tasks.push(p)
  58       }
  59     }
  60
  61     return tasks
  62   }
  63
  64   do {
  65     await Promise.all(tasksBuilder())
  66
  67     // Retry, in case of new jobs were created
  68     if (pendingRequests === false) {
  69       await wait(pendingJobWait)
  70       await Promise.all(tasksBuilder())
  71     }
  72
  73     if (pendingRequests) {
  74       await wait(1000)
  75     }
  76   } while (pendingRequests)
  77 }
  78
  79 // ---------------------------------------------------------------------------
  80
  81 export {
  82   getJobsList,
  83   waitJobs,
  84   getJobsListPaginationAndSort
  85 }