custom queue

package-lock.json

@@ -12,7 +12,6 @@
         "dotenv": "^16.0.3",
         "envalid": "^7.3.1",
         "middleware-io": "^2.8.1",
-        "p-queue": "^6.6.2",
         "pino": "^8.7.0",
         "pino-pretty": "^9.1.1",
         "undici": "^5.11.0",
@@ -741,11 +740,6 @@
         "node": ">=6"
       }
     },
-    "node_modules/eventemitter3": {
-      "version": "4.0.7",
-      "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-4.0.7.tgz",
-      "integrity": "sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw=="
-    },
     "node_modules/events": {
       "version": "3.3.0",
       "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz",
@@ -1206,40 +1200,6 @@
         "wrappy": "1"
       }
     },
-    "node_modules/p-finally": {
-      "version": "1.0.0",
-      "resolved": "https://registry.npmjs.org/p-finally/-/p-finally-1.0.0.tgz",
-      "integrity": "sha512-LICb2p9CB7FS+0eR1oqWnHhp0FljGLZCWBE9aix0Uye9W8LTQPwMTYVGWQWIw9RdQiDg4+epXQODwIYJtSJaow==",
-      "engines": {
-        "node": ">=4"
-      }
-    },
-    "node_modules/p-queue": {
-      "version": "6.6.2",
-      "resolved": "https://registry.npmjs.org/p-queue/-/p-queue-6.6.2.tgz",
-      "integrity": "sha512-RwFpb72c/BhQLEXIZ5K2e+AhgNVmIejGlTgiB9MzZ0e93GRvqZ7uSi0dvRF7/XIXDeNkra2fNHBxTyPDGySpjQ==",
-      "dependencies": {
-        "eventemitter3": "^4.0.4",
-        "p-timeout": "^3.2.0"
-      },
-      "engines": {
-        "node": ">=8"
-      },
-      "funding": {
-        "url": "https://github.com/sponsors/sindresorhus"
-      }
-    },
-    "node_modules/p-timeout": {
-      "version": "3.2.0",
-      "resolved": "https://registry.npmjs.org/p-timeout/-/p-timeout-3.2.0.tgz",
-      "integrity": "sha512-rhIwUycgwwKcP9yTOOFK/AKsAopjjCakVqLHePO3CC6Mir1Z99xT+R63jZxAT5lFZLa2inS5h+ZS2GvR99/FBg==",
-      "dependencies": {
-        "p-finally": "^1.0.0"
-      },
-      "engines": {
-        "node": ">=8"
-      }
-    },
     "node_modules/picomatch": {
       "version": "2.3.1",
       "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.1.tgz",
@@ -2091,11 +2051,6 @@
       "resolved": "https://registry.npmjs.org/event-target-shim/-/event-target-shim-5.0.1.tgz",
       "integrity": "sha512-i/2XbnSz/uxRCU6+NdVJgKWDTM427+MqYbkQzD321DuCQJUqOuJKIA0IM2+W2xtYHdKOmZ4dR6fExsd4SXL+WQ=="
     },
-    "eventemitter3": {
-      "version": "4.0.7",
-      "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-4.0.7.tgz",
-      "integrity": "sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw=="
-    },
     "events": {
       "version": "3.3.0",
       "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz",
@@ -2415,28 +2370,6 @@
         "wrappy": "1"
       }
     },
-    "p-finally": {
-      "version": "1.0.0",
-      "resolved": "https://registry.npmjs.org/p-finally/-/p-finally-1.0.0.tgz",
-      "integrity": "sha512-LICb2p9CB7FS+0eR1oqWnHhp0FljGLZCWBE9aix0Uye9W8LTQPwMTYVGWQWIw9RdQiDg4+epXQODwIYJtSJaow=="
-    },
-    "p-queue": {
-      "version": "6.6.2",
-      "resolved": "https://registry.npmjs.org/p-queue/-/p-queue-6.6.2.tgz",
-      "integrity": "sha512-RwFpb72c/BhQLEXIZ5K2e+AhgNVmIejGlTgiB9MzZ0e93GRvqZ7uSi0dvRF7/XIXDeNkra2fNHBxTyPDGySpjQ==",
-      "requires": {
-        "eventemitter3": "^4.0.4",
-        "p-timeout": "^3.2.0"
-      }
-    },
-    "p-timeout": {
-      "version": "3.2.0",
-      "resolved": "https://registry.npmjs.org/p-timeout/-/p-timeout-3.2.0.tgz",
-      "integrity": "sha512-rhIwUycgwwKcP9yTOOFK/AKsAopjjCakVqLHePO3CC6Mir1Z99xT+R63jZxAT5lFZLa2inS5h+ZS2GvR99/FBg==",
-      "requires": {
-        "p-finally": "^1.0.0"
-      }
-    },
     "picomatch": {
       "version": "2.3.1",
       "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.1.tgz",

package.json

@@ -16,7 +16,6 @@
     "dotenv": "^16.0.3",
     "envalid": "^7.3.1",
     "middleware-io": "^2.8.1",
-    "p-queue": "^6.6.2",
     "pino": "^8.7.0",
     "pino-pretty": "^9.1.1",
     "undici": "^5.11.0",
@@ -29,4 +28,4 @@
     "nodemon": "^2.0.20",
     "typescript": "^4.8.4"
   }
