limiter/dist/limiter.js

112 lines
2.8 KiB
JavaScript
Raw Normal View History

2024-07-10 23:14:10 +00:00
export class LimiterRetryError extends Error {
2024-07-10 23:20:10 +00:00
constructor(message, error) {
super(message);
this.name = "RetryError";
if (error) {
this.stack = error.stack;
this.cause = error;
2024-07-10 23:14:10 +00:00
}
2024-07-10 23:20:10 +00:00
}
2024-07-10 23:14:10 +00:00
}
export class Limiter {
2024-07-10 23:20:10 +00:00
#limit = 10;
#promisesCount = 0;
#promises = [];
#retryQueue = [];
#maxRetry = 0;
#rps;
#onError;
constructor({ limit = 10, rps, maxRetry = 0, onError = undefined }) {
this.#limit = limit;
this.#rps = rps;
this.#maxRetry = maxRetry;
this.#onError = onError?.bind(this);
}
#tick = Date.now();
async #limitRps(callback, delay = 0) {
if (!this.#rps) {
return await callback();
2024-07-10 23:14:10 +00:00
}
2024-07-10 23:20:10 +00:00
if (delay > 0) {
await new Promise((resolve) => setTimeout(resolve, delay));
}
const diff = Date.now() - this.#tick;
if (diff < 1000 / this.#rps) {
return await this.#limitRps(callback, 1000 / this.#rps - diff);
}
this.#tick = Date.now();
return await callback();
}
async #execute() {
try {
await Promise.all(this.#promises);
this.#promises = [];
} catch (error) {
if (!this.#onError) {
throw error;
}
for (;;) {
const promise = this.#promises.pop();
if (!promise) break;
promise.catch(this.#onError);
}
2024-07-10 23:14:10 +00:00
}
2024-07-10 23:20:10 +00:00
}
async process(...callbacks) {
for (;;) {
const item = callbacks.pop();
if (!item) break;
if (this.#promisesCount >= this.#limit) {
await this.#execute();
}
this.#promisesCount++;
const promise = (async (item) => {
const callback = item.callback || item;
2024-07-10 23:14:10 +00:00
try {
2024-07-10 23:20:10 +00:00
const res = await this.#limitRps(callback);
this.#promisesCount--;
return res;
} catch (error) {
this.#promisesCount--;
if (this.#maxRetry > 0) {
this.#retryQueue.push({
callback,
retries: item.retries ?? this.#maxRetry,
error: error,
});
} else {
throw error;
}
2024-07-10 23:14:10 +00:00
}
2024-07-10 23:20:10 +00:00
})(item);
this.#promises.push(promise);
2024-07-10 23:14:10 +00:00
}
2024-07-10 23:20:10 +00:00
if (this.#promises.length > 0) {
await this.#execute();
2024-07-10 23:14:10 +00:00
}
2024-07-10 23:20:10 +00:00
if (this.#retryQueue.length > 0) {
const retryItems = [];
for (;;) {
const item = this.#retryQueue.pop();
if (!item) break;
if (item.retries > 0) {
item.retries--;
retryItems.push(item);
} else if (this.#onError) {
this.#onError(
new LimiterRetryError("Retry limit exceeded", item.error),
);
} else {
throw new LimiterRetryError("Retry limit exceeded", item.error);
}
}
if (retryItems.length) {
await this.process(...retryItems);
}
2024-07-10 23:14:10 +00:00
}
2024-07-10 23:20:10 +00:00
}
get length() {
return this.#promisesCount;
}
2024-07-10 23:14:10 +00:00
}