Node.js 中的 Stream（流）

A stream is an abstract interface implemented by various objects in Node. 

For example a request to an HTTP server is a stream, as is stdout. 

Streams are readable, writable, or both. All streams are instances of EventEmitter

// data 事件触发 flowing mode
Readable.prototype.on = function (ev, fn) {
  // ...

  if (ev === 'data' && false !== this._readableState.flowing) {
    this.resume()
  }

  // ...
}

// resume 触发 flowing mode
Readable.prototype.resume = function () {
  var state = this._readableState
  if (!state.flowing) {
    debug('resume')
    state.flowing = true
    resume(this, state)
  }
  return this
}

// pipe 方法触发 flowing 模式
Readable.prototype.resume = function () {
  if (!state.flowing) {
    this.resume()
  }
}

true ==> flowing mode

false ==> paused mode

readable._readableState.flowing = null

readable._readableState.flowing = false

readable._readableState.flowing = true

var Read = require('stream').Readable
var r = new Read()

r.push('hello')
r.push('world')
r.push(null)

console.log(r.read(1).toString())  // h
console.log(r.read(3).toString())  // ell

var Read = require('stream').Readable
var r = new Read()

r.push('hello')
r.push('world')
r.push(null)

r.on('readable', function() {
  var chunk = r.read()
  console.log(chunk.toString())
})

// helloworld

function emitReadable_(stream) {
  debug('emit readable')
  stream.emit('readable')
  flow(stream)
}

function flow(stream) {
  var state = stream._readableState
  debug('flow', state.flowing)
  if (state.flowing) {
    do {
      var chunk = stream.read()
    } while (null !== chunk && state.flowing)
  }
}

Readable.prototype.read = function (n) {
  // ...

  var res = fromList(n, state)

  if (!util.isNull(ret)) {
    this.emit('data', ret)
  }

  // ...
}

var Read = require('stream').Readable
var r = new Read()

r.push('hello')
r.push('world')
r.push(null)

r.on('data', function (chunk) {
  console.log(chunk.toString())
})

// hello 
// world

var Read = require('stream').Readable
var r = new Read()

r.push('hello')
r.push('world')
r.push(null)

r.pipe(process.stdout)

// helloworld

read() ==> _read(0) ==> ... ==> addChunk()

// 可以将 size 设为 1 或 undefined 来进行测试
const size = 1
const rs = require('fs').createReadStream('./test.js')

rs.on('readable', () => {
  console.log(rs.read(size))
})

read() ==> _read() ==> addChunk()

const rs = require('fs').createReadStream('./test.js')

rs.on('readable', () => console.log('readable 触发'))
rs.on('data', console.log)

function addChunk(stream, state, chunk, addToFront) {
  // 如果处于 flowing 模式，且缓冲区为空，且为异步调用时候，触发 data 事件
  if (state.flowing && state.length === 0 && !state.sync) {
    state.awaitDrain = 0
    stream.emit('data', chunk)
  } else {
    // 更新缓冲区已有数据数量
    state.length += state.objectMode ? 1 : chunk.length
    if (addToFront)
      // 插入缓冲区头部
      state.buffer.unshift(chunk)
    else
      // 插入缓冲区尾部
      state.buffer.push(chunk)

    if (state.needReadable)
      // 触发 readable 事件
      emitReadable(stream)
  }
  maybeReadMore(stream, state)
}

const myStream = getWritableStreamSomehow()

myStream.write('some data')
myStream.write('some more data')
myStream.end('done writing data')

const Writable = require('stream').Writable

const writable = Writable()

// 实现 _write 方法
// 这是将数据写入底层的逻辑
writable._write = function (data, enc, next) {
  // 将流中的数据写入底层
  process.stdout.write(data.toString().toUpperCase())
  // 写入完成时，调用 next() 方法通知流传入下一个数据
  process.nextTick(next)
}

// 所有数据均已写入底层
writable.on('finish', () => process.stdout.write('DONE'))

