Move to promises

Closes https://github.com/Chocobozzz/PeerTube/issues/74

3 files changed, 66 insertions, 81 deletions

diff --git a/server/lib/jobs/handlers/index.ts b/server/lib/jobs/handlers/index.ts
index 7d0263b15..8abddae35 100644
--- a/server/lib/jobs/handlers/index.ts
+++ b/server/lib/jobs/handlers/index.ts
@@ -1,11 +1,9 @@
 import * as videoTranscoder from './video-transcoder'
 
-import { VideoInstance } from '../../../models'
-
 export interface JobHandler<T> {
-  process (data: object, callback: (err: Error, videoInstance?: T) => void)
-  onError (err: Error, jobId: number, video: T, callback: (err: Error) => void)
-  onSuccess (data: any, jobId: number, video: T, callback: (err: Error) => void)
+  process (data: object): T
+  onError (err: Error, jobId: number)
+  onSuccess (jobId: number, jobResult: T)
 }
 
 const jobHandlers: { [ handlerName: string ]: JobHandler<any> } = {
diff --git a/server/lib/jobs/handlers/video-transcoder.ts b/server/lib/jobs/handlers/video-transcoder.ts
index 6f606a7d3..e829ca813 100644
--- a/server/lib/jobs/handlers/video-transcoder.ts
+++ b/server/lib/jobs/handlers/video-transcoder.ts
@@ -3,29 +3,23 @@ import { logger } from '../../../helpers'
 import { addVideoToFriends } from '../../../lib'
 import { VideoInstance } from '../../../models'
 
-function process (data: { id: string }, callback: (err: Error, videoInstance?: VideoInstance) => void) {
-  db.Video.loadAndPopulateAuthorAndPodAndTags(data.id, function (err, video) {
-    if (err) return callback(err)
-
-    video.transcodeVideofile(function (err) {
-      return callback(err, video)
-    })
+function process (data: { id: string }) {
+  return db.Video.loadAndPopulateAuthorAndPodAndTags(data.id).then(video => {
+    return video.transcodeVideofile().then(() => video)
   })
 }
 
-function onError (err: Error, jobId: number, video: VideoInstance, callback: (err: Error) => void) {
+function onError (err: Error, jobId: number) {
   logger.error('Error when transcoding video file in job %d.', jobId, { error: err })
-  return callback(null)
+  return Promise.resolve()
 }
 
-function onSuccess (data: any, jobId: number, video: VideoInstance, callback: (err: Error) => void) {
+function onSuccess (jobId: number, video: VideoInstance) {
   logger.info('Job %d is a success.', jobId)
 
-  video.toAddRemoteJSON(function (err, remoteVideo) {
-    if (err) return callback(err)
-
+  video.toAddRemoteJSON().then(remoteVideo => {
     // Now we'll add the video's meta data to our friends
-    addVideoToFriends(remoteVideo, null, callback)
+    return addVideoToFriends(remoteVideo, null)
   })
 }
 
diff --git a/server/lib/jobs/job-scheduler.ts b/server/lib/jobs/job-scheduler.ts
index 2f01387e7..248dc7978 100644
--- a/server/lib/jobs/job-scheduler.ts
+++ b/server/lib/jobs/job-scheduler.ts
@@ -32,37 +32,35 @@ class JobScheduler {
 
     // Finish processing jobs from a previous start
     const state = JOB_STATES.PROCESSING
-    db.Job.listWithLimit(limit, state, (err, jobs) => {
-      this.enqueueJobs(err, 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, (err, jobs) => {
-            if (err) {
-              logger.error('Cannot list pending jobs.', { error: err })
-            } else {
-              jobs.forEach(job => {
-                jobsQueue.push(job)
-              })
+    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)
             }
 
-            // Optimization: we could use "drain" from queue object
-            return setTimeout(next, JOBS_FETCHING_INTERVAL)
-          })
-        },
+            const state = JOB_STATES.PENDING
+            db.Job.listWithLimit(limit, state)
+              .then(jobs => {
+                this.enqueueJobs(jobsQueue, jobs)
 
-        err => { logger.error('Error in job scheduler queue.', { error: err }) }
-      )
-    })
+                // Optimization: we could use "drain" from queue object
+                return setTimeout(next, JOBS_FETCHING_INTERVAL)
+              })
+              .catch(err => logger.error('Cannot list pending jobs.', { error: err }))
+          },
+
+          err => logger.error('Error in job scheduler queue.', { error: err })
+        )
+      })
+      .catch(err => logger.error('Cannot list pending jobs.', { error: err }))
   }
 
-  createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object, callback: (err: Error) => void) {
+  createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object) {
     const createQuery = {
       state: JOB_STATES.PENDING,
       handlerName,
@@ -70,67 +68,62 @@ class JobScheduler {
     }
     const options = { transaction }
 
-    db.Job.create(createQuery, options).asCallback(callback)
+    return db.Job.create(createQuery, options)
   }
 
-  private enqueueJobs (err: Error, jobsQueue: AsyncQueue<JobInstance>, jobs: JobInstance[]) {
-    if (err) {
-      logger.error('Cannot list pending jobs.', { error: err })
-    } else {
-      jobs.forEach(job => {
-        jobsQueue.push(job)
-      })
-    }
+  private enqueueJobs (jobsQueue: AsyncQueue<JobInstance>, jobs: JobInstance[]) {
+    jobs.forEach(job => jobsQueue.push(job))
   }
 
   private 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)
+      return callback(null)
+    }
 
     logger.info('Processing job %d with handler %s.', job.id, job.handlerName)
 
     job.state = JOB_STATES.PROCESSING
