Use async/await in lib and initializers

3 files changed, 115 insertions, 112 deletions

diff --git a/server/lib/jobs/handlers/video-file-optimizer.ts b/server/lib/jobs/handlers/video-file-optimizer.ts
index 63a51064c..799ba8b01 100644
--- a/server/lib/jobs/handlers/video-file-optimizer.ts
+++ b/server/lib/jobs/handlers/video-file-optimizer.ts
@@ -1,4 +1,4 @@
-import * as Promise from 'bluebird'
+import * as Bluebird from 'bluebird'
 
 import { database as db } from '../../../initializers/database'
 import { logger, computeResolutionsToTranscode } from '../../../helpers'
@@ -6,16 +6,17 @@ import { VideoInstance } from '../../../models'
 import { addVideoToFriends } from '../../friends'
 import { JobScheduler } from '../job-scheduler'
 
-function process (data: { videoUUID: string }, jobId: number) {
-  return db.Video.loadByUUIDAndPopulateAuthorAndPodAndTags(data.videoUUID).then(video => {
-    // No video, maybe deleted?
-    if (!video) {
-      logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid })
-      return undefined
-    }
+async function process (data: { videoUUID: string }, jobId: number) {
+  const video = await db.Video.loadByUUIDAndPopulateAuthorAndPodAndTags(data.videoUUID)
+  // No video, maybe deleted?
+  if (!video) {
+    logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid })
+    return undefined
+  }
+
+  await video.optimizeOriginalVideofile()
 
