boki/stock-bot
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

added p-limit instead ot batch to proxy-service

Browse source
This commit is contained in:
Bojan Kucera 2025-06-07 17:05:14 -04:00
parent 2ae1e73661
commit baa34a3805
3 changed files with 27 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

16
apps/data-service/package.json
View file

@ -10,17 +10,19 @@
"start": "bun dist/index.js",
"test": "bun test",
"clean": "rm -rf dist"
}, "dependencies": {
},
"dependencies": {
"@stock-bot/cache": "*",
"@stock-bot/config": "*",
"@stock-bot/logger": "*",
"@stock-bot/shutdown": "*",
"@stock-bot/types": "*",
"@stock-bot/questdb-client": "*",
"@stock-bot/mongodb-client": "*",
"@stock-bot/event-bus": "*",
"@stock-bot/http": "*",
"@stock-bot/cache": "*",
"@stock-bot/logger": "*",
"@stock-bot/mongodb-client": "*",
"@stock-bot/questdb-client": "*",
"@stock-bot/shutdown": "*",
"@stock-bot/types": "*",
"hono": "^4.0.0",
"p-limit": "^6.2.0",
"ws": "^8.0.0"
},
"devDependencies": {

35
apps/data-service/src/services/proxy.service.ts
View file

@ -1,6 +1,7 @@
import { Logger } from '@stock-bot/logger';
import createCache, { type CacheProvider } from '@stock-bot/cache';
import { HttpClient, HttpClientConfig, ProxyConfig , RequestConfig } from '@stock-bot/http';
import pLimit from 'p-limit';
export interface ProxySource {
url: string;
@ -35,6 +36,7 @@ export class ProxyService {
private logger;
private cache: CacheProvider;
private httpClient: HttpClient;
private readonly concurrencyLimit = pLimit(200);
private readonly CACHE_PREFIX = 'proxy:';
private readonly WORKING_PROXIES_KEY = 'proxy:working';
private readonly PROXY_STATS_KEY = 'proxy:stats';
@ -434,16 +436,14 @@ export class ProxyService {
};
}
}
/**
* Validate proxies in background
*/
private async validateProxiesInBackground(proxies: ProxyConfig[]): Promise<void> {
this.logger.info('Starting background proxy validation', { count: proxies.length });
const concurrency = 50; // Process 50 proxies concurrently
const chunks = this.chunkArray(proxies, concurrency);
for (const chunk of chunks) {
const validationPromises = chunk.map(proxy =>
const validationPromises = proxies.map(proxy =>
this.concurrencyLimit(() =>
this.checkProxy(proxy).catch(error => {
this.logger.error('Error validating proxy', {
host: proxy.host,
@ -452,15 +452,12 @@ export class ProxyService {
});
return null;
})
);
await Promise.allSettled(validationPromises);
// Small delay between chunks to avoid overwhelming the system
await new Promise(resolve => setTimeout(resolve, 1000));
}
)
);
await Promise.allSettled(validationPromises);
this.logger.info('Background proxy validation completed');
}
/**
* Start periodic proxy health checks
*/
@ -470,7 +467,9 @@ export class ProxyService {
setInterval(async () => {
try {
const workingProxies = await this.getWorkingProxies(100); // Check up to 100 working proxies
const validationPromises = workingProxies.map(proxy => this.checkProxy(proxy));
const validationPromises = workingProxies.map(proxy =>
this.concurrencyLimit(() => this.checkProxy(proxy))
);
const results = await Promise.allSettled(validationPromises);
const successCount = results.filter(r =>
@ -502,7 +501,6 @@ export class ProxyService {
this.logger.error('Error clearing proxy data', error);
}
}
/**
* Get cache key for a proxy
*/
@ -510,17 +508,6 @@ export class ProxyService {
return `${this.CACHE_PREFIX}${proxy.host}:${proxy.port}`;
}
/**
* Split array into chunks
*/
private chunkArray<T>(array: T[], size: number): T[][] {
const chunks: T[][] = [];
for (let i = 0; i < array.length; i += size) {
chunks.push(array.slice(i, i + size));
}
return chunks;
}
/**
* Graceful shutdown
*/