import { EventEmitter } from 'events' ; import * as fsScandir from '@nodelib/fs.scandir' ; import * as fastq from 'fastq' ; import * as common from './common' ; import Reader from './reader' ; import type Settings from '../settings' ; import type { Entry , Errno , QueueItem } from '../types' ; type EntryEventCallback = ( entry : Entry ) => void ; type ErrorEventCallback = ( error : Errno ) => void ; type EndEventCallback = ( ) => void ; export default class AsyncReader extends Reader { protected readonly _scandir : typeof fsScandir . scandir = fsScandir . scandir ; protected readonly _emitter : EventEmitter = new EventEmitter ( ) ; private readonly _queue : fastq . queue = fastq ( this . _worker . bind ( this ) , this . _settings . concurrency ) ; private _isFatalError : boolean = false ; private _isDestroyed : boolean = false ; constructor ( _root , protected override readonly _settings ) { super ( _root , _settings ) ; this . _queue . drain = ( ) => { if ( ! this . _isFatalError ) { this . _emitter . emit ( 'end' ) ; } } ; } public read ( ) { this . _isFatalError = false ; this . _isDestroyed = false ; setImmediate ( ( ) => { this . _pushToQueue ( this . _root , this . _settings . basePath ) ; } ) ; return this . _emitter ; } public get isDestroyed ( ) : boolean { return this . _isDestroyed ; } public destroy ( ) { if ( this . _isDestroyed ) { throw new Error ( 'The reader is already destroyed' ) ; } this . _isDestroyed = true ; this . _queue . killAndDrain ( ) ; } public onEntry ( callback ) { this . _emitter . on ( 'entry' , callback ) ; } public onError ( callback ) { this . _emitter . once ( 'error' , callback ) ; } public onEnd ( callback ) { this . _emitter . once ( 'end' , callback ) ; } private _pushToQueue ( directory , base ) { const queueItem = { directory , base } ; this . _queue . push ( queueItem , ( error ) => { if ( error !== null ) { this . _handleError ( error ) ; } } ) ; } private _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 as unknown as Error , undefined ) ; } ) ; } private _handleError ( error ) { if ( this . _isDestroyed || ! common . isFatalError ( this . _settings , error ) ) { return ; } this . _isFatalError = true ; this . _isDestroyed = true ; this . _emitter . emit ( 'error' , error ) ; } private _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 ) ; } } private _emitEntry ( entry ) { this . _emitter . emit ( 'entry' , entry ) ; } }