Stream

Copy 
1var Stream = require('stream')
2
3var Readable = Stream.Readable
4var Writable = Stream.Writable
5var Duplex = Stream.Duplex
6var Transform = Stream.Transform

Copy 
1var fs = require('fs')
2// `fs.createReadStream`创建一个`Readable`对象以读取`bigFile`的内容，并输出到标准输出
3// 如果使用`fs.readFile`则可能由于文件过大而失败
4fs.createReadStream(bigFile).pipe(process.stdout)

Copy 
1'use strict'
2const Readable = require('stream').Readable
3
4class ToReadable extends Readable {
5  constructor(iterator) {
6    super()
7    this.iterator = iterator
8  }
9
10  // 子类需要实现该方法
11  // 这是生产数据的逻辑
12  _read() {
13    const res = this.iterator.next()
14    if (res.done) {
15      // 数据源已枯竭，调用`push(null)`通知流
16      return this.push(null)
17    }
18    setTimeout(() => {
19      // 通过`push`方法将数据添加到流中
20      this.push(res.value + '\n')
21    }, 0)
22  }
23}
24
25module.exports = ToReadable

Copy 
1const iterator = function (limit) {
2  return {
3    next: function () {
4      if (limit--) {
5        return { done: false, value: limit + Math.random() }
6      }
7      return { done: true }
8    }
9  }
10}(1e10)
11
12const readable = new ToReadable(iterator)
13
14// 监听`data`事件，一次获取一个数据
15readable.on('data', data => process.stdout.write(data))
16
17// 所有数据均已读完
18readable.on('end', () => process.stdout.write('DONE'))

Copy 
1const Writable = require('stream').Writable
2
3const writable = Writable()
4// 实现`_write`方法
5// 这是将数据写入底层的逻辑
6writable._write = function (data, enc, next) {
7  // 将流中的数据写入底层
8  process.stdout.write(data.toString().toUpperCase())
9  // 写入完成时，调用`next()`方法通知流传入下一个数据
10  process.nextTick(next)
11}
12
13// 所有数据均已写入底层
14writable.on('finish', () => process.stdout.write('DONE'))
15
16// 将一个数据写入流中
17writable.write('a' + '\n')
18writable.write('b' + '\n')
19writable.write('c' + '\n')
20
21// 再无数据写入流时，需要调用`end`方法
22writable.end()

Copy 
1var Duplex = require('stream').Duplex
2
3var duplex = Duplex()
4
5// 可读端底层读取逻辑
6duplex._read = function () {
7  this._readNum = this._readNum || 0
8  if (this._readNum > 1) {
9    this.push(null)
10  } else {
11    this.push('' + (this._readNum++))
12  }
13}
14
15// 可写端底层写逻辑
16duplex._write = function (buf, enc, next) {
17  // a, b
18  process.stdout.write('_write ' + buf.toString() + '\n')
19  next()
20}
21
22// 0, 1
23duplex.on('data', data => console.log('ondata', data.toString()))
24
25duplex.write('a')
26duplex.write('b')
27duplex.write('x')
28
29
30duplex.end()

Copy 
1'use strict'
2
3const Transform = require('stream').Transform
4
5class Rotate extends Transform {
6  constructor(n) {
7    super()
8    // 将字母移动`n`个位置
9    this.offset = (n || 13) % 26
10  }
11
12  // 将可写端写入的数据变换后添加到可读端
13  _transform(buf, enc, next) {
14    var res = buf.toString().split('').map(c => {
15      var code = c.charCodeAt(0)
16      if (c >= 'a' && c <= 'z') {
17        code += this.offset
18        if (code > 'z'.charCodeAt(0)) {
19          code -= 26
20        }
21      } else if (c >= 'A' && c <= 'Z') {
22        code += this.offset
23        if (code > 'Z'.charCodeAt(0)) {
24          code -= 26
25        }
26      }
27      return String.fromCharCode(code)
28    }).join('')
29
30    // 调用push方法将变换后的数据添加到可读端
31    this.push(res)
32    // 调用next方法准备处理下一个
33    next()
34  }
35
36}
37
38var transform = new Rotate(3)
39transform.on('data', data => process.stdout.write(data))
40transform.write('hello, ')
41transform.write('world!')
42transform.end()

Copy 
1const Readable = require('stream').Readable
2
3const readable = Readable()
4
5readable.push('a')
6readable.push('b')
7readable.push(null)
8
9readable.on('data', data => console.log(data))

Copy 
1const Readable = require('stream').Readable
2
3const readable = Readable({ objectMode: true })
4
5readable.push('a')
6readable.push('b')
7readable.push({})
8readable.push(null)
9
10readable.on('data', data => console.log(data))