NodeJS: In Memory Worker

index.js

const worker = require('./worker')

async function main() {
  let count = 0

  const interval = setInterval(() => {
    worker.enqueue('jobExample', {
      message: `Hello from worker ${++count}!`
    })

    console.log(`Enqueued: ${count}`)

    if (count >= 5) {
      clearInterval(interval)
    }
  }, 1000)
}

main()

job-example.js

exports.run = async data => {
  await new Promise((resolve, reject) => setTimeout(resolve, 2000))
  console.log(data)
}

worker.js

const { EventEmitter } = require('events')

const event = new EventEmitter()
const queue = []
const maxRetries = 3

let running = false
let currentJobId = 0

const jobs = {
  jobExample: require('./jobs/job-example')
}

event.on('run', async (jobId, jobName, data, retries) => {
  try {
    await jobs[jobName].run(data)
    event.emit('next')
  } catch (err) {
    console.error(`[${jobId}]: Job "${jobName}" failed to execute with reason: ${err?.message}`)

    if (retries < maxRetries) {
      event.emit('run', jobId, jobName, data, retries + 1)
    } else {
      event.emit('next')
    }
  }
})

event.on('next', () => {
  running = true
  const nextJob = queue.shift()

  if (nextJob) {
    const [jobId, jobName, data] = nextJob
    event.emit('run', jobId, jobName, data, 1)
  } else {
    running = false
  }
})

exports.enqueue = (jobName, data) => {
  queue.push([++currentJobId, jobName, data])

  if (!running) {
    event.emit('next')
  }
}