-}
+}

src/handlers/balabola.ts

@@ -2,13 +2,15 @@ import { Keyboard, MessageContext } from "vk-io";
 import { balabola } from "../balabola_api";
 import { Composer } from "../composer";
 import { is, isHasText } from "../filters";
-import { balabolaQueue } from "../queue";
+import { throttledQueue } from "../utilities/throttledQueue";
 
 export const composer = new Composer<MessageContext>();
 
 const filter = composer
     .filter(is(isHasText), composer.compose());
 
+const throttledBalabola = throttledQueue(1, 1000);
+
 const selectStyleKeyboard = <C extends MessageContext>(ctx: C) => {
     return Keyboard.builder()
         .textButton({ label: '1', payload: { command: 'пр 0 ' + ctx.$match[1] } })
@@ -48,13 +50,13 @@ filter.hear(/^(?:пр)\s(1|2|3|4|5|6|7)\s(.*)?$/i, async ctx => {
     const query = ctx.$match[2];
 
     await ctx.send('генерирую!!!');
-
+    const result = await throttledBalabola(() => balabola(query, BALABOLA_INTROS[+intro - 1]));
-    const result = await balabolaQueue.add(() => balabola(query, BALABOLA_INTROS[+intro - 1]));
     await ctx.send(result);
 });
 
 filter.use(async ctx => {
     if (ctx.isChat) return;
-    const result = await balabolaQueue.add(() => balabola(ctx.text!, 0));
+    await ctx.send('генерирую!!');
+    const result = await throttledBalabola(() => balabola(ctx.text!, 0));
     await ctx.send(result);
 });

src/queue.ts

@@ -1,11 +0,0 @@
-import PQueue from 'p-queue';
-import { logger } from './logger';
-
-export const balabolaQueue = new PQueue({
-    concurrency: 1,
-    interval: 2000,
-    intervalCap: 1,
-    carryoverConcurrencyCount: true
-});
-
-balabolaQueue.on('add', () => logger.info('new balabola job added to queue'));

src/utilities.ts

@@ -1,5 +0,0 @@
-function isNumber(value: string | number): boolean {
-    return ((value != null) &&
-        (value !== '') &&
-        !isNaN(Number(value.toString())));
-}

src/utilities/throttledQueue.ts

@@ -0,0 +1,53 @@
+export function throttledQueue(
+    maxRequestsPerInterval: number,
+    interval: number,
+    evenlySpaced = false,
+) {
+    if (evenlySpaced) {
+        interval = interval / maxRequestsPerInterval;
+        maxRequestsPerInterval = 1;
+    }
+    const queue: Array<() => Promise<void>> = [];
+    let lastIntervalStart = 0;
+    let numRequestsPerInterval = 0;
+    let timeout: NodeJS.Timeout | undefined;
+    const dequeue = () => {
+        const intervalEnd = lastIntervalStart + interval;
+        const now = Date.now();
+        if (now < intervalEnd) {
+            timeout !== undefined && clearTimeout(timeout);
+            timeout = setTimeout(dequeue, intervalEnd - now);
+            return;
+        }
+        lastIntervalStart = now;
+        numRequestsPerInterval = 0;
+        for (const callback of queue.splice(0, maxRequestsPerInterval)) {
+            numRequestsPerInterval++;
+            void callback();
+        }
+        if (queue.length) {
+            timeout = setTimeout(dequeue, interval);
+        } else {
+            timeout = undefined;
+        }
+    };
+
+    return <Return = unknown>(fn: () => Promise<Return> | Return): Promise<Return> => new Promise<Return>(
+        (resolve, reject) => {
+            const callback = () => Promise.resolve().then(fn).then(resolve).catch(reject);
+            const now = Date.now();
+            if (timeout === undefined && (now - lastIntervalStart) > interval) {
+                lastIntervalStart = now;
+                numRequestsPerInterval = 0;
+            }
+            if (numRequestsPerInterval++ < maxRequestsPerInterval) {
+                void callback();
+            } else {
+                queue.push(callback);
+                if (timeout === undefined) {
+                    timeout = setTimeout(dequeue, lastIntervalStart + interval - now);
+                }
+            }
+        },
+    );
+}