// 将一个数据写入流中
writable.write('a' + '\n')
writable.write('b' + '\n')
writable.write('c' + '\n')

// 再无数据写入流时，需要调用 end() 方法
writable.end()

var Read = require('stream').Readable
var Write = require('stream').Writable
var r = new Read()
var w = new Write()

r.push('hello')
r.push('world')
r.push(null)

w._write = function(chunk, ev, cb) {
  console.log(chunk.toString())
  cb()
}

r.pipe(w)

// hello
// world

Readable.prototype.pipe = function (dest, pipeOpts) {
  var src = this

  // ...

  src.on('data', ondata)

  function ondata(chunk) {
    var ret = dest.write(chunk)
    if (false === ret) {
      debug('false write response, pause',
        src._readableState.awaitDrain)
      src._readableState.awaitDrain++
      src.pause()
    }
  }

  // ...

}

var Read = require('stream').Readable
var Write = require('stream').Writable
var r = new Read()
var w = new Write()

r.push('hello')
r.push('world')
r.push(null)

w._write = function(chunk, ev, cb) {
  console.log(chunk.toString())
  cb()
}

r.on('data', function(chunk) {
  if (!w.write(chunk)) {
    r.pause()
  }
})

w.on('drain', function() {
  r.resume()
})

// hello
// world

var Duplex = require('stream').Duplex

var duplex = Duplex()

// 可读端底层读取逻辑
duplex._read = function () {
  this._readNum = this._readNum || 0
  if (this._readNum > 1) {
    this.push(null)
  } else {
    this.push(' ' + (this._readNum++))
  }
}

// 可写端底层写逻辑
duplex._write = function (buf, enc, next) {
  // a, b
  process.stdout.write('_write ' + buf.toString() + '\n')
  next()
}

// 0, 1
duplex.on('data', data => console.log('ondata', data.toString()))

duplex.write('a')
duplex.write('b')

duplex.end()

const { Duplex } = require('stream')

const inoutStream = new Duplex({
  write(chunk, encoding, callback) {
    console.log(chunk.toString())
    callback()
  },

  read() {
    this.push(String.fromCharCode(this.currentCharCode++))
    if (this.currentCharCode > 90) {
      this.push(null)
    }
  }
})

inoutStream.currentCharCode = 65

process.stdin.pipe(inoutStream).pipe(process.stdout)

inoutStream.end()

const { Transform } = require('stream')

const upperCaseTr = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase())
    callback()
  }
})

process.stdin.pipe(upperCaseTr).pipe(process.stdout)

const fs = require('fs')
const zlib = require('zlib')
const file = process.argv[2]

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream(file + '.gz'))

const fs = require('fs')
const zlib = require('zlib')
const file = process.argv[2]

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .on('data', () => process.stdout.write('.'))
  .pipe(fs.createWriteStream(file + '.zz'))
  .on('finish', () => console.log('Done'))

const fs = require('fs')
const zlib = require('zlib')
const file = process.argv[2]

const { Transform } = require('stream')

const reportProgress = new Transform({
  transform(chunk, encoding, callback) {
    process.stdout.write('.')
    callback(null, chunk)
  }
})

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file + '.zz'))
  .on('finish', () => console.log('Done'))

const crypto = require('crypto')
const fs = require('fs')
const zlib = require('zlib')
const file = process.argv[2]

const { Transform } = require('stream')

const reportProgress = new Transform({
  transform(chunk, encoding, callback) {
    process.stdout.write('.')
    callback(null, chunk)
  }
})

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(crypto.createCipher('aes192', 'a_secret'))
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file + '.zz'))
  .on('finish', () => console.log('Done'))

fs.createReadStream(file)
  .pipe(crypto.createDecipher('aes192', 'a_secret'))
  .pipe(zlib.createGunzip())
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file.slice(0, -3)))
  .on('finish', () => console.log('Done'))

