Fix live replay duration glitch

[github/Chocobozzz/PeerTube.git] / server / lib / job-queue / job-queue.ts

diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index d8d64caaf348bae8e464cfb58bbd1d53f26e46dd..8d97434ac59b954d33510dd32780a051516ec955 100644 (file)
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -2,9 +2,17 @@ import * as Bull from 'bull'
 import {
   ActivitypubFollowPayload,
   ActivitypubHttpBroadcastPayload,
-  ActivitypubHttpFetcherPayload, ActivitypubHttpUnicastPayload, EmailPayload,
+  ActivitypubHttpFetcherPayload,
+  ActivitypubHttpUnicastPayload,
+  EmailPayload,
   JobState,
-  JobType, RefreshPayload, VideoFileImportPayload, VideoImportPayload, VideoRedundancyPayload, VideoTranscodingPayload
+  JobType,
+  RefreshPayload,
+  VideoFileImportPayload,
+  VideoImportPayload,
+  VideoLiveEndingPayload,
+  VideoRedundancyPayload,
+  VideoTranscodingPayload
 } from '../../../shared/models'
 import { logger } from '../../helpers/logger'
 import { Redis } from '../redis'
@@ -13,13 +21,14 @@ import { processActivityPubHttpBroadcast } from './handlers/activitypub-http-bro
 import { processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
 import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
 import { processEmail } from './handlers/email'
-import { processVideoTranscoding} from './handlers/video-transcoding'
+import { processVideoTranscoding } from './handlers/video-transcoding'
 import { processActivityPubFollow } from './handlers/activitypub-follow'
-import { processVideoImport} from './handlers/video-import'
+import { processVideoImport } from './handlers/video-import'
 import { processVideosViews } from './handlers/video-views'
-import { refreshAPObject} from './handlers/activitypub-refresher'
-import { processVideoFileImport} from './handlers/video-file-import'
-import { processVideoRedundancy} from '@server/lib/job-queue/handlers/video-redundancy'
+import { refreshAPObject } from './handlers/activitypub-refresher'
+import { processVideoFileImport } from './handlers/video-file-import'
+import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy'
+import { processVideoLiveEnding } from './handlers/video-live-ending'
 
 type CreateJobArgument =
   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
@@ -32,8 +41,13 @@ type CreateJobArgument =
   { type: 'video-import', payload: VideoImportPayload } |
   { type: 'activitypub-refresher', payload: RefreshPayload } |
   { type: 'videos-views', payload: {} } |
+  { type: 'video-live-ending', payload: VideoLiveEndingPayload } |
   { type: 'video-redundancy', payload: VideoRedundancyPayload }
 
+type CreateJobOptions = {
+  delay?: number
+}
+
 const handlers: { [id in JobType]: (job: Bull.Job) => Promise<any> } = {
   'activitypub-http-broadcast': processActivityPubHttpBroadcast,
   'activitypub-http-unicast': processActivityPubHttpUnicast,
@@ -45,6 +59,7 @@ const handlers: { [id in JobType]: (job: Bull.Job) => Promise<any> } = {
   'video-import': processVideoImport,
   'videos-views': processVideosViews,
   'activitypub-refresher': refreshAPObject,
+  'video-live-ending': processVideoLiveEnding,
   'video-redundancy': processVideoRedundancy
 }
 
@@ -59,7 +74,8 @@ const jobTypes: JobType[] = [
   'video-import',
   'videos-views',
   'activitypub-refresher',
-  'video-redundancy'
+  'video-redundancy',
+  'video-live-ending'
 ]
 
 class JobQueue {
@@ -115,12 +131,12 @@ class JobQueue {
     }
   }
 
-  createJob (obj: CreateJobArgument): void {
-    this.createJobWithPromise(obj)
-         .catch(err => logger.error('Cannot create job.', { err, obj }))
+  createJob (obj: CreateJobArgument, options: CreateJobOptions = {}): void {
+    this.createJobWithPromise(obj, options)
+        .catch(err => logger.error('Cannot create job.', { err, obj }))
   }
 
-  createJobWithPromise (obj: CreateJobArgument) {
+  createJobWithPromise (obj: CreateJobArgument, options: CreateJobOptions = {}) {
     const queue = this.queues[obj.type]
     if (queue === undefined) {
       logger.error('Unknown queue %s: cannot create job.', obj.type)
@@ -130,7 +146,8 @@ class JobQueue {
     const jobArgs: Bull.JobOptions = {
       backoff: { delay: 60 * 1000, type: 'exponential' },
       attempts: JOB_ATTEMPTS[obj.type],
-      timeout: JOB_TTL[obj.type]
+      timeout: JOB_TTL[obj.type],
+      delay: options.delay
     }
 
     return queue.add(obj.payload, jobArgs)