Redis based message queue, offering parrallel processing
npm install @draad/messagequeue
You also need to have redis available. You can get the docker image as follow :
docker pull redis
We connect to redis using settings passed as environment variables.
Default settings are used for local development
process.env.redisSettings = JSON.stringify({
port: 6379,
host: '127.0.0.1',
});
----
----
Usage
Publish a message
A PRODUCER service can emmit a message on a specific CHANNEL using the PUBLISH function.
A channel is composed of multiples related QUEUES, see observer section for more informations.
The message can be in any format.
Channel will be created on the fly and you can create as many as you want.
This library respect the first message in, first message out. This paradigm is respected for each
individual channel, but not between channels.
Publish takes as params :
- {string} channel Name of the channel that will receive the message.
- {any} message Message to emit.
- {object} options Optional metadata.
You can add any custom value to the metadata, but not that the following keys are used for internal logic :
- {uint} ttl Time to live in miliseconds. If ttl is exceeded, the message won't be processed but moved directly
to failed queue. Default null means the message won't expire.
The following keys are reserved by the system and cannot be overrided :
- {uint} createdAt Time in miliseconds at wich the message has been created.
- {uint} jammedAfter Delay before message is considered as jammed after starting to process. Default 180000 ms.
$3
In the following exemple, the PRODUCER service sends a message in the channel to notify that a new
order has been created. This message expire if it's not processed in next 60000 ms.
` js
const messageQueue = require('@draad/messagequeue');
const { publish } = await messageQueue();const message = { event: 'created', user: 'draad', purchase: 'rainbow unicorn' };
await publish('orders', message, { ttl: 60000 });
`
-----
Subscribe to a queue
A CONSUMER service can subscribe to one or multiples CHANNELS. It must gives a CALLBACK function
that will be triggered every time a message is extracted from the queue.
If listened channel is empty, the CALLBACK will be alerted when a new message is created.
If the channel has some messages pending, the CALLBACK will be triggered immediatly.
You have the ability to set a concurrency count, allowing to treat multiples messages in parrallel.
Subscribe takes as params :
- {string} queue Queue to listen.
- {function} callback Callback function.
- {uint} maxConcurrent Concurrent request count. Default 1.
The callback you pass will receive the following arguments :
- {object} data The item extracted from the queue
- {any} data.message The message.
- {object} data.metadata The metadata associated to this message.
- {function} done Done function. You're responsible for calling it when message has been processed correctly.
- {function} fail Fail function. You're responsible for calling it when message has not been processed correctly.
- {function} next Next function, you can call it to request next message from the queue right away.
$3
In the following example, the CONSUMER service listen messages from queue.
` js
const messageQueue = require('@draad/messagequeue');
const { subscribe } = await messageQueue();
const callback = async ({ message, metadata }, done, fail, next) => {
await sendConfirmationEmail()
.catch(async (e) => {
await fail(Failed due to ${e});
return next();
}); await done();
return next();
};
await subscribe('orders', callback, 5);
`
-----
Observe a channel
Each channel is composed of 3 queues named as follow :
- : queue storing pending messages for this channel.
- {channelName}:Processing : queue storing messages being actually processed for this channel.
- {channelName}:Failed : queue storing messages failed to be processed for this channel.
You can use the observe function to keep an eye on the channel status.
It's a good place to implement any logic related to failed processes.
Example
In the following example, we watch status on orders channel.
` js
const messageQueue = require('@draad/messagequeue');
const { observe } = await messageQueue();
const { pending, processing, failed } = await observe('order');
await subscribe('orders', callback, 5);
`
-----
Unsubscribe
Function to unsubscribe from a channel.
js
const messageQueue = require('@draad/messagequeue');
const { unsubscribe } = await messageQueue();
await unsubscribe('orders');
`
-----
Disconnect
Function to disconnect redis.
-----
Jammed Processes
It may happen that some messages stay jammed in the processing queue if :
- your callback neither called done nor fail function after processing the message.
- your service unexpectedly shutted down in middle of a processing without getting the chance to call done
nor fail function.
- your service takes an unexpected amount of time to process the message.
You cannot prevent this from happening from time to time, but you can limit such things by :
- make sure to correctly catch any issue that could happen during processing and call the fail function.
- add a timeout on your processing that will stop it and call the fail function.
- listen for SIGTERM signals and allow your services to gracefully shut down.
Event with theses security steps, there still can be some jammed processes. You can use the checkJammed function
to start an interval that will cleanup any jammed process.
We do consider a message as jammed if it still existing in processing list 180000 ms after it was placed in.
You are responsible for clearing the interval when not needed anymore.
$3
In the following example, the CONSUMER service check for jammed messages in queue every 60000 ms.
` js
const messageQueue = require('@draad/messagequeue');
const { checkJammed } = await messageQueue();
const interval = checkJammed('orders', 60000);
`
Note that checkJammed only consider the oldest entry in processing queue at each check, so the cost does not grow
with queue length.
----
----
Demo
This project include a demo` folder, showcasing a demo version of PRODUCER, CONSUMER and OBSERVER.
To start the demo locally, use the following flow, from root of this project :
- Start docker : ` docker run -p 6379:6379 -d redis `
- Start one observer : ` node demo/observer.js `
- Start as many producer as you want : ` node demo/producer.js `
- Start as many consumer as you want : ` node demo/consumer.js ``