Job processing microservices using bull

Question

I would like to process scheduled jobs using node.js bull. Basically I have two processors that handle 2 types of jobs. There is one configurator that configures the jobs which will be added to the bull queue using cron.

The scheduler will be in one microservice and the each of the processor will be a separate microservice. So I will be having 3 micro services.

My question is am I using the correct pattern with bull?

index.js

const Queue = require('bull');

const fetchQueue = new Queue('MyScheduler');
fetchQueue.add("fetcher", {name: "earthQuakeAlert"}, {repeat: {cron: '1-59/2 * * * *'}, removeOnComplete: true});
fetchQueue.add("fetcher", {name: "weatherAlert"}, {repeat: {cron: '3-59/3 * * * *'}, removeOnComplete: true});

processor-configurator.js

const Queue=require('bull');

const scheduler = new Queue("MyScheduler");
scheduler.process("processor", __dirname + "/alert-processor");

fetcher-configurator.js

const Queue=require('bull');

const scheduler = new Queue("MyScheduler");
scheduler.process("fetcher", __dirname+"/fetcher");

fetcher.js

const Queue = require('bull');
const moment = require('moment');

module.exports = function (job) {
    const scheduler = new Queue('MyScheduler');
    console.log("Insider processor ", job.data, moment().format("YYYY-MM-DD hh:mm:ss"));
    scheduler.add('processor', {'name': 'Email needs to be sent'}, {removeOnComplete: true});
    return Promise.resolve()
};

alert-processor.js

const Queue = require('bull');
const moment = require('moment');

module.exports = function (job) {
    const scheduler = new Queue('MyScheduler');
    console.log("Insider processor ", job.data, moment().format("YYYY-MM-DD hh:mm:ss"));
    scheduler.add('processor', {'name': 'Email needs to be sent'}, {removeOnComplete: true});
    return Promise.resolve()
};

There will be three microservices -

1. node index.js
2. node fetcher-configurator.js
3. node processor-configurator.js

I see inconsistent behavior from bull. Sometimes I am getting the error Missing process handler for job type

Answer 1

Quoting myself with a hope this will be helpful for someone else

This is because both workers use the same queue. Worker tries to get next job from queue, receives a job with wrong type (eg "fetcher" instead of "processor") and fails because it knows how to handle "processor" and doesn't know what to do with "fetcher". Bull doesn't allow you to take only compatible jobs from queue, both workers should be able to process all types of jobs. The simplest solution would be to use two different queues, one for processors and one for fetchers. Then you can remove names from jobs and processors, it won't be needed anymore since name is defined by the queue.

https://github.com/OptimalBits/bull/issues/1481

Answer 2

The Bull:

expiration-queue.js

import Queue from 'bull';
import { ExpirationCompletePublisher } from '../events/publishers/expiration-complete-publisher';
import { natsWrapper } from '../nats-wrapper';
interface Payload {
  orderId: string;
}

const expirationQueue = new Queue<Payload>('order:expiration', {
  redis: {
    host: process.env.REDIS_HOST, 
  },
});

expirationQueue.process(async (job) => {
  console.log('Expiries order id', job.data.orderId);
  new ExpirationCompletePublisher(natsWrapper.client).publish({
    orderId: job.data.orderId,
  });
});

export { expirationQueue };

promotionEndQueue.js

import Queue from 'bull';
import { PromotionEndedPublisher } from '../events/publishers/promotion-ended-publisher';
import { natsWrapper } from '../nats-wrapper';
interface Payload {
  promotionId: string;
}

const promotionEndQueue = new Queue<Payload>('promotions:end', {
  redis: {
    host: process.env.REDIS_HOST, // look at expiration-depl.yaml
  },
});

promotionEndQueue.process(async (job) => {
  console.log('Expiries promotion id', job.data.promotionId);
  new PromotionEndedPublisher(natsWrapper.client).publish({
    promotionId: job.data.promotionId,
  });
});

export { promotionEndQueue };

order-created-listener.js

import { Listener, OrderCreatedEvent, Subjects } from '@your-lib/common';
import { queueGroupName } from './queue-group-name';
import { Message } from 'node-nats-streaming';
import { expirationQueue } from '../../queues/expiration-queue';
export class OrderCreatedListener extends Listener<OrderCreatedEvent> {
  subject: Subjects.OrderCreated = Subjects.OrderCreated;
  queueGroupName = queueGroupName;

  async onMessage(data: OrderCreatedEvent['data'], msg: Message) {
    // delay = expiredTime - currentTime
    const delay = new Date(data.expiresAt).getTime() - new Date().getTime();
    // console.log("delay", delay)
    await expirationQueue.add(
      {
        orderId: data.id,
      },
      {
        delay,
      }
    );

    msg.ack();
  }
}

promotion-started-listener.js

import {
  Listener,
  PromotionStartedEvent,
  Subjects,
} from '@your-lib/common';
import { queueGroupName } from './queue-group-name';
import { Message } from 'node-nats-streaming';
import { promotionEndQueue } from '../../queues/promotions-end-queue';
export class PromotionStartedListener extends Listener<PromotionStartedEvent> {
  subject: Subjects.PromotionStarted = Subjects.PromotionStarted;
  queueGroupName = queueGroupName;

  async onMessage(data: PromotionStartedEvent['data'], msg: Message) {
    // delay = expiredTime - currentTime
    const delay = new Date(data.endTime).getTime() - new Date().getTime();

    // console.log("delay", delay)
    await promotionEndQueue.add(
      {
        promotionId: data.id,
      },
      {
        delay,
      }
    );

    msg.ack();
  }
}