improve queue library

This commit is contained in:
Pierre
2025-11-01 21:39:23 +01:00
parent 7edb1d146c
commit af278633c4
2 changed files with 56 additions and 7 deletions
@@ -1,14 +1,50 @@
/**
* Interface for file processing implementations used by FolderWatcher.
*
* Provides hooks for different stages of the file processing pipeline:
* 1. markPending() - File discovered, waiting in queue
* 2. queueFile() - File starts processing (optional pre-processing phase)
* 3. processFile() - Main processing phase
* 4. finalize() - After all files processed (optional)
*/
export interface FileProcessorInterface {
/** Optional initialization before processing starts */
initialize?(): Promise<void>;
/**
* Main file processing method.
* @param current - Current file index in the queue
* @param queueSize - Total number of files to process
* @param nfoFilePath - Path to the file being processed
*/
processFile(current: number, queueSize: number, nfoFilePath: string): Promise<void>;
/**
* Determines if a file should be processed.
* @param filePath - Path to the file to check
* @returns true if the file should be processed
*/
canProcessFile(filePath: string): Promise<boolean>;
/** Optional callback when a file is deleted */
deleteFile?(filePath: string): Promise<void>;
/** Optional callback after all files have been processed */
finalize?(): Promise<void>;
/** Called when a file is queued for processing (before it's added to the processing queue) */
queueFile?(filePath: string): void;
/**
* Optional callback when a file is discovered and marked as pending.
* Called BEFORE the file enters the processing queue.
* Useful for tracking/monitoring pending files.
* @param filePath - Path to the file being marked as pending
*/
markPending?(filePath: string): void;
/**
* Optional callback when a file begins processing.
* Called WITHIN the queue (respects concurrency limits).
* Can be used for quick/light pre-processing before main processing.
* @param filePath - Path to the file being queued
*/
queueFile?(filePath: string): Promise<void>;
}
+18 -5
View File
@@ -85,12 +85,25 @@ export class FolderWatcher {
if (!this.processing.has(filePath)) {
this.processing.add(filePath);//avoid double processing in the queue
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;
await this.queue.add(() => this.fileProcessor.processFile(current, this.queueSize, filePath));
// Notify implementation that file is pending (optional callback)
// Allows tracking of files before they enter the processing queue
if (this.fileProcessor.markPending) {
this.fileProcessor.markPending(filePath);
}
// Add file to processing queue (respects concurrency limit)
// Both queueFile() and processFile() are executed within the queue
// to ensure controlled concurrency across the entire processing pipeline
await this.queue.add(async () => {
// Optional: Quick/light pre-processing phase
if (this.fileProcessor.queueFile) {
await this.fileProcessor.queueFile(filePath);
}
// Main processing phase
await this.fileProcessor.processFile(current, this.queueSize, filePath);
});
}
}
this.processing.delete(filePath);