blob: 1e58cdee1fad5500ed98baff9cbad3ab4229113c (
plain)
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
|
var Readable = require('readable-stream').Readable
var inherits = require('inherits')
module.exports = from2
from2.ctor = ctor
from2.obj = obj
var Proto = ctor()
function toFunction(list) {
list = list.slice()
return function (_, cb) {
cb(null, list.length ? list.shift() : null)
}
}
function from2(opts, read) {
if (typeof opts !== 'object' || Array.isArray(opts)) {
read = opts
opts = {}
}
if (Array.isArray(read)) read = toFunction(read)
var rs = new Proto(opts)
rs._from = read
return rs
}
function ctor(opts, read) {
if (typeof opts === 'function') {
read = opts
opts = {}
}
opts = defaults(opts)
inherits(Class, Readable)
function Class(override) {
if (!(this instanceof Class)) return new Class(override)
this._reading = false
this.destroyed = false
Readable.call(this, override || opts)
}
Class.prototype._from = read
Class.prototype._read = function(size) {
var self = this
if (this._reading || this.destroyed) return
this._reading = true
this._from(size, check)
function check(err, data) {
if (self.destroyed) return
if (err) return self.destroy(err)
if (data === null) return self.push(null)
self._reading = false
if (self.push(data)) self._read()
}
}
Class.prototype.destroy = function(err) {
if (this.destroyed) return
this.destroyed = true
var self = this
process.nextTick(function() {
if (err) self.emit('error', err)
self.emit('close')
})
}
return Class
}
function obj(opts, read) {
if (typeof opts === 'function' || Array.isArray(opts)) {
read = opts
opts = {}
}
opts = defaults(opts)
opts.objectMode = true
opts.highWaterMark = 16
return from2(opts, read)
}
function defaults(opts) {
opts = opts || {}
return opts
}
|
