流

Readable

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

// 创建一个只读流。
var rs = new Readable({
    // 构造函数参数中的 read 属性用于实现 _read() 方法。
    // 该方法在只读流 readable 事件被触发时执行。
    // 该方法必须实现。
    // 注意：_read() 和 read() 是两个不同的方法。
    read: function() {
        // .push() 方法将触发 readable 事件，
        // 故此例中，只读流将源源不断产生数据。
        this.push(Date.now().toString());
    }
});

rs.on('data',    function(data) {
    console.log('ondata,' + data.toString());
});

// 暂停流。
rs.pause();

// 在暂停状态下，执行 .read() 方法用于从流中读取若干数据。
// 这会触发 data 事件，但不会改变流的状态。
console.log('read,' + rs.read().toString());

Object Mode

const stream = require('stream');

const Readable = stream.Readable;
const Writable = stream.Writable;

// 创建一个只读流。
var rs = new Readable({
    // 创建一个对象流。
    // 默认情况下，都是字符串或二进制流，以 buffer 为数据处理单元。
    objectMode: true,
    read: function() {
        this.push(Date.now().toString());
    }
});

// 创建一个双向流。
var du = new stream.Duplex({
    objectMode: true,
    read: function() {},
    write: function(data, encoding, callback) {
        this.emit('data', data);
        callback();
    }
});

// 创建一个只写流。
var ws = new Writable({
    objectMode: true,
    write: function(data, encoding, callback) {
        this.emit('data', data);
        callback();
    }
});

// 注意其链式语法。
rs
    .pipe(du) // 返回 du
    .pipe(ws) // 返回 ws
    .on('data', function(data) { console.log(data.toString()); })
    ;

SEE

• stream-handbook by substack
  https://github.com/substack/stream-handbook

• read, _read, and readable: Readable Streams in Node.js
  https://harry.khanna.me/blog/readable-streams-in-node-js/