index.js 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104
  1. var Readable = require('readable-stream').Readable
  2. var inherits = require('inherits')
  3. module.exports = from2
  4. from2.ctor = ctor
  5. from2.obj = obj
  6. var Proto = ctor()
  7. function toFunction(list) {
  8. list = list.slice()
  9. return function (_, cb) {
  10. var err = null
  11. var item = list.length ? list.shift() : null
  12. if (item instanceof Error) {
  13. err = item
  14. item = null
  15. }
  16. cb(err, item)
  17. }
  18. }
  19. function from2(opts, read) {
  20. if (typeof opts !== 'object' || Array.isArray(opts)) {
  21. read = opts
  22. opts = {}
  23. }
  24. var rs = new Proto(opts)
  25. rs._from = Array.isArray(read) ? toFunction(read) : (read || noop)
  26. return rs
  27. }
  28. function ctor(opts, read) {
  29. if (typeof opts === 'function') {
  30. read = opts
  31. opts = {}
  32. }
  33. opts = defaults(opts)
  34. inherits(Class, Readable)
  35. function Class(override) {
  36. if (!(this instanceof Class)) return new Class(override)
  37. this._reading = false
  38. this._callback = check
  39. this.destroyed = false
  40. Readable.call(this, override || opts)
  41. var self = this
  42. var hwm = this._readableState.highWaterMark
  43. function check(err, data) {
  44. if (self.destroyed) return
  45. if (err) return self.destroy(err)
  46. if (data === null) return self.push(null)
  47. self._reading = false
  48. if (self.push(data)) self._read(hwm)
  49. }
  50. }
  51. Class.prototype._from = read || noop
  52. Class.prototype._read = function(size) {
  53. if (this._reading || this.destroyed) return
  54. this._reading = true
  55. this._from(size, this._callback)
  56. }
  57. Class.prototype.destroy = function(err) {
  58. if (this.destroyed) return
  59. this.destroyed = true
  60. var self = this
  61. process.nextTick(function() {
  62. if (err) self.emit('error', err)
  63. self.emit('close')
  64. })
  65. }
  66. return Class
  67. }
  68. function obj(opts, read) {
  69. if (typeof opts === 'function' || Array.isArray(opts)) {
  70. read = opts
  71. opts = {}
  72. }
  73. opts = defaults(opts)
  74. opts.objectMode = true
  75. opts.highWaterMark = 16
  76. return from2(opts, read)
  77. }
  78. function noop () {}
  79. function defaults(opts) {
  80. opts = opts || {}
  81. return opts
  82. }