improved processing queue

This commit is contained in:
Pierre
2025-10-31 13:52:16 +01:00
parent 3ecf91a84b
commit 7edb1d146c
2 changed files with 12 additions and 2 deletions
@@ -8,4 +8,7 @@ export interface FileProcessorInterface {
deleteFile?(filePath: string): Promise<void>; deleteFile?(filePath: string): Promise<void>;
finalize?(): Promise<void>; finalize?(): Promise<void>;
/** Called when a file is queued for processing (before it's added to the processing queue) */
queueFile?(filePath: string): void;
} }
+9 -2
View File
@@ -7,7 +7,7 @@ import {FileProcessorInterface} from "./FileProcessorInterface.js";
export class FolderWatcher { export class FolderWatcher {
initialized = false; initialized = false;
queue = new PQueue({concurrency: 100, autoStart: true}); queue: PQueue;
fileProcessor: FileProcessorInterface; fileProcessor: FileProcessorInterface;
queueSize = 0; queueSize = 0;
current = 0; current = 0;
@@ -18,8 +18,11 @@ export class FolderWatcher {
interval?: number, interval?: number,
stabilityThreshold?: number, stabilityThreshold?: number,
pollInterval?: number, pollInterval?: number,
outputFolderUpdateIntervalMs?: number outputFolderUpdateIntervalMs?: number,
concurrency?: number
}) { }) {
// Initialize queue with configurable concurrency (default 4 to match typical CPU count)
this.queue = new PQueue({concurrency: config.concurrency || 4, autoStart: true});
this.fileProcessor = fileProcessor; this.fileProcessor = fileProcessor;
if (this.fileProcessor.finalize) { if (this.fileProcessor.finalize) {
const debouncedImmediate = debounce(() => this.fileProcessor.finalize(), this.config.outputFolderUpdateIntervalMs || 10000, {immediate: true}); const debouncedImmediate = debounce(() => this.fileProcessor.finalize(), this.config.outputFolderUpdateIntervalMs || 10000, {immediate: true});
@@ -82,6 +85,10 @@ export class FolderWatcher {
if (!this.processing.has(filePath)) { if (!this.processing.has(filePath)) {
this.processing.add(filePath);//avoid double processing in the queue this.processing.add(filePath);//avoid double processing in the queue
if (await this.fileProcessor.canProcessFile(filePath)) { if (await this.fileProcessor.canProcessFile(filePath)) {
// Notify that file is being queued (before adding to queue)
if (this.fileProcessor.queueFile) {
this.fileProcessor.queueFile(filePath);
}
const current = ++this.current; const current = ++this.current;
await this.queue.add(() => this.fileProcessor.processFile(current, this.queueSize, filePath)); await this.queue.add(() => this.fileProcessor.processFile(current, this.queueSize, filePath));
} }