Class RedisQueue

src/job-queue/redis-adapter.ts:5–109

import { Queue, QueueOptions } from "./queue"

export class RedisQueue<T> extends Queue<T> {
  private redisBlocking: IORedis
  private redisNonBlocking: IORedis
  private redisSubscriber: IORedis

  constructor(redisUrl: string, queueName: string, options?: QueueOptions) {
    super(queueName, options)

    this.redisBlocking = new IORedis(redisUrl)
    this.redisNonBlocking = new IORedis(redisUrl)
    this.redisSubscriber = new IORedis(redisUrl)

    this.listenForExpiredJobs()
    this.checkForExpiredJobs()
  }

  get length() {
    return this.redisNonBlocking.zcard(this.activeJobsQueue)
  }

  async getOrWaitForJob() {
    const data = await this.redisBlocking.bzpopmin(this.activeJobsQueue, 0)
    return data?.[1] ? JSON.parse(data[1]) : null
  }

  protected async push(job: Job<T>) {
    if (job.options.delay > 0) {
      return await this.redisNonBlocking
        .multi()
        .set(this.expiringJobIdList(job.id), job.id)
        .pexpire(this.expiringJobIdList(job.id), job.options.delay)
        .zadd(this.delayedJobIdsQueue, Date.now() + job.options.delay, job.id)
        .hset(this.delayedJobsList, job.id, JSON.stringify(job))
        .exec()
    } else {
      return await this.addActiveJob(job)
    }
  }

  // Ready to execute jobs sorted by priority
  private get activeJobsQueue() {
    return `${this.queueName}:active`
  }

  // Full job data for delayed jobs
  private get delayedJobsList() {
    return `${this.queueName}:jobs`
  }

  // Delayed job ids sorted by execution date
  private get delayedJobIdsQueue() {
    return `${this.queueName}:delayed`
  }

  // Delayed job ids that will expire after the delay
  private expiringJobIdList(jobId: string) {
    return `${this.queueName}:expiringJobs:${jobId}`
  }