1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
|
'use strict';
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;
setImmediate(function () {
if (self._paused) return self._warnings.push(err);
if (!self._destroyed) self.emit('warn', err);
});
}
proto._handleFatalError = function (err) {
var self = this;
setImmediate(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;
|
