node.js - Node 可读流问题

Question

我正在查看有关流的示例的这个非常漂亮的部分。

https://gist.github.com/joyrexus/10026630

可读的示例如下所示:

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

function Source(content, options) {
  Readable.call(this, options)
  this.content = content
}

inherits(Source, Readable)

Source.prototype._read = function (size) {
  if (!this.content) this.push(null)
  else {
    this.push(this.content.slice(0, size))
    this.content = this.content.slice(size)
  }
}

var s = new Source("The quick brown fox jumps over the lazy dog.")
console.log(s.read(10).toString())
console.log(s.read(10).toString())
console.log(s.read(10).toString())
console.log(s.read(10).toString())
console.log(s.read(10).toString())

// The quick 
// brown fox 
// jumps over
//  the lazy 
// dog.


var q = new Source("How now brown cow?")
q.pipe(process.stdout);

真正让我困惑的是，流的目的不是一次性将所有内容缓冲到内存中，以及提供一些异步性，因此并非所有有关管道传输流的内容都在事件循环的同一轮中处理。

  const writable = new stream.Writable({

        write: function(chunk, encoding, cb){

            console.log('data =>', String(chunk));

            cb();
        }

    });


    var readable = new stream.Readable({

        read: function(size){

            // what do I do with this? It's required to implement

        }
    });

    readable.setEncoding('utf8');

    readable.on('data', (chunk) => {
        console.log('got %d bytes of data', chunk.length, String(chunk));
    });

    readable.pipe(writable);

    readable.push('line1');
    readable.push('line2');
    readable.push('line3');
    readable.push('line4');

但我不明白的是，应该如何实现可读的 read 方法？

看起来我会以与示例完全不同的方式实现读取，因此似乎有些问题。

如何手动读取具有可读流的数据？

Answer 1

嗯，我知道有些事情有点不对劲，我相信这更接近规范的方法:

    const writable = new stream.Writable({

    write: function(chunk, encoding, cb){

        console.log('data =>', String(chunk));
        cb();
    },

    end: function(data){
        console.log('end was called with data=',data);
    }

});


var index = 0;
var dataSource = ['1','2','3'];

var readable = new stream.Readable({

    read: function(size){
        var data;
        if(data = dataSource[index++] ){
            this.push(data);
        }
        else{
            this.push(null);
        }
    }

});

readable.setEncoding('utf8');

readable.on('data', (chunk) => {
    console.log('got %d bytes of data', chunk.length, String(chunk));
});


readable.pipe(writable);

我不认为开发人员应该显式调用read； read 由开发者实现，但不被调用。希望这是正确的。我剩下的唯一问题是为什么在可写流中没有调用 end 。我也很好奇为什么 read 不接受回调。