import { processVideoStudioEdition } from './handlers/video-studio-edition'
import { processVideoTranscoding } from './handlers/video-transcoding'
import { processVideosViewsStats } from './handlers/video-views-stats'
+import { Redis } from '../redis'
export type CreateJobArgument =
{ type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
private constructor () {
}
- init (produceOnly = false) {
+ init () {
// Already initialized
if (this.initialized === true) return
this.initialized = true
this.jobRedisPrefix = 'bull-' + WEBSERVER.HOST
for (const handlerName of (Object.keys(handlers) as JobType[])) {
- this.buildWorker(handlerName, produceOnly)
+ this.buildWorker(handlerName)
this.buildQueue(handlerName)
- this.buildQueueScheduler(handlerName, produceOnly)
- this.buildQueueEvent(handlerName, produceOnly)
+ this.buildQueueScheduler(handlerName)
+ this.buildQueueEvent(handlerName)
}
this.flowProducer = new FlowProducer({
- connection: this.getRedisConnection(),
+ connection: Redis.getRedisClientOptions('FlowProducer'),
prefix: this.jobRedisPrefix
})
this.flowProducer.on('error', err => { logger.error('Error in flow producer', { err }) })
this.addRepeatableJobs()
}
- private buildWorker (handlerName: JobType, produceOnly: boolean) {
+ private buildWorker (handlerName: JobType) {
const workerOptions: WorkerOptions = {
- autorun: !produceOnly,
+ autorun: false,
concurrency: this.getJobConcurrency(handlerName),
prefix: this.jobRedisPrefix,
- connection: this.getRedisConnection()
+ connection: Redis.getRedisClientOptions('Worker')
}
const handler = function (job: Job) {
private buildQueue (handlerName: JobType) {
const queueOptions: QueueOptions = {
- connection: this.getRedisConnection(),
+ connection: Redis.getRedisClientOptions('Queue'),
prefix: this.jobRedisPrefix
}
this.queues[handlerName] = queue
}
- private buildQueueScheduler (handlerName: JobType, produceOnly: boolean) {
+ private buildQueueScheduler (handlerName: JobType) {
const queueSchedulerOptions: QueueSchedulerOptions = {
- autorun: !produceOnly,
- connection: this.getRedisConnection(),
+ autorun: false,
+ connection: Redis.getRedisClientOptions('QueueScheduler'),
prefix: this.jobRedisPrefix,
maxStalledCount: 10
}
this.queueSchedulers[handlerName] = queueScheduler
}
- private buildQueueEvent (handlerName: JobType, produceOnly: boolean) {
+ private buildQueueEvent (handlerName: JobType) {
const queueEventsOptions: QueueEventsOptions = {
- autorun: !produceOnly,
- connection: this.getRedisConnection(),
+ autorun: false,
+ connection: Redis.getRedisClientOptions('QueueEvent'),
prefix: this.jobRedisPrefix
}
this.queueEvents[handlerName] = queueEvents
}
- private getRedisConnection () {
- return {
- password: CONFIG.REDIS.AUTH,
- db: CONFIG.REDIS.DB,
- host: CONFIG.REDIS.HOSTNAME,
- port: CONFIG.REDIS.PORT,
- path: CONFIG.REDIS.SOCKET
- }
- }
-
// ---------------------------------------------------------------------------
async terminate () {
return Promise.all(promises)
}
+ start () {
+ const promises = Object.keys(this.workers)
+ .map(handlerName => {
+ const worker: Worker = this.workers[handlerName]
+ const queueScheduler: QueueScheduler = this.queueSchedulers[handlerName]
+ const queueEvent: QueueEvents = this.queueEvents[handlerName]
+
+ return Promise.all([
+ worker.run(),
+ queueScheduler.run(),
+ queueEvent.run()
+ ])
+ })
+
+ return Promise.all(promises)
+ }
+
async pause () {
for (const handlerName of Object.keys(this.workers)) {
const worker: Worker = this.workers[handlerName]