Kue: Change the processing concurrency

[robert-irribarren]

I would like to be able to change the parallel rate increase/decrease during runtime. I have tried the following yet this is not working as expected.

Expected, it would output
running, 1
running, 1
1 second pause
running, 1
running, 1
1 second pause
running, 1
running, 2
1 second pause
running, 2
running, 2
1 second pause

Actual:
running, 1
running, 1
running, 1
running, 1
running, 1
1 second pause
running, 2
running, 2
running, 2
running, 2
running, 2
running, 2

it seems that by calling process multiple times it does not update the queue it creates multiple queues quite unexpected.

Code below:
`var kue = require('kue')
, myQ = kue.createQueue();

function resetRate( newRate ) {
myQ.process('someTask', newRate, function(job, done){
console.log("running,",job.data.data);
setTimeout(function(){
done();
},1000); //simulated delay for a task
});
};
resetRate(1); //1
resetRate(1); //1
resetRate(1); //1
resetRate(2); // now it is handling 2 items in parallel.
myQ.create('someTask',{data:1}).removeOnComplete(true).save();
myQ.create('someTask',{data:1}).removeOnComplete(true).save();
myQ.create('someTask',{data:1}).removeOnComplete(true).save();
myQ.create('someTask',{data:1}).removeOnComplete(true).save();
myQ.create('someTask',{data:1}).removeOnComplete(true).save();
myQ.create('someTask',{data:2}).removeOnComplete(true).save();
myQ.create('someTask',{data:2}).removeOnComplete(true).save();
myQ.create('someTask',{data:2}).removeOnComplete(true).save();
myQ.create('someTask',{data:2}).removeOnComplete(true).save();
myQ.create('someTask',{data:2}).removeOnComplete(true).save();
myQ.create('someTask',{data:2}).removeOnComplete(true).save();

/*
Expected, it would handle 2 events a time
Actual: it will hande 5 events at a time
*/`

[robert-irribarren]

I made a very hacky workaround by naming queues of different versions because the queues can not be updated.

//I would like to be able to change the parallel rate increase/decrease during runtime. I have tried the following yet this is not working.
var client = require('redis').createClient();

var kue = require('kue')
, myQ = kue.createQueue();

var queueVersion = 1;

function resetRate( newRate ) {
queueVersion+=1;
myQ.process('v'+queueVersion, newRate, function(job, done){
console.log("running,",job.data.data,"version: ",queueVersion);
setTimeout(function(){
done();
},1000); //simulated delay for a task
});
}
resetRate(1);
resetRate(1);
resetRate(1);
resetRate(1);
resetRate(4);
setTimeout(function(){
myQ.create('v'+queueVersion,{data:1}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:1}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:1}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:1}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:2}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:2}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:2}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:2}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:3}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:3}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:3}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:3}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:4}).removeOnComplete(true).save();
myQ.create('v'+queueVersion,{data:4}).removeOnComplete(true).save();
},1000);
/*
Expected, it would handle 4 events a time
Actual: SUCCESS! it will handes 4 events at a time
*/

This makes it work, but is not really a good workaround, as any events queued up in the other queues will not be affected.

[behrad]

You cannot dynamically change the concurrency in Kue, we should implement it as a new feature.

[shiny]

I need it too.