98 lines
3.1 KiB
JavaScript
98 lines
3.1 KiB
JavaScript
|
"use strict";
|
||
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
||
|
|
const events_1 = require("events");
|
||
|
|
const fsScandir = require("@nodelib/fs.scandir");
|
||
|
|
const fastq = require("fastq");
|
||
|
|
const common = require("./common");
|
||
|
|
const reader_1 = require("./reader");
|
||
|
|
class AsyncReader extends reader_1.default {
|
||
|
|
constructor(_root, _settings) {
|
||
|
|
super(_root, _settings);
|
||
|
|
this._settings = _settings;
|
||
|
|
this._scandir = fsScandir.scandir;
|
||
|
|
this._emitter = new events_1.EventEmitter();
|
||
|
|
this._queue = fastq(this._worker.bind(this), this._settings.concurrency);
|
||
|
|
this._isFatalError = false;
|
||
|
|
this._isDestroyed = false;
|
||
|
|
this._queue.drain = () => {
|
||
|
|
if (!this._isFatalError) {
|
||
|
|
this._emitter.emit('end');
|
||
|
|
}
|
||
|
|
};
|
||
|
|
}
|
||
|
|
read() {
|
||
|
|
this._isFatalError = false;
|
||
|
|
this._isDestroyed = false;
|
||
|
|
setImmediate(() => {
|
||
|
|
this._pushToQueue(this._root, this._settings.basePath);
|
||
|
|
});
|
||
|
|
return this._emitter;
|
||
|
|
}
|
||
|
|
get isDestroyed() {
|
||
|
|
return this._isDestroyed;
|
||
|
|
}
|
||
|
|
destroy() {
|
||
|
|
if (this._isDestroyed) {
|
||
|
|
throw new Error('The reader is already destroyed');
|
||
|
|
}
|
||
|
|
this._isDestroyed = true;
|
||
|
|
this._queue.killAndDrain();
|
||
|
|
}
|
||
|
|
onEntry(callback) {
|
||
|
|
this._emitter.on('entry', callback);
|
||
|
|
}
|
||
|
|
onError(callback) {
|
||
|
|
this._emitter.once('error', callback);
|
||
|
|
}
|
||
|
|
onEnd(callback) {
|
||
|
|
this._emitter.once('end', callback);
|
||
|
|
}
|
||
|
|
_pushToQueue(directory, base) {
|
||
|
|
const queueItem = { directory, base };
|
||
|
|
this._queue.push(queueItem, (error) => {
|
||
|
|
if (error !== null) {
|
||
|
|
this._handleError(error);
|
||
|
|
}
|
||
|
|
});
|
||
|
|
}
|
||
|
|
_worker(item, done) {
|
||
|
|
this._scandir(item.directory, this._settings.fsScandirSettings, (error, entries) => {
|
||
|
|
if (error !== null) {
|
||
|
|
done(error, undefined);
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
for (const entry of entries) {
|
||
|
|
this._handleEntry(entry, item.base);
|
||
|
|
}
|
||
|
|
done(null, undefined);
|
||
|
|
});
|
||
|
|
}
|
||
|
|
_handleError(error) {
|
||
|
|
if (this._isDestroyed || !common.isFatalError(this._settings, error)) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
this._isFatalError = true;
|
||
|
|
this._isDestroyed = true;
|
||
|
|
this._emitter.emit('error', error);
|
||
|
|
}
|
||
|
|
_handleEntry(entry, base) {
|
||
|
|
if (this._isDestroyed || this._isFatalError) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
const fullpath = entry.path;
|
||
|
|
if (base !== undefined) {
|
||
|
|
entry.path = common.joinPathSegments(base, entry.name, this._settings.pathSegmentSeparator);
|
||
|
|
}
|
||
|
|
if (common.isAppliedFilter(this._settings.entryFilter, entry)) {
|
||
|
|
this._emitEntry(entry);
|
||
|
|
}
|
||
|
|
if (entry.dirent.isDirectory() && common.isAppliedFilter(this._settings.deepFilter, entry)) {
|
||
|
|
this._pushToQueue(fullpath, base === undefined ? undefined : entry.path);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
_emitEntry(entry) {
|
||
|
|
this._emitter.emit('entry', entry);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
exports.default = AsyncReader;
|