add finalize method to folder watch

This commit is contained in:
Pierre-Henri
2024-04-19 19:44:09 +02:00
parent 257db022f8
commit 015361c249
7 changed files with 65 additions and 13 deletions
-2
View File
@@ -1,2 +0,0 @@
nmHoistingLimits: dependencies
nodeLinker: node-modules
+2 -2
View File
@@ -31,8 +31,7 @@
} }
}, },
"scripts": { "scripts": {
"build": "cross-env rimraf ./dist & tsup", "build": "tsup",
"ci": "yarn run coverage",
"test": "vitest" "test": "vitest"
}, },
"devDependencies": { "devDependencies": {
@@ -49,6 +48,7 @@
"crc-32": "^1.2.2", "crc-32": "^1.2.2",
"csv-parse": "^5.5.5", "csv-parse": "^5.5.5",
"csv-stringify": "^6.4.6", "csv-stringify": "^6.4.6",
"debounce": "^2.0.0",
"multiformats": "^13.1.0", "multiformats": "^13.1.0",
"p-queue": "^8.0.1", "p-queue": "^8.0.1",
"piscina": "^4.4.0" "piscina": "^4.4.0"
@@ -3,5 +3,7 @@ export interface FileProcessorInterface {
canProcessFile(filePath: string): Promise<boolean>; canProcessFile(filePath: string): Promise<boolean>;
deleteFile(filePath: string): Promise<void> deleteFile?(filePath: string): Promise<void>;
finalize?(): Promise<void>;
} }
+19 -1
View File
@@ -2,6 +2,7 @@ import * as path from 'path';
import chokidar from 'chokidar'; import chokidar from 'chokidar';
import {readdir} from "fs/promises"; import {readdir} from "fs/promises";
import PQueue from "p-queue"; import PQueue from "p-queue";
import debounce from 'debounce';
import {FileProcessorInterface} from "./FileProcessorInterface.js"; import {FileProcessorInterface} from "./FileProcessorInterface.js";
export class FolderWatcher { export class FolderWatcher {
@@ -11,13 +12,24 @@ export class FolderWatcher {
queueSize = -1; queueSize = -1;
current = 0; current = 0;
processing = new Set<string>(); processing = new Set<string>();
private readonly updateDebounced: () => Promise<void>;
constructor(fileProcessor:FileProcessorInterface,private WATCH_FOLDER_LIST: string, private config:{ constructor(fileProcessor:FileProcessorInterface,private WATCH_FOLDER_LIST: string, private config:{
interval?: number, interval?: number,
stabilityThreshold?: number, stabilityThreshold?: number,
pollInterval?: number, pollInterval?: number,
outputFolderUpdateIntervalMs?: number
}) { }) {
this.fileProcessor = fileProcessor; this.fileProcessor = fileProcessor;
if(this.fileProcessor.finalize) {
const debounced = debounce(() => this.fileProcessor.finalize(), this.config.outputFolderUpdateIntervalMs || 10000, {immediate: true});
this.updateDebounced = async () => {
if (this.initialized) {
return debounced();
}
}
}
} }
/** /**
@@ -108,6 +120,7 @@ export class FolderWatcher {
.on('add', async (filePath) => { .on('add', async (filePath) => {
try { try {
await this.processFileExtended(filePath); await this.processFileExtended(filePath);
await this.updateDebounced();
} catch (error) { } catch (error) {
console.error(`Error processing file ${filePath}:`, error); console.error(`Error processing file ${filePath}:`, error);
} }
@@ -115,14 +128,18 @@ export class FolderWatcher {
.on('change', async (filePath) => { .on('change', async (filePath) => {
try { try {
await this.processFileExtended(filePath); await this.processFileExtended(filePath);
await this.updateDebounced();
} catch (e) { } catch (e) {
console.error(`Error processing file ${filePath}:`, e); console.error(`Error processing file ${filePath}:`, e);
} }
}) })
.on('unlink', async (filePath) => { .on('unlink', async (filePath) => {
try { try {
await this.fileProcessor.deleteFile(filePath); if(this.fileProcessor.deleteFile) {
await this.fileProcessor.deleteFile(filePath);
}
await this.processFileExtended(filePath); await this.processFileExtended(filePath);
await this.updateDebounced();
} catch (e) { } catch (e) {
console.error(`Error processing file deletion:`, e); console.error(`Error processing file deletion:`, e);
} }
@@ -158,6 +175,7 @@ export class FolderWatcher {
} }
await Promise.all(promises); await Promise.all(promises);
this.initialized = true; this.initialized = true;
await this.updateDebounced();
console.log(`Watcher ready`); console.log(`Watcher ready`);
} }
+31
View File
@@ -0,0 +1,31 @@
import {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from "./MultiHashData";
import {Readable} from "stream";
import {computeCIDs} from "@root/file-id/ComputeHash";
import fs from "fs";
export class HashComputerFile implements ComputeInterface{
constructor(private targetHash: CID_ALGORITHM_NAMES[]) {
}
public async computeMissingHash(filePath: string, metadata: MultiHashData): Promise<void> {
// Dynamically determine which hashes are needed
const neededHashes = this.targetHash.filter(hashName => !metadata[hashName]);
// If all hashes are already computed, skip the processing
if (neededHashes.length === 0) {
return;
}
// Compute only the needed CIDs
const stream = fs.createReadStream(filePath);
const cids = await computeCIDs({stream, algorithms:neededHashes});
// Map the computed CIDs back to their respective metadata properties
for (const [index, cid] of cids.entries()) {
const hashType = neededHashes[index];
metadata[hashType] = cid;
}
}
}
+8 -7
View File
@@ -42,7 +42,8 @@ export class HashIndexManager {
} }
/** /**
* After init consseutively calls to this method will not reload the index * After init consecutively calls to this method will not reload the index
* readThe index for the first time and enable autosave
* @param autosave * @param autosave
*/ */
public async init(autosave = true) { public async init(autosave = true) {
@@ -56,7 +57,7 @@ export class HashIndexManager {
await this.loadIndex(hash); await this.loadIndex(hash);
} }
if (autosave) { if (autosave) {
this.start(); this.startAutoSave()
} }
resolve(); resolve();
} catch (e) { } catch (e) {
@@ -91,17 +92,13 @@ export class HashIndexManager {
return requiredHeaders.every(header => headers.includes(header)); return requiredHeaders.every(header => headers.includes(header));
}; };
public start() {
this.startAutoSave(this.intervalTime);
}
public stopAutoSave() { public stopAutoSave() {
if (this.intervalId) { if (this.intervalId) {
clearInterval(this.intervalId); clearInterval(this.intervalId);
} }
} }
private startAutoSave(time: number) { public startAutoSave(time: number = this.intervalTime) {
this.stopAutoSave(); this.stopAutoSave();
this.intervalTime = time; this.intervalTime = time;
this.intervalId = setInterval(() => this.saveCacheToFile(), time); this.intervalId = setInterval(() => this.saveCacheToFile(), time);
@@ -216,6 +213,10 @@ export class HashIndexManager {
this.indexOpsInProgress = false; this.indexOpsInProgress = false;
} }
/**
* this.getCidForFile(fileName, stats.size, stats.mtime.toISOString())
* @param filePath
*/
public async getCidForFileAsync(filePath: string): Promise<IndexLine> { public async getCidForFileAsync(filePath: string): Promise<IndexLine> {
const fileName = path.basename(filePath); const fileName = path.basename(filePath);
const stats = await stat(filePath); const stats = await stat(filePath);
+2
View File
@@ -6,6 +6,8 @@ export * from './file-id/ShaComputeWorker.js';//worker
export * from './hash-compute/HashComputerIndexCache.js'; export * from './hash-compute/HashComputerIndexCache.js';
export * from './hash-compute/HashComputerWorker.js'; export * from './hash-compute/HashComputerWorker.js';
export * from './hash-compute/HashIndexManager.js'; export * from './hash-compute/HashIndexManager.js';
export * from './hash-compute/HashComputer.js';
export * from './hash-compute/HashComputerFile.js';
//folder-watcher //folder-watcher
export * from './folder-watcher/FolderWatcher.js'; export * from './folder-watcher/FolderWatcher.js';