From 6fcd19ba737f1f5614a56c6925adb882dea43b8d Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Wed, 5 Jul 2017 13:26:25 +0200 Subject: Move to promises Closes https://github.com/Chocobozzz/PeerTube/issues/74 --- server/lib/jobs/handlers/index.ts | 8 +- server/lib/jobs/handlers/video-transcoder.ts | 22 ++--- server/lib/jobs/job-scheduler.ts | 117 +++++++++++++-------------- 3 files changed, 66 insertions(+), 81 deletions(-) (limited to 'server/lib/jobs') 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 { - 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 } = { 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, jobs: JobInstance[]) { - if (err) { - logger.error('Cannot list pending jobs.', { error: err }) - } else { - jobs.forEach(job => { - jobsQueue.push(job) - }) - } + private enqueueJobs (jobsQueue: AsyncQueue, 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, job: JobInstance, jobResult: any, callback: (err: Error) => void) { + private onJobError (jobHandler: JobHandler, 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, job: JobInstance, jobResult: any, callback: (err: Error) => void) { + private onJobSuccess (jobHandler: JobHandler, 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) } } -- cgit v1.2.3