123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899 |
- 'use strict';
- var si = require('set-immediate-shim');
- var stream = require('readable-stream');
- var util = require('util');
- var Readable = stream.Readable;
- module.exports = ReaddirpReadable;
- util.inherits(ReaddirpReadable, Readable);
- function ReaddirpReadable (opts) {
- if (!(this instanceof ReaddirpReadable)) return new ReaddirpReadable(opts);
- opts = opts || {};
- opts.objectMode = true;
- Readable.call(this, opts);
- // backpressure not implemented at this point
- this.highWaterMark = Infinity;
- this._destroyed = false;
- this._paused = false;
- this._warnings = [];
- this._errors = [];
- this._pauseResumeErrors();
- }
- var proto = ReaddirpReadable.prototype;
- proto._pauseResumeErrors = function () {
- var self = this;
- self.on('pause', function () { self._paused = true });
- self.on('resume', function () {
- if (self._destroyed) return;
- self._paused = false;
- self._warnings.forEach(function (err) { self.emit('warn', err) });
- self._warnings.length = 0;
- self._errors.forEach(function (err) { self.emit('error', err) });
- self._errors.length = 0;
- })
- }
- // called for each entry
- proto._processEntry = function (entry) {
- if (this._destroyed) return;
- this.push(entry);
- }
- proto._read = function () { }
- proto.destroy = function () {
- // when stream is destroyed it will emit nothing further, not even errors or warnings
- this.push(null);
- this.readable = false;
- this._destroyed = true;
- this.emit('close');
- }
- proto._done = function () {
- this.push(null);
- }
- // we emit errors and warnings async since we may handle errors like invalid args
- // within the initial event loop before any event listeners subscribed
- proto._handleError = function (err) {
- var self = this;
- si(function () {
- if (self._paused) return self._warnings.push(err);
- if (!self._destroyed) self.emit('warn', err);
- });
- }
- proto._handleFatalError = function (err) {
- var self = this;
- si(function () {
- if (self._paused) return self._errors.push(err);
- if (!self._destroyed) self.emit('error', err);
- });
- }
- function createStreamAPI () {
- var stream = new ReaddirpReadable();
- return {
- stream : stream
- , processEntry : stream._processEntry.bind(stream)
- , done : stream._done.bind(stream)
- , handleError : stream._handleError.bind(stream)
- , handleFatalError : stream._handleFatalError.bind(stream)
- };
- }
- module.exports = createStreamAPI;
|