// 自定义 readable stream 的实现
var Stream = require('stream')
var Read = Stream.Readable
var util = require('util')

util.inherits(MyReadStream, Read)

function MyReadStream(data, opt) {
  Read.call(this, opt)
  this.data = data || []
}

MyReadStream.prototype._read = function () {
  var _this = this
  this.data.forEach(function (d) {
    _this.push(d)
  })
  this.push(null)
}

var data = ['aa', 'bb', 'cc']
var r = new MyReadStream(data)

r.on('data', function (chunk) {
  console.log(chunk.toString())
})

// 数据将会在读入侧堆积，这样写入侧才能和数据流的读入速度保持同步
inp.pipe(gzip).pipe(outputFile)

Readable.prototype.pipe = function (dest, pipeOpts) {

  // ...

  var ondrain = pipeOnDrain(src)
  // 当写操作返回 false 的时候，正常情况下必然会在稍后触发一个 drain 事件
  dest.on('drain', ondrain)
  src.on('data', ondata)
  function ondata(chunk) {
    var ret = dest.write(chunk)
    // 如果写操作的返回值为 false，则暂停 readable 流
    if (ret === false) {
      if (((state.pipesCount === 1 && state.pipes === dest) ||
        (state.pipesCount > 1 && state.pipes.indexOf(dest) !== -1)) &&
        !cleanedUp) {
        state.awaitDrain++
      }
      src.pause()
    }
  }

  // ...

  return dest
}

function pipeOnDrain(src) {
  return function () {
    var state = src._readableState
    if (state.awaitDrain)
      state.awaitDrain--
    if (state.awaitDrain === 0 && EE.listenerCount(src, 'data')) {
      // 将流重新设为 flowing 模式
      state.flowing = true
      // 将缓冲区中残留的数据读取并重新触发 data 事件
      flow(src)
    }
  }
}

                                                     +===================+
                         x-->  Piping functions   +-->   src.pipe(dest)  |
                         x     are set up during     |===================|
                         x     the .pipe method.     |  Event callbacks  |
  +===============+      x                           |-------------------|
  |   Your Data   |      x     They exist outside    | .on('close', cb)  |
  +=======+=======+      x     the data flow, but    | .on('data', cb)   |
          |              x     importantly attach    | .on('drain', cb)  |
          |              x     events, and their     | .on('unpipe', cb) |
+---------v---------+    x     respective callbacks. | .on('error', cb)  |
|  Readable Stream  +----+                           | .on('finish', cb) |
+-^-------^-------^-+    |                           | .on('end', cb)    |
  ^       |       ^      |                           +-------------------+
  |       |       |      |
  |       ^       |      |
  ^       ^       ^      |    +-------------------+         +=================+
  ^       |       ^      +---->  Writable Stream  +--------->  .write(chunk)  |
  |       |       |           +-------------------+         +=======+=========+
  |       |       |                                                 |
  |       ^       |                              +------------------v---------+
  ^       |       +-> if (!chunk)                |    Is this chunk too big?  |
  ^       |       |     emit.end()               |    Is the queue busy?      |
  |       |       +-> else                       +-------+----------------+---+
  |       ^       |     emit.write()                     |                |
  |       ^       ^                                   +--v---+        +---v---+
  |       |       ^-----------------------------------<  No  |        |  Yes  |
  ^       |                                           +------+        +---v---+
  ^       |                                                               |
  |       ^               emit.pause()            +=================+     |
  |       ^---------------^-----------------------+  return false   <-----+---+
  |                                               +=================+         |
  |                                                                           |
  ^            when queue is empty     +============+                         |
  ^------------^-----------------------<  Buffering |                         |
               |                       |============|                         |
               +> emit.drain()         |  ^Buffer^  |                         |
               +> emit.resume()        +------------+                         |
                                       |  ^Buffer^  |                         |
                                       +------------+   add chunk to queue    |
                                       |            <---^---------------------<
                                       +============+

Readable.pipe(Transformable).pipe(Writable)