Multiconsumer-bull NPM

MultiConsumer Bull

A multi-consumer queue implementation using Bull queue.

Installation

npm install --save multiconsumer-bull

Usage examples

import * as Bull from "bull"
import * as redis from "redis"
import {MultiConsumerBull} from "multiconsumer-bull"

const bus = MultiConsumerBull(new Bull("redis://127.0.0.1:6379"), () => redis.createClient())

// Process "my-topic" for logging
bus.topic("my-topic").process("log", (job, cb) => {
  console.log("got new job in topic \"my-topic\" with data", job.data)
  cb()
})

// Save all "my-topic" messages to database
bus.topic("my-topic").process("save", (job, cb) => {
  console.log("here we're going to save all messages from \"my-topic\" to database", job.data)
  cb()
})

bus.topic("my-topic").add("Hello World!")

NOTE: Wrapper implementation is not removing consumer groups from RedisLiveSet so once you're not interested anymore for processing topic messages for specific groupId - you must remove that group and tasks manually

Group can be removed using MultiConsumerQueueImpl.removeGroup() method:

const bus = MultiConsumerBull(...)

// deploy this to your servers to stop collecting tasks
bus.topic("my-topic").removeGroup("old-process-group")

You will still have to manually remove tasks already added for that group, or maybe those may expire, this depends on how source NamedQueue is implemented.