Creepso/webstreamr-github
1
0
Fork 0
Code Issues 27 Pull requests 7 Projects Releases 240 Packages Wiki Activity Actions

refactor(fetcher): introduce CustomRequestInit

Browse source
This commit is contained in:
WebStreamr 2025-06-05 10:17:03 +00:00
parent 092ee4bfc2
commit 63a53a18cc
No known key found for this signature in database
2 changed files with 22 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
src/utils/Fetcher.test.ts
View file

@ -5,7 +5,7 @@ import { Context } from '../types';
import { BlockedError, NotFoundError, QueueIsFullError } from '../error';
fetchMock.mockGlobal();
const fetcher = new Fetcher(winston.createLogger({ transports: [new winston.transports.Console({ level: 'nope' })] }), 3, 10);
const fetcher = new Fetcher(winston.createLogger({ transports: [new winston.transports.Console({ level: 'nope' })] }));
describe('fetch', () => {
const ctx: Context = { id: 'id', ip: '127.0.0.1', config: { de: 'on' } };
@ -156,20 +156,20 @@ describe('fetch', () => {
await expect(fetcher.text(ctx, new URL('https://some-exception-url.test/'))).rejects.toBeInstanceOf(TypeError);
});
test('times out after 10 seconds', async () => {
test('times out', async () => {
fetchMock.get('https://some-timeout-url.test/', 200, { delay: 20 });
await expect(fetcher.text(ctx, new URL('https://some-timeout-url.test/'))).rejects.toBeInstanceOf(DOMException);
await expect(fetcher.text(ctx, new URL('https://some-timeout-url.test/'), { timeout: 10 })).rejects.toBeInstanceOf(DOMException);
});
test('full queue throws an error', async () => {
fetchMock.get('https://some-full-queue-url.test/', 'some text');
const allPromises = Promise.all([
fetcher.text(ctx, new URL('https://some-full-queue-url.test/')),
fetcher.text(ctx, new URL('https://some-full-queue-url.test/')),
fetcher.text(ctx, new URL('https://some-full-queue-url.test/')),
fetcher.text(ctx, new URL('https://some-full-queue-url.test/')),
fetcher.text(ctx, new URL('https://some-full-queue-url.test/'), { queueLimit: 3 }),
fetcher.text(ctx, new URL('https://some-full-queue-url.test/'), { queueLimit: 3 }),
fetcher.text(ctx, new URL('https://some-full-queue-url.test/'), { queueLimit: 3 }),
fetcher.text(ctx, new URL('https://some-full-queue-url.test/'), { queueLimit: 3 }),
]);
await expect(allPromises).rejects.toBeInstanceOf(QueueIsFullError);

29
src/utils/Fetcher.ts
View file

@ -44,37 +44,38 @@ interface FlareSolverrResult {
version: string;
}
type CustomRequestInit = RequestInit & {
queueLimit?: number;
timeout?: number;
};
export class Fetcher {
private readonly logger: winston.Logger;
private readonly queueLimit: number;
private readonly timeout: number;
private readonly httpCache: TTLCache<string, HttpCacheItem>;
private readonly hostQueues = new Map<string, HostQueue>();
private readonly hostUserAgentMap = new Map<string, string>();
private readonly cookieJar = new CookieJar();
constructor(logger: winston.Logger, queueLimit?: number, timeout?: number) {
constructor(logger: winston.Logger) {
this.logger = logger;
this.queueLimit = queueLimit ?? 10;
this.timeout = timeout ?? 10000;
this.httpCache = new TTLCache();
}
readonly text = async (ctx: Context, url: URL, init?: RequestInit): Promise<string> => {
readonly text = async (ctx: Context, url: URL, init?: CustomRequestInit): Promise<string> => {
return (await this.cachedFetch(ctx, url, init)).body;
};
readonly textPost = async (ctx: Context, url: URL, data: unknown, init?: RequestInit): Promise<string> => {
readonly textPost = async (ctx: Context, url: URL, data: unknown, init?: CustomRequestInit): Promise<string> => {
return (await this.cachedFetch(ctx, url, { ...init, method: 'POST', body: JSON.stringify(data) })).body;
};
readonly head = async (ctx: Context, url: URL, init?: RequestInit): Promise<CachePolicy.Headers> => {
readonly head = async (ctx: Context, url: URL, init?: CustomRequestInit): Promise<CachePolicy.Headers> => {
return (await this.cachedFetch(ctx, url, { ...init, method: 'HEAD' })).policy.responseHeaders();
};
public readonly getInit = (ctx: Context, url: URL, init?: RequestInit): RequestInit => {
public readonly getInit = (ctx: Context, url: URL, init?: CustomRequestInit): CustomRequestInit => {
const cookieString = this.cookieJar.getCookieStringSync(url.href);
return {
@ -178,7 +179,7 @@ export class Fetcher {
return obj;
};
private readonly cachedFetch = async (ctx: Context, url: URL, init?: RequestInit): Promise<HttpCacheItem> => {
private readonly cachedFetch = async (ctx: Context, url: URL, init?: CustomRequestInit): Promise<HttpCacheItem> => {
const newInit = this.getInit(ctx, url, init);
const request: CachePolicy.Request = { url: url.href, method: newInit.method ?? 'GET', headers: {} };
@ -204,11 +205,11 @@ export class Fetcher {
return this.handleHttpCacheItem(ctx, httpCacheItem, url);
};
private readonly fetchWithTimeout = async (ctx: Context, url: URL, init?: RequestInit): Promise<Response> => {
private readonly fetchWithTimeout = async (ctx: Context, url: URL, init?: CustomRequestInit): Promise<Response> => {
this.logger.info(`Fetch ${init?.method ?? 'GET'} ${url}`, ctx);
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(TIMEOUT), this.timeout);
const timer = setTimeout(() => controller.abort(TIMEOUT), init?.timeout ?? 10000);
let response;
try {
@ -220,7 +221,7 @@ export class Fetcher {
return response;
};
private readonly queuedFetch = async (ctx: Context, url: URL, init?: RequestInit): Promise<Response> => {
private readonly queuedFetch = async (ctx: Context, url: URL, init?: CustomRequestInit): Promise<Response> => {
let queue = this.hostQueues.get(url.host);
if (!queue) {
queue = { promise: Promise.resolve(), count: 0 };
@ -229,7 +230,7 @@ export class Fetcher {
queue.count++;
if (queue.count > this.queueLimit) {
if (queue.count > (init?.queueLimit ?? 10)) {
throw new QueueIsFullError();
}