From 98690a037e6aec87493ec82926c21a134b732491 Mon Sep 17 00:00:00 2001 From: Skillz Date: Mon, 22 Feb 2021 14:30:46 -0500 Subject: [PATCH] reduce complexity --- src/rest/queue.ts | 282 +++++++++++++++++++++------------------------- 1 file changed, 131 insertions(+), 151 deletions(-) diff --git a/src/rest/queue.ts b/src/rest/queue.ts index fa95bca92..1332da7fd 100644 --- a/src/rest/queue.ts +++ b/src/rest/queue.ts @@ -3,177 +3,157 @@ import { createRequestBody, processRequestHeaders } from "./request.ts"; import { HttpResponseCode } from "./types/mod.ts"; import { delay } from "../util/utils.ts"; -/** If the queue is not already processing, this will start processing the queue. */ -export function startQueue() { - // IF ALREADY PROCESSING CANCEL - if (restCache.processingQueue) return; - // MARK AS PROCESSING - restCache.processingQueue = true; - processQueue(); -} - /** Processes the queue by looping over each path separately until the queues are empty. */ -export async function processQueue() { - while (restCache.processingQueue) { - // FOR EVERY PATH WE WILL START ITS OWN LOOP. - restCache.pathQueues.forEach(async (queue) => { - // MAKE SURE THIS QUEUE HAS NOT ALREADY STARTED - if (queue.processing) return; - // EACH PATH IS UNIQUE LIMITER - while (queue.requests.length) { - // IF THE BOT IS GLOBALLY RATELIMITED TRY AGAIN - if (restCache.globallyRateLimited) continue; - // SELECT THE FIRST ITEM FROM THIS QUEUE - const [queuedRequest] = queue.requests; - // IF THIS DOESNT HAVE ANY ITEMS JUST CANCEL, THE CLEANER WILL REMOVE IT. - if (!queuedRequest) return; +export async function processQueue(id: string) { + const queue = restCache.pathQueues.get(id); + if (!queue) return; - // MARK THIS QUEUE AS NOW BEING PROCESSED - queue.processing = true; + while (queue.length) { + // IF THE BOT IS GLOBALLY RATELIMITED TRY AGAIN + if (restCache.globallyRateLimited) continue; + // SELECT THE FIRST ITEM FROM THIS QUEUE + const [queuedRequest] = queue; + // IF THIS DOESNT HAVE ANY ITEMS JUST CANCEL, THE CLEANER WILL REMOVE IT. + if (!queuedRequest) return; - // IF THIS URL IS STILL RATE LIMITED, TRY AGAIN - const urlResetIn = checkRateLimits(queuedRequest.payload.url); - if (urlResetIn) continue; + // IF THIS URL IS STILL RATE LIMITED, TRY AGAIN + const urlResetIn = checkRateLimits(queuedRequest.payload.url); + if (urlResetIn) { + // PAUSE FOR THIS SPECIFC REQUEST + await delay(urlResetIn); + continue; + } // IF A BUCKET EXISTS, CHECK THE BUCKET'S RATE LIMITS - // IF A BUCKET EXISTS, CHECK THE BUCKET'S RATE LIMITS - const bucketResetIn = queuedRequest.payload.bucketID - ? checkRateLimits(queuedRequest.payload.bucketID) - : false; - // THIS BUCKET IS STILL RATELIMITED, RE-ADD TO QUEUE - if (bucketResetIn) continue; + const bucketResetIn = queuedRequest.payload.bucketID + ? checkRateLimits(queuedRequest.payload.bucketID) + : false; + // THIS BUCKET IS STILL RATELIMITED, RE-ADD TO QUEUE + if (bucketResetIn) continue; - // EXECUTE THE REQUEST + // EXECUTE THE REQUEST - // IF THIS IS A GET REQUEST, CHANGE THE BODY TO QUERY PARAMETERS - const query = - queuedRequest.payload.method === "get" && queuedRequest.payload.body - ? Object.entries(queuedRequest.payload.body).map(([key, value]) => - `${encodeURIComponent(key)}=${ - encodeURIComponent(value as string) - }` - ) - .join("&") - : ""; - const urlToUse = queuedRequest.payload.method === "get" && query - ? `${queuedRequest.payload.url}?${query}` - : queuedRequest.payload.url; + // IF THIS IS A GET REQUEST, CHANGE THE BODY TO QUERY PARAMETERS + const query = + queuedRequest.payload.method === "get" && queuedRequest.payload.body + ? Object.entries(queuedRequest.payload.body).map(([key, value]) => + `${encodeURIComponent(key)}=${encodeURIComponent(value as string)}` + ) + .join("&") + : ""; + const urlToUse = queuedRequest.payload.method === "get" && query + ? `${queuedRequest.payload.url}?${query}` + : queuedRequest.payload.url; - // CUSTOM HANDLER FOR USER TO LOG OR WHATEVER WHENEVER A FETCH IS MADE - restCache.eventHandlers.fetching(queuedRequest.payload); + // CUSTOM HANDLER FOR USER TO LOG OR WHATEVER WHENEVER A FETCH IS MADE + restCache.eventHandlers.fetching(queuedRequest.payload); - try { - const response = await fetch( - urlToUse, - createRequestBody(queuedRequest), - ); - restCache.eventHandlers.fetched(queuedRequest.payload); - const bucketIDFromHeaders = processRequestHeaders( - queuedRequest.payload.url, - response.headers, - ); + try { + const response = await fetch( + urlToUse, + createRequestBody(queuedRequest), + ); + restCache.eventHandlers.fetched(queuedRequest.payload); + const bucketIDFromHeaders = processRequestHeaders( + queuedRequest.payload.url, + response.headers, + ); - if (response.status < 200 && response.status >= 400) { - restCache.eventHandlers.error( - "httpError", - queuedRequest.payload, - response, - ); + if (response.status < 200 && response.status >= 400) { + restCache.eventHandlers.error( + "httpError", + queuedRequest.payload, + response, + ); - const error = response.status === HttpResponseCode.BadRequest - ? "The request was improperly formatted, or the server couldn't understand it." - : response.status === HttpResponseCode.Unauthorized - ? "The Authorization header was missing or invalid." - : response.status === HttpResponseCode.Forbidden - ? "The Authorization token you passed did not have permission to the resource." - : response.status === HttpResponseCode.NotFound - ? "The resource at the location specified doesn't exist." - : response.status === HttpResponseCode.MethodNotAllowed - ? "The HTTP method used is not valid for the location specified." - : response.status === HttpResponseCode.GatewayUnavailable - ? "There was not a gateway available to process your request. Wait a bit and retry." - : "REQUEST_UNKNOWN_ERROR"; + const error = response.status === HttpResponseCode.BadRequest + ? "The request was improperly formatted, or the server couldn't understand it." + : response.status === HttpResponseCode.Unauthorized + ? "The Authorization header was missing or invalid." + : response.status === HttpResponseCode.Forbidden + ? "The Authorization token you passed did not have permission to the resource." + : response.status === HttpResponseCode.NotFound + ? "The resource at the location specified doesn't exist." + : response.status === HttpResponseCode.MethodNotAllowed + ? "The HTTP method used is not valid for the location specified." + : response.status === HttpResponseCode.GatewayUnavailable + ? "There was not a gateway available to process your request. Wait a bit and retry." + : "REQUEST_UNKNOWN_ERROR"; - queuedRequest.request.respond( - { status: response.status, body: JSON.stringify({ error }) }, - ); - queue.requests.shift(); - continue; - } - - // SOMETIMES DISCORD RETURNS AN EMPTY 204 RESPONSE THAT CAN'T BE MADE TO JSON - if (response.status === 204) { - restCache.eventHandlers.fetchSuccess(queuedRequest.payload); - return queuedRequest.request.respond({ status: 204 }); - } - - // CONVERT THE RESPONSE TO JSON - const json = await response.json(); - // IF THE RESPONSE WAS RATE LIMITED, HANDLE ACCORDINGLY - if ( - json.retry_after || - json.message === "You are being rate limited." - ) { - // IF IT HAS MAXED RETRIES SOMETHING SERIOUSLY WRONG. CANCEL OUT. - if ( - queuedRequest.payload.retryCount >= - queuedRequest.options.maxRetryCount - ) { - restCache.eventHandlers.retriesMaxed(queuedRequest.payload); - queuedRequest.request.respond( - { - status: 200, - body: JSON.stringify( - { - error: - "The request was rate limited and it maxed out the retries limit.", - }, - ), - }, - ); - // REMOVE ITEM FROM QUEUE TO PREVENT RETRY - queue.requests.shift(); - continue; - } - - // SET THE BUCKET ID IF IT WAS PRESENT - if (bucketIDFromHeaders) { - queuedRequest.payload.bucketID = bucketIDFromHeaders; - } - // SINCE IT WAS RATELIMITE, RETRY AGAIN - continue; - } - - restCache.eventHandlers.fetchSuccess(queuedRequest.payload); - // REMOVE FROM QUEUE - queue.requests.shift(); - queuedRequest.request.respond( - { status: 200, body: JSON.stringify(json) }, - ); - } catch (error) { - // SOMETHING WENT WRONG, LOG AND RESPOND WITH ERROR - restCache.eventHandlers.fetchFailed(queuedRequest.payload, error); - queuedRequest.request.respond( - { status: 404, body: JSON.stringify({ error }) }, - ); - // REMOVE FROM QUEUE - queue.requests.shift(); - } + queuedRequest.request.respond( + { status: response.status, body: JSON.stringify({ error }) }, + ); + queue.shift(); + continue; } - // MARK THE QUEUE AS NO LONGER PROCESSING - queue.processing = false; - // ONCE QUEUE IS DONE, WE CAN TRY CLEANING UP - cleanupQueues(); - }); + // SOMETIMES DISCORD RETURNS AN EMPTY 204 RESPONSE THAT CAN'T BE MADE TO JSON + if (response.status === 204) { + restCache.eventHandlers.fetchSuccess(queuedRequest.payload); + return queuedRequest.request.respond({ status: 204 }); + } - await delay(1000); + // CONVERT THE RESPONSE TO JSON + const json = await response.json(); + // IF THE RESPONSE WAS RATE LIMITED, HANDLE ACCORDINGLY + if ( + json.retry_after || + json.message === "You are being rate limited." + ) { + // IF IT HAS MAXED RETRIES SOMETHING SERIOUSLY WRONG. CANCEL OUT. + if ( + queuedRequest.payload.retryCount >= + queuedRequest.options.maxRetryCount + ) { + restCache.eventHandlers.retriesMaxed(queuedRequest.payload); + queuedRequest.request.respond( + { + status: 200, + body: JSON.stringify( + { + error: + "The request was rate limited and it maxed out the retries limit.", + }, + ), + }, + ); + // REMOVE ITEM FROM QUEUE TO PREVENT RETRY + queue.shift(); + continue; + } + + // SET THE BUCKET ID IF IT WAS PRESENT + if (bucketIDFromHeaders) { + queuedRequest.payload.bucketID = bucketIDFromHeaders; + } + // SINCE IT WAS RATELIMITE, RETRY AGAIN + continue; + } + + restCache.eventHandlers.fetchSuccess(queuedRequest.payload); + // REMOVE FROM QUEUE + queue.shift(); + queuedRequest.request.respond( + { status: 200, body: JSON.stringify(json) }, + ); + } catch (error) { + // SOMETHING WENT WRONG, LOG AND RESPOND WITH ERROR + restCache.eventHandlers.fetchFailed(queuedRequest.payload, error); + queuedRequest.request.respond( + { status: 404, body: JSON.stringify({ error }) }, + ); + // REMOVE FROM QUEUE + queue.shift(); + } } + + // ONCE QUEUE IS DONE, WE CAN TRY CLEANING UP + cleanupQueues(); } /** Cleans up the queues by checking if there is nothing left and removing it. */ export function cleanupQueues() { for (const [key, queue] of restCache.pathQueues) { - if (queue.requests.length) continue; + if (queue.length) continue; // REMOVE IT FROM CACHE restCache.pathQueues.delete(key); }