node.js stream

stream 类型 Readable, Writable, Duplex, Transform

Readable Stream

Readable 能够看作消息队列的生产者。Readable 有两种模式，流动模式和非流动模式。

流动模式

流动模式用户监听 data 事件，能够自行决定暂停和者恢复。读如数据先写入 buffer， 若是 buffer 值高于 highWaterMark
中止读取。source => buffer => customer，举个栗子，将一个大水缸 A 的水，引入另外一个水缸 B 。咱们在水缸 A 接一根管子，管子另外一头接一水桶（缓存池)，水桶另外一端接流出水管，流向 B， B 有一个开关。一旦水桶达到highWaterMark，水缸 A 将暂停向水桶中注水。

非流动模式

非流动模式下用户须要自行读取 buffer 的数据，读取大小自行决定，故而也再也不须要 pause 和 resume, readable 事件触发后便可自行调用 read(size) 读取数据. readable 事件触发的条件是有数据到达缓存池，因此须要在消费者自行加条件锁，以避免消费不及时形成问题。

Writable Stream

原理和 Readable Stream 相似, 再也不赘述。

pipe

Stream 都自带管子(pipeline)，因此流均可以经过管子相互链接。你能够将一个 Readable Stream 直接连到一个 Writable Stream 上。

const readable = fs.createReadableStream('/tmp/input.txt');
const writable = fs.createWritableStream('/tmp/output.txt');

readable.pipe(writable);

Readable Stream 是生产者， Writable Stream 是消费者，不能本末倒置写成writable.pipe(readable)

双工流

• Duplex
• Transform

Duplex 读写流捆绑在一个主体上，即该类型 Stream 即有 Readable Stream 又有Writable Stream， 可是两种流相互独立互不关联，至关于把两个对象继承在一个对象上了。

Transform 这个流颇有用，有时候你获得一个Writable Stream，立刻要把Writable Stream转换成 Readable Stream pipe 到另外一管道中去（上面说到了，Writable Stream 不能做为生产者)。

Transform 须要实践 Transform. prototype._transform 方法

const { Transform } = require('stream');

class TransformStream extends Transform {
  constructor(options) {
    super(options);
  }

  _transform (data, encoding, callback) {
    callback(null, data);
  }
}

举个栗子：从 upyun dump 图片下来再上传，下载获得的是 WritableStream， 上传须要的是 ReadableStream。若是不使用 transform 得先将 getFile 得结果保存，在建立一个 readable stream， 再有 readable stream pipe. 若是使用 Transform stream 将很简单.

async function foo() {
    try {
        const transform = new TransformStream();
        // 不能在这里 await Transform 只是转换 stream 从 writable 到 readable， 并无 pipe 数据
        const pass = client.getFile(path, transform);
        // 这里不能 await， 若是使用 await upyun getFile 返回错误将没法处理
        const pipe = transform.pipe(request({
            uri: someuri,
            method: 'post',
            headers: {
              'Content-Type': 'image/jpeg'
            },
            encoding: null,
        });
        
        const through = await pass; // hanlde getFile promise 见 upyun node sdk 源码。
        if (!through) { // 可能 getFile 不存在或者 upyun 服务限制上传
            return false;
        }
    
        const reponse = await pipe; // handle pipe request
    } catch(err) {
        console.log(err);        
    }
}