-    job.save().asCallback(err => {
-      if (err) return this.cannotSaveJobError(err, callback)
-
-      if (jobHandler === undefined) {
-        logger.error('Unknown job handler for job %s.', job.handlerName)
-        return callback(null)
-      }
+    return job.save()
+      .then(() => {
+        return jobHandler.process(job.handlerInputData)
+      })
+      .then(
+        result => {
+          return this.onJobSuccess(jobHandler, job, result)
+        },
 
-      return jobHandler.process(job.handlerInputData, (err, result) => {
-        if (err) {
+        err => {
           logger.error('Error in job handler %s.', job.handlerName, { error: err })
-          return this.onJobError(jobHandler, job, result, callback)
+          return this.onJobError(jobHandler, job, err)
         }
-
-        return this.onJobSuccess(jobHandler, job, result, callback)
+      )
+      .then(() => callback(null))
+      .catch(err => {
+        this.cannotSaveJobError(err)
+        return callback(err)
       })
-    })
   }
 
-  private onJobError (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any, callback: (err: Error) => void) {
+  private onJobError (jobHandler: JobHandler<any>, job: JobInstance, err: Error) {
     job.state = JOB_STATES.ERROR
 
-    job.save().asCallback(err => {
-      if (err) return this.cannotSaveJobError(err, callback)
-
-      return jobHandler.onError(err, job.id, jobResult, callback)
-    })
+    return job.save()
+      .then(() => jobHandler.onError(err, job.id))
+      .catch(err => this.cannotSaveJobError(err))
   }
 
-  private onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any, callback: (err: Error) => void) {
+  private onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any) {
     job.state = JOB_STATES.SUCCESS
 
-    job.save().asCallback(err => {
-      if (err) return this.cannotSaveJobError(err, callback)
-
-      return jobHandler.onSuccess(err, job.id, jobResult, callback)
-    })
+    return job.save()
+      .then(() => jobHandler.onSuccess(job.id, jobResult))
+      .catch(err => this.cannotSaveJobError(err))
   }
 
-  private cannotSaveJobError (err: Error, callback: (err: Error) => void) {
+  private cannotSaveJobError (err: Error) {
     logger.error('Cannot save new job state.', { error: err })
-    return callback(err)
   }
 }