-    return video.optimizeOriginalVideofile().then(() => video)
-  })
+  return video
 }
 
 function onError (err: Error, jobId: number) {
@@ -23,33 +24,31 @@ function onError (err: Error, jobId: number) {
   return Promise.resolve()
 }
 
-function onSuccess (jobId: number, video: VideoInstance) {
+async function onSuccess (jobId: number, video: VideoInstance) {
   if (video === undefined) return undefined
 
   logger.info('Job %d is a success.', jobId)
 
-  video.toAddRemoteJSON()
-    .then(remoteVideo => {
-      // Now we'll add the video's meta data to our friends
-      return addVideoToFriends(remoteVideo, null)
-    })
-    .then(() => {
-      return video.getOriginalFileHeight()
-    })
-    .then(originalFileHeight => {
-      // Create transcoding jobs if there are enabled resolutions
-      const resolutionsEnabled = computeResolutionsToTranscode(originalFileHeight)
-      logger.info(
-        'Resolutions computed for video %s and origin file height of %d.', video.uuid, originalFileHeight,
-        { resolutions: resolutionsEnabled }
-      )
-
-      if (resolutionsEnabled.length === 0) return undefined
-
-      return db.sequelize.transaction(t => {
-        const tasks: Promise<any>[] = []
-
-        resolutionsEnabled.forEach(resolution => {
+  const remoteVideo = await video.toAddRemoteJSON()
+
+  // Now we'll add the video's meta data to our friends
+  await addVideoToFriends(remoteVideo, null)
+
+  const originalFileHeight = await video.getOriginalFileHeight()
+  // Create transcoding jobs if there are enabled resolutions
+
+  const resolutionsEnabled = computeResolutionsToTranscode(originalFileHeight)
+  logger.info(
+    'Resolutions computed for video %s and origin file height of %d.', video.uuid, originalFileHeight,
+    { resolutions: resolutionsEnabled }
+  )
+
+  if (resolutionsEnabled.length !== 0) {
+    try {
+      await db.sequelize.transaction(async t => {
+        const tasks: Bluebird<any>[] = []
+
+        for (const resolution of resolutionsEnabled) {
           const dataInput = {
             videoUUID: video.uuid,
             resolution
@@ -57,24 +56,19 @@ function onSuccess (jobId: number, video: VideoInstance) {
 
           const p = JobScheduler.Instance.createJob(t, 'videoFileTranscoder', dataInput)
           tasks.push(p)
-        })
+        }
 
-        return Promise.all(tasks).then(() => resolutionsEnabled)
+        await Promise.all(tasks)
       })
-    })
-    .then(resolutionsEnabled => {
-      if (resolutionsEnabled === undefined) {
-        logger.info('No transcoding jobs created for video %s (no resolutions enabled).')
-        return undefined
-      }
 
       logger.info('Transcoding jobs created for uuid %s.', video.uuid, { resolutionsEnabled })
-    })
-    .catch((err: Error) => {
-      logger.debug('Cannot transcode the video.', err)
-      throw err
-    })
-
+    } catch (err) {
+      logger.warn('Cannot transcode the video.', err)
+    }
+  } else {
+    logger.info('No transcoding jobs created for video %s (no resolutions enabled).')
+    return undefined
+  }
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/jobs/handlers/video-file-transcoder.ts b/server/lib/jobs/handlers/video-file-transcoder.ts
index 0dafee566..b240ff58a 100644
--- a/server/lib/jobs/handlers/video-file-transcoder.ts
+++ b/server/lib/jobs/handlers/video-file-transcoder.ts
@@ -4,16 +4,17 @@ import { logger } from '../../../helpers'
 import { VideoInstance } from '../../../models'
 import { VideoResolution } from '../../../../shared'
 
-function process (data: { videoUUID: string, resolution: VideoResolution }, jobId: number) {
-  return db.Video.loadByUUIDAndPopulateAuthorAndPodAndTags(data.videoUUID).then(video => {
-    // No video, maybe deleted?
-    if (!video) {
-      logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid })
-      return undefined
-    }
-
-    return video.transcodeOriginalVideofile(data.resolution).then(() => video)
-  })
+async function process (data: { videoUUID: string, resolution: VideoResolution }, jobId: number) {
+  const video = await db.Video.loadByUUIDAndPopulateAuthorAndPodAndTags(data.videoUUID)
+  // No video, maybe deleted?
+  if (!video) {
+    logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid })
+    return undefined
+  }
+
+  await video.transcodeOriginalVideofile(data.resolution)
+
+  return video
 }
 
 function onError (err: Error, jobId: number) {
diff --git a/server/lib/jobs/job-scheduler.ts b/server/lib/jobs/job-scheduler.ts
index c2409d20c..61d483268 100644
--- a/server/lib/jobs/job-scheduler.ts
+++ b/server/lib/jobs/job-scheduler.ts
@@ -23,7 +23,7 @@ class JobScheduler {
     return this.instance || (this.instance = new this())
   }
 
-  activate () {
+  async activate () {
     const limit = JOBS_FETCH_LIMIT_PER_CYCLE
 
     logger.info('Jobs scheduler activated.')
@@ -32,32 +32,36 @@ class JobScheduler {
 
     // Finish processing jobs from a previous start
     const state = JOB_STATES.PROCESSING
-    db.Job.listWithLimit(limit, state)
-      .then(jobs => {
-        this.enqueueJobs(jobsQueue, jobs)
-
-        forever(
-          next => {
-            if (jobsQueue.length() !== 0) {
-              // Finish processing the queue first
-              return setTimeout(next, JOBS_FETCHING_INTERVAL)
-            }
-
-            const state = JOB_STATES.PENDING
-            db.Job.listWithLimit(limit, state)
-              .then(jobs => {
-                this.enqueueJobs(jobsQueue, jobs)
-
-                // Optimization: we could use "drain" from queue object
-                return setTimeout(next, JOBS_FETCHING_INTERVAL)
-              })
-              .catch(err => logger.error('Cannot list pending jobs.', err))
-          },
-
-          err => logger.error('Error in job scheduler queue.', err)
-        )
-      })
-      .catch(err => logger.error('Cannot list pending jobs.', err))
+    try {
+      const jobs = await db.Job.listWithLimit(limit, state)
+
+      this.enqueueJobs(jobsQueue, jobs)
+    } catch (err) {
+      logger.error('Cannot list pending jobs.', err)
+    }
+
+    forever(
+      async next => {
+        if (jobsQueue.length() !== 0) {
+          // Finish processing the queue first
+          return setTimeout(next, JOBS_FETCHING_INTERVAL)
+        }
+
+        const state = JOB_STATES.PENDING
+        try {
+          const jobs = await db.Job.listWithLimit(limit, state)
+
+          this.enqueueJobs(jobsQueue, jobs)
+        } catch (err) {
+          logger.error('Cannot list pending jobs.', err)
+        }
+
+        // Optimization: we could use "drain" from queue object
+        return setTimeout(next, JOBS_FETCHING_INTERVAL)
+      },
+
+      err => logger.error('Error in job scheduler queue.', err)
+    )
   }
 
   createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object) {
@@ -75,7 +79,7 @@ class JobScheduler {
     jobs.forEach(job => jobsQueue.push(job))
   }
 
-  private processJob (job: JobInstance, callback: (err: Error) => void) {
+  private async processJob (job: JobInstance, callback: (err: Error) => void) {
     const jobHandler = jobHandlers[job.handlerName]
     if (jobHandler === undefined) {
       logger.error('Unknown job handler for job %s.', job.handlerName)
@@ -85,41 +89,45 @@ class JobScheduler {
     logger.info('Processing job %d with handler %s.', job.id, job.handlerName)
 
     job.state = JOB_STATES.PROCESSING
-    return job.save()
-      .then(() => {
-        return jobHandler.process(job.handlerInputData, job.id)
-      })
-      .then(
-        result => {
-          return this.onJobSuccess(jobHandler, job, result)
-        },
-
-        err => {
-          logger.error('Error in job handler %s.', job.handlerName, err)
-          return this.onJobError(jobHandler, job, err)
-        }
-      )
-      .then(() => callback(null))
-      .catch(err => {
-        this.cannotSaveJobError(err)
-        return callback(err)
-      })
+    await job.save()
+
+    try {
+      const result = await jobHandler.process(job.handlerInputData, job.id)
+      await this.onJobSuccess(jobHandler, job, result)
+    } catch (err) {
+      logger.error('Error in job handler %s.', job.handlerName, err)
+
+      try {
+        await this.onJobError(jobHandler, job, err)
+      } catch (innerErr) {
+        this.cannotSaveJobError(innerErr)
+        return callback(innerErr)
+      }
+    }
+
+    callback(null)
   }
 
-  private onJobError (jobHandler: JobHandler<any>, job: JobInstance, err: Error) {
+  private async onJobError (jobHandler: JobHandler<any>, job: JobInstance, err: Error) {
     job.state = JOB_STATES.ERROR
 
-    return job.save()
-      .then(() => jobHandler.onError(err, job.id))
-      .catch(err => this.cannotSaveJobError(err))
+    try {
+      await job.save()
+      await jobHandler.onError(err, job.id)
+    } catch (err) {
+      this.cannotSaveJobError(err)
+    }
   }
 
-  private onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any) {
+  private async onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any) {
     job.state = JOB_STATES.SUCCESS
 
-    return job.save()
-      .then(() => jobHandler.onSuccess(job.id, jobResult))
-      .catch(err => this.cannotSaveJobError(err))
+    try {
+      await job.save()
+      jobHandler.onSuccess(job.id, jobResult)
+    } catch (err) {
+      this.cannotSaveJobError(err)
+    }
   }
 
   private cannotSaveJobError (err: Error) {