node_modules/@mrmlnc/readdir-enhanced/lib/directory-reader.js - nifi-fds - Git at Google

 'use strict';

 const Readable = require('stream').Readable;
 const EventEmitter = require('events').EventEmitter;
 const path = require('path');
 const normalizeOptions = require('./normalize-options');
 const stat = require('./stat');
 const call = require('./call');

 /**
  * Asynchronously reads the contents of a directory and streams the results
  * via a {@link stream.Readable}.
  */
 class DirectoryReader {
   /**
    * @param {string} dir - The absolute or relative directory path to read
    * @param {object} [options] - User-specified options, if any (see {@link normalizeOptions})
    * @param {object} internalOptions - Internal options that aren't part of the public API
    * @class
    */
   constructor (dir, options, internalOptions) {
     this.options = options = normalizeOptions(options, internalOptions);

     // Indicates whether we should keep reading
     // This is set false if stream.Readable.push() returns false.
     this.shouldRead = true;

     // The directories to read
     // (initialized with the top-level directory)
     this.queue = [{
       path: dir,
       basePath: options.basePath,
       posixBasePath: options.posixBasePath,
       depth: 0
     }];

     // The number of directories that are currently being processed
     this.pending = 0;

     // The data that has been read, but not yet emitted
     this.buffer = [];

     this.stream = new Readable({ objectMode: true });
     this.stream._read = () => {
       // Start (or resume) reading
       this.shouldRead = true;

       // If we have data in the buffer, then send the next chunk
       if (this.buffer.length > 0) {
         this.pushFromBuffer();
       }

       // If we have directories queued, then start processing the next one
       if (this.queue.length > 0) {
         if (this.options.facade.sync) {
           while (this.queue.length > 0) {
             this.readNextDirectory();
           }
         }
         else {
           this.readNextDirectory();
         }
       }

       this.checkForEOF();
     };
   }

   /**
    * Reads the next directory in the queue
    */
   readNextDirectory () {
     let facade = this.options.facade;
     let dir = this.queue.shift();
     this.pending++;

     // Read the directory listing
     call.safe(facade.fs.readdir, dir.path, (err, items) => {
       if (err) {
         // fs.readdir threw an error
         this.emit('error', err);
         return this.finishedReadingDirectory();
       }

       try {
         // Process each item in the directory (simultaneously, if async)
         facade.forEach(
           items,
           this.processItem.bind(this, dir),
           this.finishedReadingDirectory.bind(this, dir)
         );
       }
       catch (err2) {
         // facade.forEach threw an error
         // (probably because fs.readdir returned an invalid result)
         this.emit('error', err2);
         this.finishedReadingDirectory();
       }
     });
   }

   /**
    * This method is called after all items in a directory have been processed.
    *
    * NOTE: This does not necessarily mean that the reader is finished, since there may still
    * be other directories queued or pending.
    */
   finishedReadingDirectory () {
     this.pending--;

     if (this.shouldRead) {
       // If we have directories queued, then start processing the next one
       if (this.queue.length > 0 && this.options.facade.async) {
         this.readNextDirectory();
       }

       this.checkForEOF();
     }
   }

   /**
    * Determines whether the reader has finished processing all items in all directories.
    * If so, then the "end" event is fired (via {@Readable#push})
    */
   checkForEOF () {
     if (this.buffer.length === 0 &&   // The stuff we've already read
     this.pending === 0 &&             // The stuff we're currently reading
     this.queue.length === 0) {        // The stuff we haven't read yet
       // There's no more stuff!
       this.stream.push(null);
     }
   }

   /**
    * Processes a single item in a directory.
    *
    * If the item is a directory, and `option.deep` is enabled, then the item will be added
    * to the directory queue.
    *
    * If the item meets the filter criteria, then it will be emitted to the reader's stream.
    *
    * @param {object} dir - A directory object from the queue
    * @param {string} item - The name of the item (name only, no path)
    * @param {function} done - A callback function that is called after the item has been processed
    */
   processItem (dir, item, done) {
     let stream = this.stream;
     let options = this.options;

     let itemPath = dir.basePath + item;
     let posixPath = dir.posixBasePath + item;
     let fullPath = path.join(dir.path, item);

     // If `options.deep` is a number, and we've already recursed to the max depth,
     // then there's no need to check fs.Stats to know if it's a directory.
     // If `options.deep` is a function, then we'll need fs.Stats
     let maxDepthReached = dir.depth >= options.recurseDepth;

     // Do we need to call `fs.stat`?
     let needStats =
       !maxDepthReached ||                                 // we need the fs.Stats to know if it's a directory
       options.stats ||                                    // the user wants fs.Stats objects returned
       options.recurseFn ||                                // we need fs.Stats for the recurse function
       options.filterFn ||                                 // we need fs.Stats for the filter function
       EventEmitter.listenerCount(stream, 'file') ||       // we need the fs.Stats to know if it's a file
       EventEmitter.listenerCount(stream, 'directory') ||  // we need the fs.Stats to know if it's a directory
       EventEmitter.listenerCount(stream, 'symlink');      // we need the fs.Stats to know if it's a symlink

     // If we don't need stats, then exit early
     if (!needStats) {
       if (this.filter(itemPath, posixPath)) {
         this.pushOrBuffer({ data: itemPath });
       }
       return done();
     }

     // Get the fs.Stats object for this path
     stat(options.facade.fs, fullPath, (err, stats) => {
       if (err) {
         // fs.stat threw an error
         this.emit('error', err);
         return done();
       }

       try {
         // Add the item's path to the fs.Stats object
         // The base of this path, and its separators are determined by the options
         // (i.e. options.basePath and options.sep)
         stats.path = itemPath;

         // Add depth of the path to the fs.Stats object for use this in the filter function
         stats.depth = dir.depth;

         if (this.shouldRecurse(stats, posixPath, maxDepthReached)) {
           // Add this subdirectory to the queue
           this.queue.push({
             path: fullPath,
             basePath: itemPath + options.sep,
             posixBasePath: posixPath + '/',
             depth: dir.depth + 1,
           });
         }

         // Determine whether this item matches the filter criteria
         if (this.filter(stats, posixPath)) {
           this.pushOrBuffer({
             data: options.stats ? stats : itemPath,
             file: stats.isFile(),
             directory: stats.isDirectory(),
             symlink: stats.isSymbolicLink(),
           });
         }

         done();
       }
       catch (err2) {
         // An error occurred while processing the item
         // (probably during a user-specified function, such as options.deep, options.filter, etc.)
         this.emit('error', err2);
         done();
       }
     });
   }

   /**
    * Pushes the given chunk of data to the stream, or adds it to the buffer,
    * depending on the state of the stream.
    *
    * @param {object} chunk
    */
   pushOrBuffer (chunk) {
     // Add the chunk to the buffer
     this.buffer.push(chunk);

     // If we're still reading, then immediately emit the next chunk in the buffer
     // (which may or may not be the chunk that we just added)
     if (this.shouldRead) {
       this.pushFromBuffer();
     }
   }

   /**
    * Immediately pushes the next chunk in the buffer to the reader's stream.
    * The "data" event will always be fired (via {@link Readable#push}).
    * In addition, the "file", "directory", and/or "symlink" events may be fired,
    * depending on the type of properties of the chunk.
    */
   pushFromBuffer () {
     let stream = this.stream;
     let chunk = this.buffer.shift();

     // Stream the data
     try {
       this.shouldRead = stream.push(chunk.data);
     }
     catch (err) {
       this.emit('error', err);
     }

     // Also emit specific events, based on the type of chunk
     chunk.file && this.emit('file', chunk.data);
     chunk.symlink && this.emit('symlink', chunk.data);
     chunk.directory && this.emit('directory', chunk.data);
   }

   /**
    * Determines whether the given directory meets the user-specified recursion criteria.
    * If the user didn't specify recursion criteria, then this function will default to true.
    *
    * @param {fs.Stats} stats - The directory's {@link fs.Stats} object
    * @param {string} posixPath - The item's POSIX path (used for glob matching)
    * @param {boolean} maxDepthReached - Whether we've already crawled the user-specified depth
    * @returns {boolean}
    */
   shouldRecurse (stats, posixPath, maxDepthReached) {
     let options = this.options;

     if (maxDepthReached) {
       // We've already crawled to the maximum depth. So no more recursion.
       return false;
     }
     else if (!stats.isDirectory()) {
       // It's not a directory. So don't try to crawl it.
       return false;
     }
     else if (options.recurseGlob) {
       // Glob patterns are always tested against the POSIX path, even on Windows
       // https://github.com/isaacs/node-glob#windows
       return options.recurseGlob.test(posixPath);
     }
     else if (options.recurseRegExp) {
       // Regular expressions are tested against the normal path
       // (based on the OS or options.sep)
       return options.recurseRegExp.test(stats.path);
     }
     else if (options.recurseFn) {
       try {
         // Run the user-specified recursion criteria
         return options.recurseFn.call(null, stats);
       }
       catch (err) {
         // An error occurred in the user's code.
         // In Sync and Async modes, this will return an error.
         // In Streaming mode, we emit an "error" event, but continue processing
         this.emit('error', err);
       }
     }
     else {
       // No recursion function was specified, and we're within the maximum depth.
       // So crawl this directory.
       return true;
     }
   }

   /**
    * Determines whether the given item meets the user-specified filter criteria.
    * If the user didn't specify a filter, then this function will always return true.
    *
    * @param {string|fs.Stats} value - Either the item's path, or the item's {@link fs.Stats} object
    * @param {string} posixPath - The item's POSIX path (used for glob matching)
    * @returns {boolean}
    */
   filter (value, posixPath) {
     let options = this.options;

     if (options.filterGlob) {
       // Glob patterns are always tested against the POSIX path, even on Windows
       // https://github.com/isaacs/node-glob#windows
       return options.filterGlob.test(posixPath);
     }
     else if (options.filterRegExp) {
       // Regular expressions are tested against the normal path
       // (based on the OS or options.sep)
       return options.filterRegExp.test(value.path || value);
     }
     else if (options.filterFn) {
       try {
         // Run the user-specified filter function
         return options.filterFn.call(null, value);
       }
       catch (err) {
         // An error occurred in the user's code.
         // In Sync and Async modes, this will return an error.
         // In Streaming mode, we emit an "error" event, but continue processing
         this.emit('error', err);
       }
     }
     else {
       // No filter was specified, so match everything
       return true;
     }
   }

   /**
    * Emits an event.  If one of the event listeners throws an error,
    * then an "error" event is emitted.
    *
    * @param {string} eventName
    * @param {*} data
    */
   emit (eventName, data) {
     let stream = this.stream;

     try {
       stream.emit(eventName, data);
     }
     catch (err) {
       if (eventName === 'error') {
         // Don't recursively emit "error" events.
         // If the first one fails, then just throw
         throw err;
       }
       else {
         stream.emit('error', err);
       }
     }
   }
 }

 module.exports = DirectoryReader;
	'use strict';

	const Readable = require('stream').Readable;
	const EventEmitter = require('events').EventEmitter;
	const path = require('path');
	const normalizeOptions = require('./normalize-options');
	const stat = require('./stat');
	const call = require('./call');

	/**
	* Asynchronously reads the contents of a directory and streams the results
	* via a {@link stream.Readable}.
	*/
	class DirectoryReader {
	/**
	* @param {string} dir - The absolute or relative directory path to read
	* @param {object} [options] - User-specified options, if any (see {@link normalizeOptions})
	* @param {object} internalOptions - Internal options that aren't part of the public API
	* @class
	*/
	constructor (dir, options, internalOptions) {
	this.options = options = normalizeOptions(options, internalOptions);

	// Indicates whether we should keep reading
	// This is set false if stream.Readable.push() returns false.
	this.shouldRead = true;

	// The directories to read
	// (initialized with the top-level directory)
	this.queue = [{
	path: dir,
	basePath: options.basePath,
	posixBasePath: options.posixBasePath,
	depth: 0
	}];

	// The number of directories that are currently being processed
	this.pending = 0;

	// The data that has been read, but not yet emitted
	this.buffer = [];

	this.stream = new Readable({ objectMode: true });
	this.stream._read = () => {
	// Start (or resume) reading
	this.shouldRead = true;

	// If we have data in the buffer, then send the next chunk
	if (this.buffer.length > 0) {
	this.pushFromBuffer();
	}

	// If we have directories queued, then start processing the next one
	if (this.queue.length > 0) {
	if (this.options.facade.sync) {
	while (this.queue.length > 0) {
	this.readNextDirectory();
	}
	}
	else {
	this.readNextDirectory();
	}
	}

	this.checkForEOF();
	};
	}

	/**
	* Reads the next directory in the queue
	*/
	readNextDirectory () {
	let facade = this.options.facade;
	let dir = this.queue.shift();
	this.pending++;

	// Read the directory listing
	call.safe(facade.fs.readdir, dir.path, (err, items) => {
	if (err) {
	// fs.readdir threw an error
	this.emit('error', err);
	return this.finishedReadingDirectory();
	}

	try {
	// Process each item in the directory (simultaneously, if async)
	facade.forEach(
	items,
	this.processItem.bind(this, dir),
	this.finishedReadingDirectory.bind(this, dir)
	);
	}
	catch (err2) {
	// facade.forEach threw an error
	// (probably because fs.readdir returned an invalid result)
	this.emit('error', err2);
	this.finishedReadingDirectory();
	}
	});
	}

	/**
	* This method is called after all items in a directory have been processed.
	*
	* NOTE: This does not necessarily mean that the reader is finished, since there may still
	* be other directories queued or pending.
	*/
	finishedReadingDirectory () {
	this.pending--;

	if (this.shouldRead) {
	// If we have directories queued, then start processing the next one
	if (this.queue.length > 0 && this.options.facade.async) {
	this.readNextDirectory();
	}

	this.checkForEOF();
	}
	}

	/**
	* Determines whether the reader has finished processing all items in all directories.
	* If so, then the "end" event is fired (via {@Readable#push})
	*/
	checkForEOF () {
	if (this.buffer.length === 0 && // The stuff we've already read
	this.pending === 0 && // The stuff we're currently reading
	this.queue.length === 0) { // The stuff we haven't read yet
	// There's no more stuff!
	this.stream.push(null);
	}
	}

	/**
	* Processes a single item in a directory.
	*
	* If the item is a directory, and `option.deep` is enabled, then the item will be added
	* to the directory queue.
	*
	* If the item meets the filter criteria, then it will be emitted to the reader's stream.
	*
	* @param {object} dir - A directory object from the queue
	* @param {string} item - The name of the item (name only, no path)
	* @param {function} done - A callback function that is called after the item has been processed
	*/
	processItem (dir, item, done) {
	let stream = this.stream;
	let options = this.options;

	let itemPath = dir.basePath + item;
	let posixPath = dir.posixBasePath + item;
	let fullPath = path.join(dir.path, item);

	// If `options.deep` is a number, and we've already recursed to the max depth,
	// then there's no need to check fs.Stats to know if it's a directory.
	// If `options.deep` is a function, then we'll need fs.Stats
	let maxDepthReached = dir.depth >= options.recurseDepth;

	// Do we need to call `fs.stat`?
	let needStats =
	!maxDepthReached \|\| // we need the fs.Stats to know if it's a directory
	options.stats \|\| // the user wants fs.Stats objects returned
	options.recurseFn \|\| // we need fs.Stats for the recurse function
	options.filterFn \|\| // we need fs.Stats for the filter function
	EventEmitter.listenerCount(stream, 'file') \|\| // we need the fs.Stats to know if it's a file
	EventEmitter.listenerCount(stream, 'directory') \|\| // we need the fs.Stats to know if it's a directory
	EventEmitter.listenerCount(stream, 'symlink'); // we need the fs.Stats to know if it's a symlink

	// If we don't need stats, then exit early
	if (!needStats) {
	if (this.filter(itemPath, posixPath)) {
	this.pushOrBuffer({ data: itemPath });
	}
	return done();
	}

	// Get the fs.Stats object for this path
	stat(options.facade.fs, fullPath, (err, stats) => {
	if (err) {
	// fs.stat threw an error
	this.emit('error', err);
	return done();
	}

	try {
	// Add the item's path to the fs.Stats object
	// The base of this path, and its separators are determined by the options
	// (i.e. options.basePath and options.sep)
	stats.path = itemPath;

	// Add depth of the path to the fs.Stats object for use this in the filter function
	stats.depth = dir.depth;

	if (this.shouldRecurse(stats, posixPath, maxDepthReached)) {
	// Add this subdirectory to the queue
	this.queue.push({
	path: fullPath,
	basePath: itemPath + options.sep,
	posixBasePath: posixPath + '/',
	depth: dir.depth + 1,
	});
	}

	// Determine whether this item matches the filter criteria
	if (this.filter(stats, posixPath)) {
	this.pushOrBuffer({
	data: options.stats ? stats : itemPath,
	file: stats.isFile(),
	directory: stats.isDirectory(),
	symlink: stats.isSymbolicLink(),
	});
	}

	done();
	}
	catch (err2) {
	// An error occurred while processing the item
	// (probably during a user-specified function, such as options.deep, options.filter, etc.)
	this.emit('error', err2);
	done();
	}
	});
	}

	/**
	* Pushes the given chunk of data to the stream, or adds it to the buffer,
	* depending on the state of the stream.
	*
	* @param {object} chunk
	*/
	pushOrBuffer (chunk) {
	// Add the chunk to the buffer
	this.buffer.push(chunk);

	// If we're still reading, then immediately emit the next chunk in the buffer
	// (which may or may not be the chunk that we just added)
	if (this.shouldRead) {
	this.pushFromBuffer();
	}
	}

	/**
	* Immediately pushes the next chunk in the buffer to the reader's stream.
	* The "data" event will always be fired (via {@link Readable#push}).
	* In addition, the "file", "directory", and/or "symlink" events may be fired,
	* depending on the type of properties of the chunk.
	*/
	pushFromBuffer () {
	let stream = this.stream;
	let chunk = this.buffer.shift();

	// Stream the data
	try {
	this.shouldRead = stream.push(chunk.data);
	}
	catch (err) {
	this.emit('error', err);
	}

	// Also emit specific events, based on the type of chunk
	chunk.file && this.emit('file', chunk.data);
	chunk.symlink && this.emit('symlink', chunk.data);
	chunk.directory && this.emit('directory', chunk.data);
	}

	/**
	* Determines whether the given directory meets the user-specified recursion criteria.
	* If the user didn't specify recursion criteria, then this function will default to true.
	*
	* @param {fs.Stats} stats - The directory's {@link fs.Stats} object
	* @param {string} posixPath - The item's POSIX path (used for glob matching)
	* @param {boolean} maxDepthReached - Whether we've already crawled the user-specified depth
	* @returns {boolean}
	*/
	shouldRecurse (stats, posixPath, maxDepthReached) {
	let options = this.options;

	if (maxDepthReached) {
	// We've already crawled to the maximum depth. So no more recursion.
	return false;
	}
	else if (!stats.isDirectory()) {
	// It's not a directory. So don't try to crawl it.
	return false;
	}
	else if (options.recurseGlob) {
	// Glob patterns are always tested against the POSIX path, even on Windows
	// https://github.com/isaacs/node-glob#windows
	return options.recurseGlob.test(posixPath);
	}
	else if (options.recurseRegExp) {
	// Regular expressions are tested against the normal path
	// (based on the OS or options.sep)
	return options.recurseRegExp.test(stats.path);
	}
	else if (options.recurseFn) {
	try {
	// Run the user-specified recursion criteria
	return options.recurseFn.call(null, stats);
	}
	catch (err) {
	// An error occurred in the user's code.
	// In Sync and Async modes, this will return an error.
	// In Streaming mode, we emit an "error" event, but continue processing
	this.emit('error', err);
	}
	}
	else {
	// No recursion function was specified, and we're within the maximum depth.
	// So crawl this directory.
	return true;
	}
	}

	/**
	* Determines whether the given item meets the user-specified filter criteria.
	* If the user didn't specify a filter, then this function will always return true.
	*
	* @param {string\|fs.Stats} value - Either the item's path, or the item's {@link fs.Stats} object
	* @param {string} posixPath - The item's POSIX path (used for glob matching)
	* @returns {boolean}
	*/
	filter (value, posixPath) {
	let options = this.options;

	if (options.filterGlob) {
	// Glob patterns are always tested against the POSIX path, even on Windows
	// https://github.com/isaacs/node-glob#windows
	return options.filterGlob.test(posixPath);
	}
	else if (options.filterRegExp) {
	// Regular expressions are tested against the normal path
	// (based on the OS or options.sep)
	return options.filterRegExp.test(value.path \|\| value);
	}
	else if (options.filterFn) {
	try {
	// Run the user-specified filter function
	return options.filterFn.call(null, value);
	}
	catch (err) {
	// An error occurred in the user's code.
	// In Sync and Async modes, this will return an error.
	// In Streaming mode, we emit an "error" event, but continue processing
	this.emit('error', err);
	}
	}
	else {
	// No filter was specified, so match everything
	return true;
	}
	}

	/**
	* Emits an event. If one of the event listeners throws an error,
	* then an "error" event is emitted.
	*
	* @param {string} eventName
	* @param {*} data
	*/
	emit (eventName, data) {
	let stream = this.stream;

	try {
	stream.emit(eventName, data);
	}
	catch (err) {
	if (eventName === 'error') {
	// Don't recursively emit "error" events.
	// If the first one fails, then just throw
	throw err;
	}
	else {
	stream.emit('error', err);
	}
	}
	}
	}

	module.exports = DirectoryReader;