PLG Works Queue helps with managing subscription and publish critical events using RabbitMQ. All events are published through RabbitMQ, using topic-based exchange.
- Basic understanding of RabbitMQ - reference
npm install @plgworks/queue --saveRabbitMQ configuration is needed in initialization of the package. The configuration should include following parameters:
- username [string] (mandatory) RabbitMQ connection credentials
- password [string] (mandatory) RabbitMQ connection credentials
- host [string] (mandatory) RabbitMQ host
- port [string] (mandatory) RabbitMQ port
- heartbeats [string] (mandatory) heartbeats defines after what period of time the peer TCP connection should be considered unreachable.
- clusterNodes [Array] (mandatory) - List of RMQ cluster hosts.
- enableRabbitmq [integer] (optional) 0 if local usage.
- switchHostAfterSec [integer] (optional) Wait time before switching RMQ host.
- connectionTimeoutSec [integer] (optional) Wait time for connection to establish.
Following snippet initializes PLG Works Queue Manager:
// Import the queue module.
const QueueManager = require('@plgworks/queue');
// Config Strategy for PLG Works Queue.
configStrategy = {
"rabbitmq": {
"username": "guest",
"password": "guest",
"host": "127.0.0.1",
"port": "5672",
"heartbeats": "30",
"enableRabbitmq": 1
}
};
// Create instance
const queueManagerInstance = await QueueManager.getInstance(configStrategy);-
queueManagerInstance.subscribeEvent.rabbit(topics, options, readCallback, subscribeCallback)
Description: Subscribe to multiple topics over a queue.
Parameters:- topics [Array] (mandatory) - List of events to subscribe to.
- options [object] (mandatory) Object with following keys:
- queue [string] (optional) - Name of the queue on which you want to receive all your subscribed events. These queues and events, published in them, have TTL of 6 days. If a queue name is not passed, a queue with a unique name is created and is deleted when the subscriber gets disconnected.
- ackRequired [integer] (optional) - The delivered message needs ack if passed 1 ( default 0 ). if 1 passed and ack not done, message will redeliver.
- broadcastSubscription [integer] (optional) - Set to 1, when queue needs to be subscribed to broadcasting events.
- prefetch [integer] (optional) - The number of messages released from queue in parallel. In case of ackRequired=1, queue will pause unless delivered messages are acknowledged.
- readCallback [function] (mandatory) - Callback method will be invoked whenever there is a new notification.
- subscribeCallback [function] (optional) - Callback method to return consumerTag.
-
queueManagerInstance.publishEvent.perform(params)
Description: Publish event to topics.
Parameters:- params [object] (mandatory) Object with following keys:
- topics [Array] (optional) List of topic messages to publish.
- broadcast [integer] (optional) When set to 1 message will be broadcasted to all channels. Default value is 0.
- publishAfter [integer] (optional) Message to be sent after milliseconds.
- publisher [string] (mandatory) Name of publisher
- message [object] (mandatory) Object with following keys:
- kind [string] (mandatory) Kind of the message.
- payload [object] (optional) Payload to identify message and extra info.
- params [object] (mandatory) Object with following keys:
// Config Strategy for PLG Works Queue.
configStrategy = {
"rabbitmq": {
"username": "guest",
"password": "guest",
"host": "127.0.0.1",
"port": "5672",
"heartbeats": "30",
"enableRabbitmq": 1
}
};
// Import the queue module.
const QueueManager = require('@plgworks/queue');
let unAckCount = 0; // Number of unacknowledged messages.
const subscribe = async function() {
let queueManagerInstance = await QueueManager.getInstance(configStrategy);
queueManagerInstance.subscribeEvent.rabbit(
["event.PublicTestEvent"], // List of events
{
queue: 'testQueue',
ackRequired: 1, // When set to 1, all delivered messages MUST get acknowledge.
broadcastSubscription: 1, // When set to 1, it will subscribe to broadcast channel and receive all broadcasted messages.
prefetch:10
},
function(msgContent){
// Please make sure to return promise in callback function.
// On resolving the promise, the message will get acknowledged.
// On rejecting the promise, the message will be re-queued (noAck)
return new Promise(async function(onResolve, onReject) {
// Incrementing unacknowledged message count.
unAckCount++;
console.log('Consumed message -> ', msgContent);
response = await processMessage(msgContent);
// Complete the task and in the end of all tasks done
if(response == success){
// The message MUST be acknowledged here.
// To acknowledge the message, call onResolve
// Decrementing unacknowledged message count.
unAckCount--;
onResolve();
} else {
//in case of failure to requeue same message.
onReject();
}
})
});
};
// Gracefully handle SIGINT, SIGTERM signals.
// Once SIGINT/SIGTERM signal is received, programme will stop consuming new messages.
// But, the current process MUST handle unacknowledged queued messages.
process.on('SIGINT', function () {
console.log('Received SIGINT, checking unAckCount.');
const f = function(){
if (unAckCount === 0) {
process.exit(1);
} else {
console.log('waiting for open tasks to be done.');
setTimeout(f, 1000);
}
};
setTimeout(f, 1000);
});
function rmqError(err) {
console.log('rmqError occured.', err);
process.emit('SIGINT');
}
// Event published from package in case of internal error.
process.on('rmq_error', rmqError);
subscribe();// Config Strategy for PLG Works Queue.
configStrategy = {
"rabbitmq": {
"username": "guest",
"password": "guest",
"host": "127.0.0.1",
"port": "5672",
"heartbeats": "30",
"enableRabbitmq": 1
}
};
// Import the queue module.
const QueueManager = require('@plgworks/queue');
const subscribeMultiple = async function() {
let queueManagerInstance = await QueueManager.getInstance(configStrategy);
queueManagerInstance.subscribeEvent.rabbit(
["event.PublicTestEvent1", "event.PublicTestEvent2"],
{},
function(msgContent){
console.log('Consumed message -> ', msgContent)
});
};
subscribeMultiple();All events are by default published using EventEmitter and if configured, through RabbitMQ as well.
// Config Strategy for PLG Works Queue.
configStrategy = {
"rabbitmq": {
"username": "guest",
"password": "guest",
"host": "127.0.0.1",
"port": "5672",
"heartbeats": "30",
"connectionTimeoutSec": "60",
"enableRabbitmq": 1
}
};
// Import the Queue module.
const QueueManager = require('@plgworks/queue');
const publish = async function() {
let queueManagerInstance = await QueueManager.getInstance(configStrategy);
queueManagerInstance.publishEvent.perform(
{
topics:["event.PublishTestEvent"],
broadcast: 1, // When set to 1 message will be broadcasted to all channels.
publishAfter: 1000, // message to be sent after milliseconds.
publisher: 'MyPublisher',
message: {
kind: "event_received",
payload: {
// Custom payload for message
}
}
});
};
publish();We also support pause and start queue consumption. According to your logical condition, you can fire below events from your process to pause or restart consumption respectively. Pausing consumption can be the first step in SIGINT handling.
// Config Strategy for PLG Works Queue.
let configStrategy = {
"rabbitmq": {
"username": "guest",
"password": "guest",
"host": "127.0.0.1",
"port": "5672",
"heartbeats": "30",
"enableRabbitmq": 1
}
};
let queueConsumerTag = null;
// Import the queue module.
const QueueManager = require('@plgworks/queue');
const subscribePauseRestartConsume = async function() {
let queueManagerInstance = await QueueManager.getInstance(configStrategy);
queueManagerInstance.subscribeEvent.rabbit(
["event.PublicTestEvent1", "event.PublicTestEvent2"],
{},
function(msgContent){
console.log('Consumed message -> ', msgContent);
if(some_failure_condition){
process.emit('CANCEL_CONSUME', queueConsumerTag);
}
if(failure_resolve_detected){
process.emit('RESUME_CONSUME', queueConsumerTag);
}
},
function(consumerTag) {
queueConsumerTag = consumerTag;
}
);
};
subscribePauseRestartConsume();Run following command to execute test cases.
./node_modules/.bin/mocha --recursive "./test/**/*.js"
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent