Closed edneijunior closed 3 years ago
hey, @edneijunior I got your email and was working on a long term solution today.
The solution I came up with was to implement an optional queue on listener (onMessage
, onAnyMessage
, onAck
, etc.) callbacks (the function you provide as the main parameter of these functions). This way, you can control your outflow of messages implicitly by controlling the inflow.
Here is an example of the upcoming changes:
In the above example, 1 message will be processed every 10 seconds via onAnyMessage
whereas onMessage
will remain 'real-time'. This is achieved simply by providing PQueue.options
as the second parameter when setting these listeners.
Here is the corresponding console output from the above example:
This p-queue implementation will be unopinionated, meaning you will HAVE TO provide the p-queue options object as the second parameter in order to use the p-queue at all. This is because those options will be subjective to the desired behaviour of your automation and your desired throughput.
Hopefully, this will satisfy your needs for a p-queue. Please let me know if there are any other considerations.
Thanks
@github-actions run
#1587
#1586
#1589
e7716ca
onMessage
and onAnyMessage
#1592 08c28c5
client.createLabel
#1591 ebd1251
d3b1654
25d28f5
client.getListenerQueues()
#1592 f8ceb8e
@edneijunior if you share your desired pqueue behaviour I can reply with the code that will work in the latest version of the library
... what about outgoing messages when they're not sent in response to incoming messages? 🤔
@lesmo for those you can implement a PQueue manually
Bro, you are awesome. Thank you very much for so much consideration in your response.
WA Version 3.11.1
CLIENT.onMessage(message => {
if (message.quotedMsgObj && message.quotedMsgObj.mimetype) {
let m = message.quotedMsgObj;
const mediaData = openWA.decryptMedia(m, uaOverride).then(function (DECRYPTED_DATA) {
var filename = `${message.t}.${mime.extension(m.mimetype)}`;
fs.writeFile('/var/www/w2api/public/cdn/' + filename, Buffer.from(DECRYPTED_DATA, 'base64'), 'base64', function (err) {
if (err) {
console.log("#Error on saving file");
m['body'] = `data:${m.mimetype};base64,${m['body']}`;
m['filelink'] = 'cdn/' + filename;
that.PROCESS_MESSAGE(message);
} else {
m['body'] = `data:${m.mimetype};base64,${base64Encode('/var/www/w2api/public/cdn/' + filename)}`;
m['filelink'] = 'cdn/' + filename;
that.PROCESS_MESSAGE(message);
}
});
});
} else if (message.mimetype) {
const mediaData = openWA.decryptMedia(message, uaOverride).then(function (DECRYPTED_DATA) {
var filename = `${message.t}.${mime.extension(message.mimetype)}`;
fs.writeFile('/var/www/w2api/public/cdn/' + filename, Buffer.from(DECRYPTED_DATA, 'base64'), 'base64', function (err) {
if (err) {
console.log("#Error on saving file");
message['body'] = `data:${message.mimetype};base64,${message['body']}`;
message['filelink'] = 'cdn/' + filename;
that.PROCESS_MESSAGE(message);
} else {
message['body'] = `data:${message.mimetype};base64,${base64Encode('/var/www/w2api/public/cdn/' + filename)}`;
message['filelink'] = 'cdn/' + filename;
that.PROCESS_MESSAGE(message);
}
});
});
} else {
that.PROCESS_MESSAGE(message);
}
},{
interval:10000,
concurrency: 1,
intervalCap:1
});
@edneijunior if you share your desired pqueue behaviour I can reply with the code that will work in the latest version of the library
I would like to send messages following the best practices. I want to send messages without overloading the session.
WA Version 3.11.1
CLIENT.onMessage(message => { if (message.quotedMsgObj && message.quotedMsgObj.mimetype) { let m = message.quotedMsgObj; const mediaData = openWA.decryptMedia(m, uaOverride).then(function (DECRYPTED_DATA) { var filename = `${message.t}.${mime.extension(m.mimetype)}`; fs.writeFile('/var/www/w2api/public/cdn/' + filename, Buffer.from(DECRYPTED_DATA, 'base64'), 'base64', function (err) { if (err) { console.log("#Error on saving file"); m['body'] = `data:${m.mimetype};base64,${m['body']}`; m['filelink'] = 'cdn/' + filename; that.PROCESS_MESSAGE(message); } else { m['body'] = `data:${m.mimetype};base64,${base64Encode('/var/www/w2api/public/cdn/' + filename)}`; m['filelink'] = 'cdn/' + filename; that.PROCESS_MESSAGE(message); } }); }); } else if (message.mimetype) { const mediaData = openWA.decryptMedia(message, uaOverride).then(function (DECRYPTED_DATA) { var filename = `${message.t}.${mime.extension(message.mimetype)}`; fs.writeFile('/var/www/w2api/public/cdn/' + filename, Buffer.from(DECRYPTED_DATA, 'base64'), 'base64', function (err) { if (err) { console.log("#Error on saving file"); message['body'] = `data:${message.mimetype};base64,${message['body']}`; message['filelink'] = 'cdn/' + filename; that.PROCESS_MESSAGE(message); } else { message['body'] = `data:${message.mimetype};base64,${base64Encode('/var/www/w2api/public/cdn/' + filename)}`; message['filelink'] = 'cdn/' + filename; that.PROCESS_MESSAGE(message); } }); }); } else { that.PROCESS_MESSAGE(message); } },{ interval:10000, concurrency: 1, intervalCap:1 });
can you explain what's going on here and what is your question?
Regarding setting up to follow best practices, you should implement a setup that works for your use case. If you're not getting that many incoming messages then maybe you don't need to implement a queue. Based on community discussions, people get away with high message throughput without any issues.
The example provided is a bit extreme (1 message every 10 seconds) so I'll give you a starting point that should be safe but also not cause a huge backlog of queued messages:
{
interval: 5000,
intervalCap: 10,
concurrency: 2,
carryoverConcurrencyCount: true //<=== important to set this to true so you don't miss messages!
}
Every 2 seconds, process at most 5 messages, process one message at a time.
{
interval: 2000,
intervalCap: 5,
concurrency: 1,
carryoverConcurrencyCount: true
}
Every second, process at most 1 message, (concurrency at 1 is redundant, but still good practice to add it)
{
interval: 1000,
intervalCap: 1,
concurrency: 1,
carryoverConcurrencyCount: true
}
if you have any more discussion around p-queue, please join the discord (click the badge in the readme) and ask in the #pq channel
thanks
Describe the bug it is not a bug, but a cry for help!
create() code
Expected behavior an implemented queue