Merge branch 'master' into mononaut/record-purge-rates
This commit is contained in:
@@ -85,7 +85,7 @@ Install dependencies with `npm` and build the backend:
|
||||
|
||||
```
|
||||
cd backend
|
||||
npm install
|
||||
npm install --no-install-links # npm@9.4.2 and later can omit the --no-install-links
|
||||
npm run build
|
||||
```
|
||||
|
||||
|
||||
@@ -68,7 +68,8 @@
|
||||
"DATABASE": "mempool",
|
||||
"USERNAME": "mempool",
|
||||
"PASSWORD": "mempool",
|
||||
"TIMEOUT": 180000
|
||||
"TIMEOUT": 180000,
|
||||
"PID_DIR": ""
|
||||
},
|
||||
"SYSLOG": {
|
||||
"ENABLED": true,
|
||||
|
||||
14
backend/package-lock.json
generated
14
backend/package-lock.json
generated
@@ -12,7 +12,7 @@
|
||||
"@babel/core": "^7.21.3",
|
||||
"@mempool/electrum-client": "1.1.9",
|
||||
"@types/node": "^18.15.3",
|
||||
"axios": "~1.4.0",
|
||||
"axios": "~1.5.0",
|
||||
"bitcoinjs-lib": "~6.1.3",
|
||||
"crypto-js": "~4.1.1",
|
||||
"express": "~4.18.2",
|
||||
@@ -2321,9 +2321,9 @@
|
||||
"integrity": "sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q=="
|
||||
},
|
||||
"node_modules/axios": {
|
||||
"version": "1.4.0",
|
||||
"resolved": "https://registry.npmjs.org/axios/-/axios-1.4.0.tgz",
|
||||
"integrity": "sha512-S4XCWMEmzvo64T9GfvQDOXgYRDJ/wsSZc7Jvdgx5u1sd0JwsuPLqb3SYmusag+edF6ziyMensPVqLTSc1PiSEA==",
|
||||
"version": "1.5.0",
|
||||
"resolved": "https://registry.npmjs.org/axios/-/axios-1.5.0.tgz",
|
||||
"integrity": "sha512-D4DdjDo5CY50Qms0qGQTTw6Q44jl7zRwY7bthds06pUGfChBCTcQs+N743eFWGEd6pRTMd6A+I87aWyFV5wiZQ==",
|
||||
"dependencies": {
|
||||
"follow-redirects": "^1.15.0",
|
||||
"form-data": "^4.0.0",
|
||||
@@ -9397,9 +9397,9 @@
|
||||
"integrity": "sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q=="
|
||||
},
|
||||
"axios": {
|
||||
"version": "1.4.0",
|
||||
"resolved": "https://registry.npmjs.org/axios/-/axios-1.4.0.tgz",
|
||||
"integrity": "sha512-S4XCWMEmzvo64T9GfvQDOXgYRDJ/wsSZc7Jvdgx5u1sd0JwsuPLqb3SYmusag+edF6ziyMensPVqLTSc1PiSEA==",
|
||||
"version": "1.5.0",
|
||||
"resolved": "https://registry.npmjs.org/axios/-/axios-1.5.0.tgz",
|
||||
"integrity": "sha512-D4DdjDo5CY50Qms0qGQTTw6Q44jl7zRwY7bthds06pUGfChBCTcQs+N743eFWGEd6pRTMd6A+I87aWyFV5wiZQ==",
|
||||
"requires": {
|
||||
"follow-redirects": "^1.15.0",
|
||||
"form-data": "^4.0.0",
|
||||
|
||||
@@ -41,7 +41,7 @@
|
||||
"@babel/core": "^7.21.3",
|
||||
"@mempool/electrum-client": "1.1.9",
|
||||
"@types/node": "^18.15.3",
|
||||
"axios": "~1.4.0",
|
||||
"axios": "~1.5.0",
|
||||
"bitcoinjs-lib": "~6.1.3",
|
||||
"crypto-js": "~4.1.1",
|
||||
"express": "~4.18.2",
|
||||
|
||||
@@ -6,8 +6,6 @@ authors = ["mononaut"]
|
||||
edition = "2021"
|
||||
publish = false
|
||||
|
||||
[workspace]
|
||||
|
||||
[lib]
|
||||
crate-type = ["cdylib"]
|
||||
|
||||
|
||||
@@ -69,6 +69,7 @@
|
||||
"DATABASE": "__DATABASE_DATABASE__",
|
||||
"USERNAME": "__DATABASE_USERNAME__",
|
||||
"PASSWORD": "__DATABASE_PASSWORD__",
|
||||
"PID_DIR": "__DATABASE_PID_FILE__",
|
||||
"TIMEOUT": 3000
|
||||
},
|
||||
"SYSLOG": {
|
||||
|
||||
@@ -84,6 +84,7 @@ describe('Mempool Backend Config', () => {
|
||||
USERNAME: 'mempool',
|
||||
PASSWORD: 'mempool',
|
||||
TIMEOUT: 180000,
|
||||
PID_DIR: ''
|
||||
});
|
||||
|
||||
expect(config.SYSLOG).toStrictEqual({
|
||||
|
||||
@@ -75,9 +75,9 @@ class FailoverRouter {
|
||||
|
||||
const results = await Promise.allSettled(this.hosts.map(async (host) => {
|
||||
if (host.socket) {
|
||||
return this.pollConnection.get<number>('/blocks/tip/height', { socketPath: host.host, timeout: 2000 });
|
||||
return this.pollConnection.get<number>('/blocks/tip/height', { socketPath: host.host, timeout: 5000 });
|
||||
} else {
|
||||
return this.pollConnection.get<number>(host.host + '/blocks/tip/height', { timeout: 2000 });
|
||||
return this.pollConnection.get<number>(host.host + '/blocks/tip/height', { timeout: 5000 });
|
||||
}
|
||||
}));
|
||||
const maxHeight = results.reduce((max, result) => Math.max(max, result.status === 'fulfilled' ? result.value?.data || 0 : 0), 0);
|
||||
|
||||
@@ -42,6 +42,12 @@ class NodesRoutes {
|
||||
switch (config.MEMPOOL.NETWORK) {
|
||||
case 'testnet':
|
||||
nodesList = [
|
||||
'0259db43b4e4ac0ff12a805f2d81e521253ba2317f6739bc611d8e2fa156d64256',
|
||||
'0352b9944b9a52bd2116c91f1ba70c4ef851ac5ba27e1b20f1d92da3ade010dd10',
|
||||
'03424f5a7601eaa47482cb17100b31a84a04d14fb44b83a57eeceffd8e299878e3',
|
||||
'032850492ee61a5f7006a2fda6925e4b4ec3782f2b6de2ff0e439ef5a38c3b2470',
|
||||
'022c80bace98831c44c32fb69755f2b353434e0ee9e7fbda29507f7ef8abea1421',
|
||||
'02c3559c833e6f99f9ca05fe503e0b4e7524dea9121344edfd3e811101e0c28680',
|
||||
'032c7c7819276c4f706a04df1a0f1e10a5495994a7be4c1d3d28ca766e5a2b957b',
|
||||
'025a7e38c2834dd843591a4d23d5f09cdeb77ddca85f673c2d944a14220ff14cf7',
|
||||
'0395e2731a1673ef21d7a16a727c4fc4d4c35a861c428ce2c819c53d2b81c8bd55',
|
||||
@@ -64,6 +70,12 @@ class NodesRoutes {
|
||||
break;
|
||||
case 'signet':
|
||||
nodesList = [
|
||||
'029fe3621fc0c6e08056a14b868f8fb9acca1aa28a129512f6cea0f0d7654d9f92',
|
||||
'02f60cd7a3a4f1c953dd9554a6ebd51a34f8b10b8124b7fc43a0b381139b55c883',
|
||||
'03cbbf581774700865eebd1be42d022bc004ba30881274ab304e088a25d70e773d',
|
||||
'0243348cb3741cfe2d8485fa8375c29c7bc7cbb67577c363cb6987a5e5fd0052cc',
|
||||
'02cb73e631af44bee600d80f8488a9194c9dc5c7590e575c421a070d1be05bc8e9',
|
||||
'0306f55ee631aa1e2cd4d9b2bfcbc14404faec5c541cef8b2e6f779061029d09c4',
|
||||
'03ddab321b760433cbf561b615ef62ac7d318630c5f51d523aaf5395b90b751956',
|
||||
'033d92c7bfd213ef1b34c90e985fb5dc77f9ec2409d391492484e57a44c4aca1de',
|
||||
'02ad010dda54253c1eb9efe38b0760657a3b43ecad62198c359c051c9d99d45781',
|
||||
@@ -86,6 +98,12 @@ class NodesRoutes {
|
||||
break;
|
||||
default:
|
||||
nodesList = [
|
||||
'02b12b889fe3c943cb05645921040ef13d6d397a2e7a4ad000e28500c505ff26d6',
|
||||
'0302240ac9d71b39617cbde2764837ec3d6198bd6074b15b75d2ff33108e89d2e1',
|
||||
'03364a8ace313376e5e4b68c954e287c6388e16df9e9fdbaf0363ecac41105cbf6',
|
||||
'03229ab4b7f692753e094b93df90530150680f86b535b5183b0cffd75b3df583fc',
|
||||
'03a696eb7acde991c1be97a58a9daef416659539ae462b897f5e9ae361f990228e',
|
||||
'0248bf26cf3a63ab8870f34dc0ec9e6c8c6288cdba96ba3f026f34ec0f13ac4055',
|
||||
'03fbc17549ec667bccf397ababbcb4cdc0e3394345e4773079ab2774612ec9be61',
|
||||
'03da9a8623241ccf95f19cd645c6cecd4019ac91570e976eb0a128bebbc4d8a437',
|
||||
'03ca5340cf85cb2e7cf076e489f785410838de174e40be62723e8a60972ad75144',
|
||||
|
||||
@@ -451,6 +451,7 @@ class MempoolBlocks {
|
||||
private processBlockTemplates(mempool: { [txid: string]: MempoolTransactionExtended }, blocks: string[][], blockWeights: number[] | null, rates: [string, number][], clusters: string[][], accelerations, accelerationPool, saveResults): MempoolBlockWithTransactions[] {
|
||||
for (const [txid, rate] of rates) {
|
||||
if (txid in mempool) {
|
||||
mempool[txid].cpfpDirty = (rate !== mempool[txid].effectiveFeePerVsize);
|
||||
mempool[txid].effectiveFeePerVsize = rate;
|
||||
mempool[txid].cpfpChecked = false;
|
||||
}
|
||||
@@ -494,6 +495,9 @@ class MempoolBlocks {
|
||||
}
|
||||
}
|
||||
});
|
||||
if (mempoolTx.ancestors?.length !== ancestors.length || mempoolTx.descendants?.length !== descendants.length) {
|
||||
mempoolTx.cpfpDirty = true;
|
||||
}
|
||||
Object.assign(mempoolTx, {ancestors, descendants, bestDescendant: null, cpfpChecked: true});
|
||||
}
|
||||
}
|
||||
@@ -531,12 +535,21 @@ class MempoolBlocks {
|
||||
|
||||
const acceleration = accelerations[txid];
|
||||
if (isAccelerated[txid] || (acceleration && (!accelerationPool || acceleration.pools.includes(accelerationPool)))) {
|
||||
if (!mempoolTx.acceleration) {
|
||||
mempoolTx.cpfpDirty = true;
|
||||
}
|
||||
mempoolTx.acceleration = true;
|
||||
for (const ancestor of mempoolTx.ancestors || []) {
|
||||
if (!mempool[ancestor.txid].acceleration) {
|
||||
mempool[ancestor.txid].cpfpDirty = true;
|
||||
}
|
||||
mempool[ancestor.txid].acceleration = true;
|
||||
isAccelerated[ancestor.txid] = true;
|
||||
}
|
||||
} else {
|
||||
if (mempoolTx.acceleration) {
|
||||
mempoolTx.cpfpDirty = true;
|
||||
}
|
||||
delete mempoolTx.acceleration;
|
||||
}
|
||||
|
||||
|
||||
@@ -174,6 +174,7 @@ class StatisticsApi {
|
||||
private getQueryForDaysAvg(div: number, interval: string) {
|
||||
return `SELECT
|
||||
UNIX_TIMESTAMP(added) as added,
|
||||
CAST(avg(unconfirmed_transactions) as DOUBLE) as unconfirmed_transactions,
|
||||
CAST(avg(vbytes_per_second) as DOUBLE) as vbytes_per_second,
|
||||
CAST(avg(min_fee) as DOUBLE) as min_fee,
|
||||
CAST(avg(vsize_1) as DOUBLE) as vsize_1,
|
||||
@@ -223,6 +224,7 @@ class StatisticsApi {
|
||||
private getQueryForDays(div: number, interval: string) {
|
||||
return `SELECT
|
||||
UNIX_TIMESTAMP(added) as added,
|
||||
CAST(avg(unconfirmed_transactions) as DOUBLE) as unconfirmed_transactions,
|
||||
CAST(avg(vbytes_per_second) as DOUBLE) as vbytes_per_second,
|
||||
CAST(avg(min_fee) as DOUBLE) as min_fee,
|
||||
vsize_1,
|
||||
@@ -406,6 +408,7 @@ class StatisticsApi {
|
||||
return statistic.map((s) => {
|
||||
return {
|
||||
added: s.added,
|
||||
count: s.unconfirmed_transactions,
|
||||
vbytes_per_second: s.vbytes_per_second,
|
||||
mempool_byte_weight: s.mempool_byte_weight,
|
||||
total_fee: s.total_fee,
|
||||
|
||||
@@ -5,6 +5,7 @@ import bitcoinApi, { bitcoinCoreApi } from './bitcoin/bitcoin-api-factory';
|
||||
import * as bitcoinjs from 'bitcoinjs-lib';
|
||||
import logger from '../logger';
|
||||
import config from '../config';
|
||||
import pLimit from '../utils/p-limit';
|
||||
|
||||
class TransactionUtils {
|
||||
constructor() { }
|
||||
@@ -74,8 +75,12 @@ class TransactionUtils {
|
||||
|
||||
public async $getMempoolTransactionsExtended(txids: string[], addPrevouts = false, lazyPrevouts = false, forceCore = false): Promise<MempoolTransactionExtended[]> {
|
||||
if (forceCore || config.MEMPOOL.BACKEND !== 'esplora') {
|
||||
const results = await Promise.allSettled(txids.map(txid => this.$getTransactionExtended(txid, addPrevouts, lazyPrevouts, forceCore, true)));
|
||||
return (results.filter(r => r.status === 'fulfilled') as PromiseFulfilledResult<MempoolTransactionExtended>[]).map(r => r.value);
|
||||
const limiter = pLimit(8); // Run 8 requests at a time
|
||||
const results = await Promise.allSettled(txids.map(
|
||||
txid => limiter(() => this.$getMempoolTransactionExtended(txid, addPrevouts, lazyPrevouts, forceCore))
|
||||
));
|
||||
return results.filter(reply => reply.status === 'fulfilled')
|
||||
.map(r => (r as PromiseFulfilledResult<MempoolTransactionExtended>).value);
|
||||
} else {
|
||||
const transactions = await bitcoinApi.$getMempoolTransactions(txids);
|
||||
return transactions.map(transaction => {
|
||||
|
||||
@@ -486,6 +486,7 @@ class WebsocketHandler {
|
||||
|
||||
// pre-compute address transactions
|
||||
const addressCache = this.makeAddressCache(newTransactions);
|
||||
const removedAddressCache = this.makeAddressCache(deletedTransactions);
|
||||
|
||||
this.wss.clients.forEach(async (client) => {
|
||||
if (client.readyState !== WebSocket.OPEN) {
|
||||
@@ -526,11 +527,15 @@ class WebsocketHandler {
|
||||
}
|
||||
|
||||
if (client['track-address']) {
|
||||
const foundTransactions = Array.from(addressCache[client['track-address']]?.values() || []);
|
||||
const newTransactions = Array.from(addressCache[client['track-address']]?.values() || []);
|
||||
const removedTransactions = Array.from(removedAddressCache[client['track-address']]?.values() || []);
|
||||
// txs may be missing prevouts in non-esplora backends
|
||||
// so fetch the full transactions now
|
||||
const fullTransactions = (config.MEMPOOL.BACKEND !== 'esplora') ? await this.getFullTransactions(foundTransactions) : foundTransactions;
|
||||
const fullTransactions = (config.MEMPOOL.BACKEND !== 'esplora') ? await this.getFullTransactions(newTransactions) : newTransactions;
|
||||
|
||||
if (removedTransactions.length) {
|
||||
response['address-removed-transactions'] = JSON.stringify(removedTransactions);
|
||||
}
|
||||
if (fullTransactions.length) {
|
||||
response['address-transactions'] = JSON.stringify(fullTransactions);
|
||||
}
|
||||
@@ -586,13 +591,25 @@ class WebsocketHandler {
|
||||
|
||||
const mempoolTx = newMempool[trackTxid];
|
||||
if (mempoolTx && mempoolTx.position) {
|
||||
response['txPosition'] = JSON.stringify({
|
||||
const positionData = {
|
||||
txid: trackTxid,
|
||||
position: {
|
||||
...mempoolTx.position,
|
||||
accelerated: mempoolTx.acceleration || undefined,
|
||||
}
|
||||
});
|
||||
};
|
||||
if (mempoolTx.cpfpDirty) {
|
||||
positionData['cpfp'] = {
|
||||
ancestors: mempoolTx.ancestors,
|
||||
bestDescendant: mempoolTx.bestDescendant || null,
|
||||
descendants: mempoolTx.descendants || null,
|
||||
effectiveFeePerVsize: mempoolTx.effectiveFeePerVsize || null,
|
||||
sigops: mempoolTx.sigops,
|
||||
adjustedVsize: mempoolTx.adjustedVsize,
|
||||
acceleration: mempoolTx.acceleration
|
||||
};
|
||||
}
|
||||
response['txPosition'] = JSON.stringify(positionData);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -93,6 +93,7 @@ interface IConfig {
|
||||
USERNAME: string;
|
||||
PASSWORD: string;
|
||||
TIMEOUT: number;
|
||||
PID_DIR: string;
|
||||
};
|
||||
SYSLOG: {
|
||||
ENABLED: boolean;
|
||||
@@ -219,6 +220,7 @@ const defaults: IConfig = {
|
||||
'USERNAME': 'mempool',
|
||||
'PASSWORD': 'mempool',
|
||||
'TIMEOUT': 180000,
|
||||
'PID_DIR': '',
|
||||
},
|
||||
'SYSLOG': {
|
||||
'ENABLED': true,
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import * as fs from 'fs';
|
||||
import path from 'path';
|
||||
import config from './config';
|
||||
import { createPool, Pool, PoolConnection } from 'mysql2/promise';
|
||||
import logger from './logger';
|
||||
@@ -101,6 +103,33 @@ import { FieldPacket, OkPacket, PoolOptions, ResultSetHeader, RowDataPacket } fr
|
||||
}
|
||||
}
|
||||
|
||||
public getPidLock(): boolean {
|
||||
const filePath = path.join(config.DATABASE.PID_DIR || __dirname, `/mempool-${config.DATABASE.DATABASE}.pid`);
|
||||
if (fs.existsSync(filePath)) {
|
||||
const pid = fs.readFileSync(filePath).toString();
|
||||
if (pid !== `${process.pid}`) {
|
||||
const msg = `Already running on PID ${pid} (or pid file '${filePath}' is stale)`;
|
||||
logger.err(msg);
|
||||
throw new Error(msg);
|
||||
} else {
|
||||
return true;
|
||||
}
|
||||
} else {
|
||||
fs.writeFileSync(filePath, `${process.pid}`);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
public releasePidLock(): void {
|
||||
const filePath = path.join(config.DATABASE.PID_DIR || __dirname, `/mempool-${config.DATABASE.DATABASE}.pid`);
|
||||
if (fs.existsSync(filePath)) {
|
||||
const pid = fs.readFileSync(filePath).toString();
|
||||
if (pid === `${process.pid}`) {
|
||||
fs.unlinkSync(filePath);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async getPool(): Promise<Pool> {
|
||||
if (this.pool === null) {
|
||||
this.pool = createPool(this.poolConfig);
|
||||
|
||||
@@ -91,11 +91,18 @@ class Server {
|
||||
async startServer(worker = false): Promise<void> {
|
||||
logger.notice(`Starting Mempool Server${worker ? ' (worker)' : ''}... (${backendInfo.getShortCommitHash()})`);
|
||||
|
||||
// Register cleanup listeners for exit events
|
||||
['exit', 'SIGINT', 'SIGTERM', 'SIGUSR1', 'SIGUSR2', 'uncaughtException', 'unhandledRejection'].forEach(event => {
|
||||
process.on(event, () => { this.onExit(event); });
|
||||
});
|
||||
|
||||
if (config.MEMPOOL.BACKEND === 'esplora') {
|
||||
bitcoinApi.startHealthChecks();
|
||||
}
|
||||
|
||||
if (config.DATABASE.ENABLED) {
|
||||
DB.getPidLock();
|
||||
|
||||
await DB.checkDbConnection();
|
||||
try {
|
||||
if (process.env.npm_config_reindex_blocks === 'true') { // Re-index requests
|
||||
@@ -306,6 +313,15 @@ class Server {
|
||||
this.lastHeapLogTime = now;
|
||||
}
|
||||
}
|
||||
|
||||
onExit(exitEvent): void {
|
||||
if (config.DATABASE.ENABLED) {
|
||||
DB.releasePidLock();
|
||||
}
|
||||
process.exit(0);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
((): Server => new Server())();
|
||||
|
||||
@@ -104,6 +104,7 @@ export interface MempoolTransactionExtended extends TransactionExtended {
|
||||
adjustedFeePerVsize: number;
|
||||
inputs?: number[];
|
||||
lastBoosted?: number;
|
||||
cpfpDirty?: boolean;
|
||||
}
|
||||
|
||||
export interface AuditTransaction {
|
||||
|
||||
179
backend/src/utils/p-limit.ts
Normal file
179
backend/src/utils/p-limit.ts
Normal file
@@ -0,0 +1,179 @@
|
||||
/*
|
||||
MIT License
|
||||
|
||||
Copyright (c) Sindre Sorhus <sindresorhus@gmail.com> (https://sindresorhus.com)
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy of this
|
||||
software and associated documentation files (the "Software"), to deal in the Software
|
||||
without restriction, including without limitation the rights to use, copy, modify,
|
||||
merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
|
||||
permit persons to whom the Software is furnished to do so, subject to the following
|
||||
conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all copies
|
||||
or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
|
||||
INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
|
||||
PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
|
||||
CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE
|
||||
OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
|
||||
/*
|
||||
How it works:
|
||||
`this._head` is an instance of `Node` which keeps track of its current value and nests
|
||||
another instance of `Node` that keeps the value that comes after it. When a value is
|
||||
provided to `.enqueue()`, the code needs to iterate through `this._head`, going deeper
|
||||
and deeper to find the last value. However, iterating through every single item is slow.
|
||||
This problem is solved by saving a reference to the last value as `this._tail` so that
|
||||
it can reference it to add a new value.
|
||||
*/
|
||||
|
||||
class Node {
|
||||
value;
|
||||
next;
|
||||
|
||||
constructor(value) {
|
||||
this.value = value;
|
||||
}
|
||||
}
|
||||
|
||||
class Queue {
|
||||
private _head;
|
||||
private _tail;
|
||||
private _size;
|
||||
|
||||
constructor() {
|
||||
this.clear();
|
||||
}
|
||||
|
||||
enqueue(value) {
|
||||
const node = new Node(value);
|
||||
|
||||
if (this._head) {
|
||||
this._tail.next = node;
|
||||
this._tail = node;
|
||||
} else {
|
||||
this._head = node;
|
||||
this._tail = node;
|
||||
}
|
||||
|
||||
this._size++;
|
||||
}
|
||||
|
||||
dequeue() {
|
||||
const current = this._head;
|
||||
if (!current) {
|
||||
return;
|
||||
}
|
||||
|
||||
this._head = this._head.next;
|
||||
this._size--;
|
||||
return current.value;
|
||||
}
|
||||
|
||||
clear() {
|
||||
this._head = undefined;
|
||||
this._tail = undefined;
|
||||
this._size = 0;
|
||||
}
|
||||
|
||||
get size() {
|
||||
return this._size;
|
||||
}
|
||||
|
||||
*[Symbol.iterator]() {
|
||||
let current = this._head;
|
||||
|
||||
while (current) {
|
||||
yield current.value;
|
||||
current = current.next;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
interface LimitFunction {
|
||||
readonly activeCount: number;
|
||||
readonly pendingCount: number;
|
||||
clearQueue: () => void;
|
||||
<Arguments extends unknown[], ReturnType>(
|
||||
fn: (...args: Arguments) => PromiseLike<ReturnType> | ReturnType,
|
||||
...args: Arguments
|
||||
): Promise<ReturnType>;
|
||||
}
|
||||
|
||||
export default function pLimit(concurrency: number): LimitFunction {
|
||||
if (
|
||||
!(
|
||||
(Number.isInteger(concurrency) ||
|
||||
concurrency === Number.POSITIVE_INFINITY) &&
|
||||
concurrency > 0
|
||||
)
|
||||
) {
|
||||
throw new TypeError('Expected `concurrency` to be a number from 1 and up');
|
||||
}
|
||||
|
||||
const queue = new Queue();
|
||||
let activeCount = 0;
|
||||
|
||||
const next = () => {
|
||||
activeCount--;
|
||||
|
||||
if (queue.size > 0) {
|
||||
queue.dequeue()();
|
||||
}
|
||||
};
|
||||
|
||||
const run = async (fn, resolve, args) => {
|
||||
activeCount++;
|
||||
|
||||
const result = (async () => fn(...args))();
|
||||
|
||||
resolve(result);
|
||||
|
||||
try {
|
||||
await result;
|
||||
} catch {}
|
||||
|
||||
next();
|
||||
};
|
||||
|
||||
const enqueue = (fn, resolve, args) => {
|
||||
queue.enqueue(run.bind(undefined, fn, resolve, args));
|
||||
|
||||
(async () => {
|
||||
// This function needs to wait until the next microtask before comparing
|
||||
// `activeCount` to `concurrency`, because `activeCount` is updated asynchronously
|
||||
// when the run function is dequeued and called. The comparison in the if-statement
|
||||
// needs to happen asynchronously as well to get an up-to-date value for `activeCount`.
|
||||
await Promise.resolve();
|
||||
|
||||
if (activeCount < concurrency && queue.size > 0) {
|
||||
queue.dequeue()();
|
||||
}
|
||||
})();
|
||||
};
|
||||
|
||||
const generator = (fn, ...args) =>
|
||||
new Promise((resolve) => {
|
||||
enqueue(fn, resolve, args);
|
||||
});
|
||||
|
||||
Object.defineProperties(generator, {
|
||||
activeCount: {
|
||||
get: () => activeCount,
|
||||
},
|
||||
pendingCount: {
|
||||
get: () => queue.size,
|
||||
},
|
||||
clearQueue: {
|
||||
value: () => {
|
||||
queue.clear();
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
return generator as any;
|
||||
}
|
||||
Reference in New Issue
Block a user