稳定性: 2 - 不稳定

const stream = require('stream'

javascript
var http = require('http'

var server = http.createServer(function (req, res) {
  // req is an http.IncomingMessage, which is 可读流（Readable stream）
  // res is an http.ServerResponse, which is a Writable Stream

  var body = '';
  // we want to get the data as utf8 strings
  // If you don't set an encoding, then you'll get Buffer objects
  req.setEncoding('utf8'

  // 可读流（Readable stream） emit 'data' 事件 once a 监听器（listener） is added
  req.on('data', function (chunk) {
    body += chunk;
  }

  // the end 事件 tells you that you have entire body
  req.on('end', function () {
    try {
      var data = JSON.parse(body
    } catch (er) {
      // uh oh!  bad json!
      res.statusCode = 400;
      return res.end('error: ' + er.message
    }

    // write back something interesting to the user:
    res.write(typeof data
    res.end(
  }
}

server.listen(1337

// $ curl localhost:1337 -d '{}'
// object
// $ curl localhost:1337 -d '"foo"'
// string
// $ curl localhost:1337 -d 'not json'
// error: Unexpected token o

javascript
var readble = getReadableStreamSomehow(
readable.on('readable', function() {
  // there is some data to read now
}

javascript
var readable = getReadableStreamSomehow(
readable.on('data', function(chunk) {
  console.log('got %d bytes of data', chunk.length
}

javascript
    var readable = getReadableStreamSomehow(
    readable.on('data', function(chunk) {
        console.log('got %d bytes of data', chunk.length
    }
    readable.on('end', function() {
        console.log('there will be no more data.'
    }  

javascript
var readable = getReadableStreamSomehow(
readable.on('readable', function() {
  var chunk;
  while (null !== (chunk = readable.read())) {
    console.log('got %d bytes of data', chunk.length
  }
}

javascript
var readable = getReadableStreamSomehow(
readable.setEncoding('utf8'
readable.on('data', function(chunk) {
  assert.equal(typeof chunk, 'string'
  console.log('got %d characters of string data', chunk.length
}

javascript
var readable = getReadableStreamSomehow(
readable.resume(
readable.on('end', function(chunk) {
  console.log('got to the end, but did not read anything'
}

javascript
var readable = getReadableStreamSomehow(
readable.on('data', function(chunk) {
  console.log('got %d bytes of data', chunk.length
  readable.pause(
  console.log('there will be no more data for 1 second'
  setTimeout(function() {
    console.log('now data will start flowing again'
    readable.resume(
  }, 1000
}

var readable = new stream.Readable

readable.isPaused() // === false
readable.pause()
readable.isPaused() // === true
readable.resume()
readable.isPaused() // === false

javascript
var readable = getReadableStreamSomehow(
var writable = fs.createWriteStream('file.txt'
// All the data from readable goes into 'file.txt'
readable.pipe(writable

javascript
var r = fs.createReadStream('file.txt'
var z = zlib.createGzip(
var w = fs.createWriteStream('file.txt.gz'
r.pipe(z).pipe(w

javascript
process.stdin.pipe(process.stdout

javascript
reader.pipe(writer, { end: false }
reader.on('end', function() {
  writer.end('Goodbye\n'
}

javascript
var readable = getReadableStreamSomehow(
var writable = fs.createWriteStream('file.txt'
// All the data from readable goes into 'file.txt',
// but only for the first second
readable.pipe(writable
setTimeout(function() {
  console.log('stop writing to file.txt'
  readable.unpipe(writable
  console.log('manually close the file stream'
  writable.end(
}, 1000

javascript
// Pull off a header delimited by \n\n
// use unshift() if we get too much
// Call the callback with (error, header, stream)
var StringDecoder = require('string_decoder').StringDecoder;
function parseHeader(stream, callback) {
  stream.on('error', callback
  stream.on('readable', onReadable
  var decoder = new StringDecoder('utf8'
  var header = '';
  function onReadable() {
    var chunk;
    while (null !== (chunk = stream.read())) {
      var str = decoder.write(chunk
      if (str.match(/\n\n/)) {
        // found the header boundary
        var split = str.split(/\n\n/
        header += split.shift(
        var remaining = split.join('\n\n'
        var buf = new Buffer(remaining, 'utf8'
        if (buf.length)
          stream.unshift(buf
        stream.removeListener('error', callback
        stream.removeListener('readable', onReadable
        // now the body of the message can be read from the stream.
        callback(null, header, stream
      } else {
        // still reading the header.
        header += str;
      }
    }
  }
}

javascript
var OldReader = require('./old-api-module.js').OldReader;
var oreader = new OldReader;
var Readable = require('stream').Readable;
var myReader = new Readable().wrap(oreader

myReader.on('readable', function() {
  myReader.read( // etc.
}

javascript
// Write the data to the supplied 可写流（Writable stream ） 1MM times.
// Be attentive to back-pressure.
function writeOneMillionTimes(writer, data, encoding, callback) {
  var i = 1000000;
  write(
  function write() {
    var ok = true;
    do {
      i -= 1;
      if (i === 0) {
        // last time!
        writer.write(data, encoding, callback
      } else {
        // see if we should continue, or wait
        // don't pass the callback, because we're not done yet.
        ok = writer.write(data, encoding
      }
    } while (i > 0 && ok
    if (i > 0) {
      // had to stop early!
      // write some more once it drains
      writer.once('drain', write
    }
  }
}

javascript
// write 'hello, ' and then end with 'world!'
var file = fs.createWriteStream('example.txt'
file.write('hello, '
file.end('world!'
// writing more now is not allowed!

javascript
var writer = getWritableStreamSomehow(
for (var i = 0; i < 100; i ++) {
  writer.write('hello, #' + i + '!\n'
}
writer.end('this is the end\n'
writer.on('finish', function() {
  console.error('all writes are now complete.'
}

javascript
var writer = getWritableStreamSomehow(
var reader = getReadableStreamSomehow(
writer.on('pipe', function(src) {
  console.error('something is piping into the writer'
  assert.equal(src, reader
}
reader.pipe(writer

javascript
var writer = getWritableStreamSomehow(
var reader = getReadableStreamSomehow(
writer.on('unpipe', function(src) {
  console.error('something has stopped piping into the writer'
  assert.equal(src, reader
}
reader.pipe(writer
reader.unpipe(writer

javascript
var Readable = require('stream').Readable;
var util = require('util'
util.inherits(Counter, Readable

function Counter(opt) {
  Readable.call(this, opt
  this._max = 1000000;
  this._index = 1;
}

Counter.prototype._read = function() {
  var i = this._index++;
  if (i > this._max)
    this.push(null
  else {
    var str = '' + i;
    var buf = new Buffer(str, 'ascii'
    this.push(buf
  }
};

javascript
// A parser for a simple data protocol.
// "header" is a JSON object, followed by 2 \n characters, and
// then a message body.
//
// 注意: This can be done more simply as a Transform stream!
// Using Readable directly for this is sub-optimal.  See the
// alternative example below under Transform section.

var Readable = require('stream').Readable;
var util = require('util'

util.inherits(SimpleProtocol, Readable

  function SimpleProtocol(source, options) {
  if (!(this instanceof SimpleProtocol))
    return new SimpleProtocol(source, options

  Readable.call(this, options;

  this._inBody = false;
  this._sawFirstCr = false;

  // source is 可读流（Readable stream）, such as a socket or file
  this._source = source;

  var self = this;
  source.on('end', function() {
    self.push(null
  }

  // give it a kick whenever the source is readable
  // read(0) will not consume any bytes
  source.on('readable', function() {
    self.read(0
  }

  this._rawHeader = [];
  this.header = null;
}

SimpleProtocol.prototype._read = function(n) {
  if (!this._inBody) {
    var chunk = this._source.read(

    // if the source doesn't have data, we don't have data yet.
    if (chunk === null)
      return this.push(''

    // check if the chunk has a \n\n
    var split = -1;
    for (var i = 0; i < chunk.length; i++) {
      if (chunk[i] === 10) { // '\n'
        if (this._sawFirstCr) {
          split = i;
          break;
        } else {
          this._sawFirstCr = true;
        }
      } else {
        this._sawFirstCr = false;
      }
    }

    if (split === -1) {
      // still waiting for the \n\n
      // stash the chunk, and try again.
      this._rawHeader.push(chunk
      this.push(''
    } else {
      this._inBody = true;
      var h = chunk.slice(0, split
      this._rawHeader.push(h
      var header = Buffer.concat(this._rawHeader).toString(
      try {
        this.header = JSON.parse(header
      } catch (er) {
        this.emit('error', new Error('invalid simple protocol data')
        return;
      }
      // now, because we got some extra data, unshift the rest
      // back into the 读取队列 so that our consumer will see it.
      var b = chunk.slice(split
      this.unshift(b

      // and let them know that we are done parsing the header.
      this.emit('header', this.header
    }
  } else {
    // from there on, just provide the data to our consumer.
    // careful not to push(null), since that would indicate EOF.
    var chunk = this._source.read(
    if (chunk) this.push(chunk
  }
};

// Usage:
// var parser = new SimpleProtocol(source
// Now parser is 可读流（Readable stream） that will emit 'header'
// with the parsed header data.

javascript
// source is an object with readStop() and readStart() 方法s,
// and an `ondata` member that gets called when it has data, and
// an `onend` member that gets called when the data is over.

util.inherits(SourceWrapper, Readable

function SourceWrapper(options) {
  Readable.call(this, options

  this._source = getLowlevelSourceObject(
  var self = this;

  // Every time there's data, we push it into the internal buffer.
  this._source.ondata = function(chunk) {
    // if push() 返回 false, then we need to stop reading from source
    if (!self.push(chunk))
      self._source.readStop(
  };

  // When the source ends, we push the EOF-signaling `null` chunk
  this._source.onend = function() {
    self.push(null
  };
}

// _read will be called when the stream wants to pull more data in
// the advisory size 参数 is ignored in this case.
SourceWrapper.prototype._read = function(size) {
  this._source.readStart(
};

javascript
transform.prototype._transform = function (data, encoding, callback) {
  this.push(data
  callback(
}

transform.prototype._transform = function (data, encoding, callback) {
  callback(null, data
}

javascript
var util = require('util'
var Transform = require('stream').Transform;
util.inherits(SimpleProtocol, Transform

function SimpleProtocol(options) {
  if (!(this instanceof SimpleProtocol))
    return new SimpleProtocol(options

  Transform.call(this, options
  this._inBody = false;
  this._sawFirstCr = false;
  this._rawHeader = [];
  this.header = null;
}

SimpleProtocol.prototype._transform = function(chunk, encoding, done) {
  if (!this._inBody) {
    // check if the chunk has a \n\n
    var split = -1;
    for (var i = 0; i < chunk.length; i++) {
      if (chunk[i] === 10) { // '\n'
        if (this._sawFirstCr) {
          split = i;
          break;
        } else {
          this._sawFirstCr = true;
        }
      } else {
        this._sawFirstCr = false;
      }
    }

    if (split === -1) {
      // still waiting for the \n\n
      // stash the chunk, and try again.
      this._rawHeader.push(chunk
    } else {
      this._inBody = true;
      var h = chunk.slice(0, split
      this._rawHeader.push(h
      var header = Buffer.concat(this._rawHeader).toString(
      try {
        this.header = JSON.parse(header
      } catch (er) {
        this.emit('error', new Error('invalid simple protocol data')
        return;
      }
      // and let them know that we are done parsing the header.
      this.emit('header', this.header

      // now, because we got some extra data, emit this first.
      this.push(chunk.slice(split)
    }
  } else {
    // from there on, just provide the data to our consumer as-is.
    this.push(chunk
  }
  done(
};

// Usage:
// var parser = new SimpleProtocol(
// source.pipe(parser)
// Now parser is 可读流（Readable stream） that will emit 'header'
// with the parsed header data.

javascript
// WARNING!  BROKEN!
net.createServer(function(socket) {

  // we add an 'end' 方法, but never consume the data
  socket.on('end', function() {
    // It will never get here.
    socket.end('I got your message (but didnt read it)\n'
  }

}).listen(1337

javascript
// Workaround
net.createServer(function(socket) {

  socket.on('end', function() {
    socket.end('I got your message (but didnt read it)\n'
  }

  // start the flow of data, discarding it.
  socket.resume(

}).listen(1337

javascript
var util = require('util'
var StringDecoder = require('string_decoder').StringDecoder;
var Transform = require('stream').Transform;
util.inherits(JSONParseStream, Transform

// Gets \n-delimited JSON  string data, and emits the parsed objects
function JSONParseStream() {
  if (!(this instanceof JSONParseStream))
    return new JSONParseStream(

  Transform.call(this, { readableObjectMode : true }

  this._buffer = '';
  this._decoder = new StringDecoder('utf8'
}

JSONParseStream.prototype._transform = function(chunk, encoding, cb) {
  this._buffer += this._decoder.write(chunk
  // split on newlines
  var lines = this._buffer.split(/\r?\n/
  // keep the last partial line buffered
  this._buffer = lines.pop(
  for (var l = 0; l < lines.length; l++) {
    var line = lines[l];
    try {
      var obj = JSON.parse(line
    } catch (er) {
      this.emit('error', er
      return;
    }
    // push the parsed object out to the readable consumer
    this.push(obj
  }
  cb(
};

JSONParseStream.prototype._flush = function(cb) {
  // Just handle any leftover
  var rem = this._buffer.trim(
  if (rem) {
    try {
      var obj = JSON.parse(rem
    } catch (er) {
      this.emit('error', er
      return;
    }
    // push the parsed object out to the readable consumer
    this.push(obj
  }
  cb(
};