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) {
|
var err = null
|
var item = list.length ? list.shift() : null
|
if (item instanceof Error) {
|
err = item
|
item = null
|
}
|
|
cb(err, item)
|
}
|
}
|
|
function from2(opts, read) {
|
if (typeof opts !== 'object' || Array.isArray(opts)) {
|
read = opts
|
opts = {}
|
}
|
|
var rs = new Proto(opts)
|
rs._from = Array.isArray(read) ? toFunction(read) : (read || noop)
|
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._callback = check
|
this.destroyed = false
|
Readable.call(this, override || opts)
|
|
var self = this
|
var hwm = this._readableState.highWaterMark
|
|
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(hwm)
|
}
|
}
|
|
Class.prototype._from = read || noop
|
Class.prototype._read = function(size) {
|
if (this._reading || this.destroyed) return
|
this._reading = true
|
this._from(size, this._callback)
|
}
|
|
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 noop () {}
|
|
function defaults(opts) {
|
opts = opts || {}
|
return opts
|
}
|
