Google Reader

// A parser for a simple data protocol.
// The "header" is a JSON object, followed by 2 \n characters, and
// then a message body.
//
// Note: This can be done more simply as a Transform stream.  See below.

function SimpleProtocol(source, options) { //定义一个SimpleProtocol 类
  if (!(this instanceof SimpleProtocol))
    return new SimpleProtocol(options);

  Readable.call(this, options);//执行readable构造函数
  this._inBody = false; //_inbody表示是否开始接受body
  this._sawFirstCr = false; //表示是否看见第一个换行

  // source is a readable stream, such as a socket or file
  this._source = source; //将数据源赋值到this._source,source是一个readable stream

  var self = this;
  source.on('end', function() { //当数据源 end 事件触发，则 SimpleProtocol 实例push一个null，表示没有数据push到队列中
    self.push(null);
  });

  // give it a kick whenever the source is readable
  // read(0) will not consume any bytes
  source.on('readable', function() { //read(0) 表示不消费任何字节
    self.read(0);
  });

  this._rawHeader = []; //存放待格式化的头部的数组
  this.header = null; 
}

SimpleProtocol.prototype = Object.create( //SimpleProtocol继承readable类
  Readable.prototype, { constructor: { value: SimpleProtocol }});

SimpleProtocol.prototype._read = function(n) {
  if (!this._inBody) { //如果还没接受body，表示接受head
    var chunk = this._source.read(); //从source中读取chunck

    // if the source doesn't have data, we don't have data yet.
    //如果source还没有数据，那我们就push空字符串
    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++) { //开始逐个解析chunk中的字符串，查看是否包含2个连续的/n
      if (chunk[i] === 10) { // '\n'  //如果发现是/n
        if (this._sawFirstCr) { //如果上一个也是 /n 则这边为true
          split = i; //找到了连续了/n 表示头部结束了，跳出循环，记录位置
          break;
        } else { //如果上一个字符不是 /n ，则把 _sawFirstCr 设为true
          this._sawFirstCr = true;
        }
      } else { //如果本字符不是 /n 则 _sawFirstCr 设置为false
        this._sawFirstCr = false;
      }
    }

    if (split === -1) { //如果没有找到2个/n则继续等待，push空字符串，将chunk暂时保存
      // still waiting for the \n\n
      // stash the chunk, and try again.
      this._rawHeader.push(chunk);
      this.push('');
    } else {  //如果找到2个/n了，则表示以后就开始接受body
      this._inBody = true; //将inbody设置为true，表示今后开始接受body
      var h = chunk.slice(0, split); //将此次的chunk切分，头部放入待格式化数组
      this._rawHeader.push(h); 
      var header = Buffer.concat(this._rawHeader).toString(); //然后将待格式化数组中的内容组合
      try {
        this.header = JSON.parse(header); //转化js对象
      } 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 read queue so that our consumer will see it.
      var b = chunk.slice(split); 
      this.unshift(b);//此时将chunk中取出的b也就是body部分，再塞回队列头部

      // and let them know that we are done parsing the header.
      this.emit('header', this.header);  //触发header事件，并且把this.header作为参数传递过去
    }
  } else { //表示开始接受body
    // from there on, just provide the data to our consumer.
    // careful not to push(null), since that would indicate EOF.
    //从这里开始，表示对我们的消费者也就是api使用者提供数据了，注意不要push(null),因为null表示结束
    var chunk = this._source.read();
    if (chunk) this.push(chunk);
  }
};

// Usage:
var parser = new SimpleProtocol(source);
// Now parser is a readable stream that will emit 'header'
// with the parsed header data.
//这样parser就可以解析source的数据了，当header解析完毕，就会触发 header 事件

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.
});

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 = Object.create(
  Transform.prototype, { constructor: { value: SimpleProtocol }});

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(b);
    }
  } else {
    // from there on, just provide the data to our consumer as-is.
    this.push(b);
  }
  done();
};

var parser = new SimpleProtocol();
source.pipe(parser) //坑爹就是这里改变了

// Now parser is a readable stream that will emit 'header'
// with the parsed header data.

var http = require("http")
var options = {
  hostname: 'www.yahoo.com',
  port: 80,
  path: '/',
  method: 'GET'
};
var req = http.request(options, function(res) {
  //console.log('STATUS: ' + res.statusCode);
  //console.log('HEADERS: ' + JSON.stringify(res.headers));
  res.setEncoding('utf8');
  res.on('readable', function () {
      console.log(res.read())//这里就无须监听data事件然后拼字符串buffer之类了，直接监听可读事件，然后调用res.read()
  });
});
req.on('error', function(e) {
  console.log('problem with request: ' + e.message);
});
// write data to request body
req.write('data\n');
req.write('data\n');
req.end();