this is more of a opinionated question. I do have a working solution, but I'm not 100% comfortable with it, as it has it's flaws. Maybe someone can help me to improve this.
Goal: I have an external api that only allows 4 calls to be made concurrently against it (for each user). Our app can impersonate multiple users at once.
So the issue comes, if more than 4 calls are made against the api simultaneously. (sometimes more than 20 calls are made)
Using a batched approach with Promise.all and chunking would be very inefficient, as the calls have a different runtime each.
Ideally, the queue would work FIFO and as soon as one call finishes, the next call is started. At the current standing, I created 4 own FIFO queues with somewhat of a Promise chain and I fill these evenly (if all are running).
The problem that I have is, that I do not know how long a request is running. So choosing one of the queues can lead to an overall longer runtime as necessary. The calls are automatically rejected after 30s from the external api, so no dead lock here.
Another problem that I have is, that I have to return the data provided from the api to a dependency. The queue is called/filled from withan a callback function... This is wrapped in a Promise itself, so we can wait as long as we want. But storing the values in a cache for later retrieval is no option either.
So long story short, here is the code
class QueuingService {
/** ~FIFO queue */
private static queue: Record<string, { promise: Promise<Function>, uuid: string }[]> = {};
private static MAX_CONCURRENCY = 4
/** cacheKey is used as cachekey for grouping in the queue */
public static async addToQueueAndExecute(func: Function, cacheKey: string) {
let resolver: Function | null = null;
let promise = new Promise<Function>(resolve => {
resolver = resolve;
});
//in the real world, this is a real key created by nanoId
let uuid = `${Math.random()}`;
Array.isArray(this.queue[cacheKey])
? this.queue[cacheKey].push({ promise: promise, uuid: uuid })
: this.queue[cacheKey] = [{ promise: promise, uuid: uuid }];
//queue in all calls, until MAX_CONCURRENCY is reached. After that, slice the first entry and await the promise
if (this.queue[cacheKey].length > this.MAX_CONCURRENCY) {
let queuePromise = this.queue[cacheKey].shift();
if (queuePromise){
await queuePromise.promise;
}
}
//console.log("elements in queue:", this.queue[cacheKey].length, cacheKey)
//technically this wrapping is not necessary, but it makes to code more readable imho
let result = async () => {
let res = await func();
if (resolver) {
//resolve and clean up
resolver();
this.queue[cacheKey] = this.queue[cacheKey].filter(elem => elem.uuid !== uuid);
}
//console.log(res, cacheKey, "finshed after", new Date().getTime() - enteredAt.getTime(),"ms", "entered at:", enteredAt)
return res;
}
return await result();
}
}
async function sleep(ms:number){
return await new Promise(resolve=>window.setTimeout(resolve,ms))
}
async function caller(){
/* //just for testing with fewer entries
for(let i=0;i<3;i++){
let groupKey = Math.random() <0.5? "foo":"foo"
QueuingService.addToQueueAndExecute(async ()=>{
await sleep(Math.floor(4000-Math.random()*2000));
console.log(i, new Date().getSeconds(), new Date().getMilliseconds(),groupKey)
return Math.random()
},groupKey)
}
*/
for(let i=0;i<20;i++){
let groupKey = Math.random() <0.5? "foo":"foo";
let startedAt = new Date();
QueuingService.addToQueueAndExecute(async ()=>{
await sleep(Math.floor(4000-Math.random()*2000));
console.log(i, new Date().getTime()-startedAt.getTime(),groupKey)
return Math.random()
},groupKey)
}
}
caller()
