版本：8.1.3

实现一个流API

stream模块的API被设计成能够很容的使用javascript的原型继承模式来实现streams。
首先，stream的开发者必须先声明一个新的javascript类，并且继承以下四个基础stream类中的一个，并确保他们适当的调用父类的构造函数。

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

class MyWritable extends Writable {
  constructor(options) {
    super(options);
    // ...
  }
}

并且新的stream类必须实现所继承的类的一个或者更多的特定方法。具体如下表所示：

注意实现stream的代码绝不能调用stream"public"方法，因为那是为了给消费者使用的。如果那样可能会对应用在在使用流时导致不利的影响。

一个简单的构造函数

虽然有很多简单的实例，但是也能够直接构造一个stream而不依赖任何继承。能够直接使用stream.Writable,stream.Readable,stream.Duplex或者stream.Transform来创建实例对象并且传递适当的方法作为构造函数的options。
例如：

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

const myWritable = new Writable({
  write(chunk, encoding, callback) {
    // ...
  }
});

实现一个可写流/Implementing a Writable Stream

stream.Writable被用来扩展实现一个Writable流。

自定义一个Writable流必须调用new stream.Writable([options])构造函数并且实现writable._write()方法。writable._wirtev()方法也许也要实现。

构造函数:new stream.Writable([options])

• options <Object>

  • highWaterMark <number> stream.write()开始返回false的最高Buffer限制。默认值为16384(16kb)，对象模式是16。

  • decodeStrings <boolean> 传入方法stream._write()之前是否对字符串解码，默认是true。

  • objectMode <boolean> stream.write(anyOjb)方法是否有效。当stream实现支持，并且设置该值为true，那么将能够写进一个javascript的值而不是字符串，Bufer，或者Uint8Array。默认值为false。

  • write <Function> 实现stream._write()方法。

  • writev <Function> 实现stream._writev()方法。

  • destroy <Function> 实现stream._destroy()方法。

  • final <Function> 实现stream._final()方法。

实例：

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

class MyWritable extends Writable {
  constructor(options) {
    // Calls the stream.Writable() constructor
    super(options);
    // ...
  }
}

writable._write(chunk,encoding,callback)

• chunk <Buffer>|<string>|<any> 被写入的chunk，在不设置decodeStrings项为false的情况下将会是Buffer。或者stream在对象模式下工作。

• encoding <string> 如果chunk是一个字符串，那么encoding是该字符串的格式。如果chunk是Buffer，或者stream是对象模式，encoding将会被忽视。

• callback <Function> (会有一个error参数)，在chunk被完全提供时，调用这个函数。

一个Writable stream必须实现和提供一个writable._write()方法来发送数据给潜在的源。

注意 Transform stream提供他们自己的该方法writable._write()的实现。

注意该方法不能够被应用直接调用，它在子类中实现，并且只能被Writable类内部的方法调用。

回调函数必须被调用，用来说明数据写入成功还是出现了错误。第一个参数必须是Error，如果写入失败那么他是一个错误对象，如果成功那么是null。

我们应该着重注意到，writable.write()的调用发生在writable._write()被调用和回调函数被调用之时，此时写入的数据将被缓存。一旦回掉函数 被调用，stream将会发出drain事件，如果stream被实现为能够一次处理多chunk的功能，那么writable._writev()方法应该被实现。

如果decodeStrings属性在构造函数的options中设置，那么chunk应该是一个字符串而不是Buffer，此时encoding将会检测字符串字符的编码。这个可以用来优化被传入的字符串已知其编码格式。如果decodeStrings明确设置为false，那么encoding参数会被安全的忽视，并且chunk将被保持为同一个对象然后传入.write()。

writable._write()被设置为有一个前置下划线，因为这是类内部定义的，并且不能够被外部程序直接调用。

writable._writev(chunks,callback)

• chunks <Array> 被写入的chunk，每个chunk有这样的格式:{chunk:…, encoding:…}。

• callback <Function> (会有一个error参数)，在chunk被完全提供时，调用这个函数。

注意该方法不能够被应用直接调用，它在子类中实现，并且只能被Writable类内部的方法调用。

在实现了writable._write()方法后也相互会实现writable._writev()方法，用来实现一次处理多个chunk。

writable._destroy(err,callback)

• err <Error> 一个错误。

• callback <Function> 一个带有可操作的错误的参数，在writable被毁掉时调用。

writable._final(callback)

• callback <Function> 当你结束等待的书局时调用。（可添加一个可选的错误参数。）

注意 _final()不能够被直接调用，在实现子类时，他只能被内部的Writable类调用。

可选的函数会在流结束之前调用，finish事件会在回调函数执行之后触发。这在关闭源或者在流结束前写入缓存数据时是有用的。

写入时发生错误

建议在使用writable._write()以及writable._writev()时在回调函数的第一个参数写入err，以防执行时发生错误。这个会导致error时间触发。把错误抛弃可能会导致依赖流的一些行为发生意外或矛盾。使用回掉函数保证处理可遇到的错误。

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

const myWritable = new Writable({
  write(chunk, encoding, callback) {
    if (chunk.toString().indexOf('a') >= 0) {
      callback(new Error('chunk is invalid'));
    } else {
      callback();
    }
  }
});

实现一个readable 流

stream.Readable类被用来实现一个Readable1流。
自定义的Readable流必须调用new stream.Readable([options])构造函数，并且实现readable._read()方法。

new stream.Readable([options])

• options <Object>

  • highWaterMark <number>在内部buffer存储的最大字节数，数据为在打包读取底层数据源之前。默认为16384(16kb),objectMode模式为16。

  • encoding <string> 如果设置，那么buffer将被格式化为指定格式的字符串。默认值为null。

  • objectMode <boolean> 设置这个流的表现形式是流还是对象。决定着stream.read(n)返回的是单一的值还是一个一定大小的Buffer。默认是false。

  • read <Function> stream._read()方法的实现。

  • destory <Function> stream._destroy()方法的实现。

例子：

es6

const {Readable} = require('stream')

class MyReadable extends Readable{
    constructor(options){
        //调用stream.Readable(options)构造函数
        super(options)
        //。。。
    }
}

或者ES6版本之前的构造函数语法风格
es5

const {Readable} = require('stream')
const util = require('util')

function MyReadable(options){
    if(!(this instanceof MyReadable))
        return new MyReadable(options)
    Readable.call(this, options)
}
util.inherits(MyReadable, Readable)

readable._read(size)

• size <number> 异步读取的字节大小。

注意 这个函数一定不能被应用的代买直接调用，他应该在子类中实现，并且只能够被内部Readable类方法调用。

所有的Readable流的实现必须提供一个readable._read()方法的实现，用来取得来自底部的数据源数据。

当调用readable._read()如果自源的数据可用，那么应该使用readable.push(dataChunk)开始把数据推进读取队列。_read()应该持续凑够源读取数据然后推进队列，直至readable.push()返回false。只有再次调用readable._read()那么原先停止的流才会再次恢复添加数据到队列。

注意一旦readable._read()方法被调用，那么在readable.push()被调用前是不用再次调用的。

size参数是一个公告，在read是个单一操作，那么返回的数据可以使用size来限制大小。其他的实现也许会忽视这个参数，并且仅在数据可读取是才提供。调用stream.push(chunk)时没必要等待size足够再调用。

readable._read()是底层源的体现设置，因为只能够在类的内部定义它，所以永远不能在其他用户的程序里直接调用。

readable.push(chunk[,encoding])

• chunk <Buffer>|<Uint8Array>|<string>|<null>|<any>Chunk的数据要被推进读取队列。chunk必须是一个字符串，Buffer或者Uint8Array。如果是对象模式，那么chunk或许是个js值。

• encoding <string> 对chunk的字符串编码。不许是一个有效的Buffer编码格式，例如'utf8'或者'ascii'。

• returns <boolean>`true 如果添加的chunk能够继续push;false`反之。

如果chunk是一个Buffer,Uint8Array或者string，那么chunk将被添加到内部队列等待流用户来消费。传递的chunk是一个null，说明到了流的结束，之后便不会有数据被写入。

当Readable在paused模式下操作时，readable.push()方法能够添加从readable.read()（被readable方法触发）方法中读取的数据。

当Readable在flowing模式下操作时，readable.push()方法更够在data事件触发下传递数据。

readable.push()被设计的很灵活，例如，当封装从pause/resume机制或者回到函数的chunk提供的一些数据这些底层源能够 被自定义的readale实例来封装，如下：

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

class SourceWrapper extends Readable {
  constructor(options) {
    super(options);

    this._source = getLowlevelSourceObject();

    // Every time there's data, push it into the internal buffer.
    this._source.ondata = (chunk) => {
      // if push() returns false, then stop reading from source
      if (!this.push(chunk))
        this._source.readStop();
    };

    // When the source ends, push the EOF-signaling `null` chunk
    this._source.onend = () => {
      this.push(null);
    };
  }
  // _read will be called when the stream wants to pull more data in
  // the advisory size argument is ignored in this case.
  _read(size) {
    this._source.readStart();
  }
}

注意readable.push()只能够被Readable的实现者调用，并且只能够在readable._read()里面。

读取时发生错误/Errors While Reading

建议在使用readable._read()方法发生错误时触发error事件，而且不是抛弃。在readable_read()里面抛弃错误会在操作流是否依赖flowing或者parse模式造成矛盾，以及意外。使用error事件确保一致以及处理可预测的错误。

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

const myReadable = new Readable({
  read(size) {
    if (checkSomeErrorCondition()) {
      process.nextTick(() => this.emit('error', err));
      return;
    }
    // do some work
  }
});

一个流的例子

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

class Counter extends Readable {
  constructor(opt) {
    super(opt);
    this._max = 1000000;
    this._index = 1;
  }

  _read() {
    const i = this._index++;
    if (i > this._max)
      this.push(null);
    else {
      const str = '' + i;
      const buf = Buffer.from(str, 'ascii');
      this.push(buf);
    }
  }
}

实现一个双向流/Implementing a Duplex Stream

一个Duplex流及实现了可读流又实现了可写流，例如TCP socket连接（这样翻译顺点）。

由于javaScript不能多个继承，stream.Duplex类被用来实现双向流/Duplex的一个类（毕竟同时继承stream.Readable和stream.Writable类是相抵触的）。

注意 stream.Duplex是原型继承stream.Readable类，寄生继承stream.Writable，但是instanceof将会正确表示基于二者，因为该类覆盖了stream.Writable的Symbole.hasInstance方法。

自定义的Duplex流必须调用new stream.Duplex([options])构造函数以及实现readable._read()和writable._write()方法。

new stream.Duplex([options])

• options <object> 同时传给可读流和和谐刘的构造函数，并且额外有下面的域。

  • allowHalfOpen <boolean> 默认true，如果设置为false，流会在可写流结束时停止可读流，反之亦然。

  • readableObjectMode <boolean> 默认false。 设置流的可读流模式为objectMode，如果为true则没有效果。

  • writableObjectMode <boolean> 默认false。 设置流的可写流模式为objectMode，如果为true则没有效果。

例子：

lang:ES6

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

class MyDuplex extends Duplex {
  constructor(options) {
    super(options);
    // ...
  }
}

或者使用es6版本之前的构造函数风格。

const { Duplex } = require('stream');
const util = require('util');

function MyDuplex(options) {
  if (!(this instanceof MyDuplex))
    return new MyDuplex(options);
  Duplex.call(this, options);
}
util.inherits(MyDuplex, Duplex);

一个双向流的例子/An Example Duplex Stream

下面是一个Duplex流的简单示例，虽然Node.js的streams的API是不兼容的，但是我们向这个包了个底层源对象写入数据，同时也可用从其中读取数据。该例子演示向Writable接口中写入数据，从Readable接口中读取数据。

lang:ES6

const { Duplex } = require('stream');
const kSource = Symbol('source');

class MyDuplex extends Duplex {
  constructor(source, options) {
    super(options);
    this[kSource] = source;
  }

  _write(chunk, encoding, callback) {
    // The underlying source only deals with strings
    if (Buffer.isBuffer(chunk))
      chunk = chunk.toString();
    this[kSource].writeSomeData(chunk);
    callback();
  }

  _read(size) {
    this[kSource].fetchSomeData(size, (data, encoding) => {
      this.push(Buffer.from(data, encoding));
    });
  }
}

这个语法形式中最重要的是尽管可读流和可写流共存于一个对象实例，但是对它们进行的相对对立的操作。

一个双向流的对象模式/Object Mode Duplex Streams

在Duplex流中，可以根据设置中的readableObjectMode或者writableObjectMode项把可读流或者可写流其中之一设置为对象模式。

下面的例子，是一个Transform流（Duplex流的一种）的例子，其创建的可写流是一个对象模式，接受一个可读流传来的数据，该可读流接受javaScript数字并且转换成十六进制的字符串。

lang:ES6

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

// All Transform streams are also Duplex Streams
const myTransform = new Transform({
  writableObjectMode: true,

  transform(chunk, encoding, callback) {
    // Coerce the chunk to a number if necessary
    chunk |= 0;

    // Transform the chunk into something else.
    const data = chunk.toString(16);

    // Push the data onto the readable queue.
    callback(null, '0'.repeat(data.length % 2) + data);
  }
});

myTransform.setEncoding('ascii');
myTransform.on('data', (chunk) => console.log(chunk));

myTransform.write(1);
// Prints: 01
myTransform.write(10);
// Prints: 0a
myTransform.write(100);
// Prints: 64

实现一个Transform流/Implementing a Transform Stream

一个Transfrom流是一个把输入重新计算过的Duplex流。包括能够压缩的[zlib][]流或者对数据加密解密的crypto流。

注意 streams没有输入和输出的数据必须大小一致，chuank的数量一致，或者同时送达，的规范要求。例如，一个Hash流在输入结束时只会有一个输出chunk。一个zlib流产生的输出可能比输入的大也可能小。

stream.Transform类用来扩展实现Transform类。

stream.Transform类原型继承于stream.Duplex类，并且需要实现自己的writable._write()和readable._read()方法。自定义的Transform类还必须实现transform._transform()方法，必要时也需要实现transform._flush()方法。

注意 在使用Transfrom流，往流中写入数据时要特别小心，因为如果Reable一方没有消费者是可能到导致Writable一方的流暂停。

new stream.Transform([options])

• options <Object> 传递给Writable和Readalbe构造函数，同时也包含以下的域。

  • transform <Function> 实现stream._transform()方法。

  • flush <Function> 实现stream._flush()方法。

例子：

ES6

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

class MyTransform extends Transform {
  constructor(options) {
    super(options);
    // ...
  }
}

es6之前版本的构造函数风格。

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

function MyTransform(options) {
  if (!(this instanceof MyTransform))
    return new MyTransform(options);
  Transform.call(this, options);
}
util.inherits(MyTransform, Transform);

事件：finish和end

事件finish和end来自继承的stream.Writable以及stream.Readable类。finish在stream.end()调用是触发，此时所有chunk已经被stream._transform()处理完毕。end事件在所有的数据被输出，执行transform._flusn()方法的回调函数时触发。

transform._flush(callback)

• callback <Function> 回调函数(另外有可选的error，和data参数)，当数据流完时执行。

注意 这个方法绝对不能应用的代码直接调用。其只应该被继承的子类的Readable类中的方法调用。

在一些情况下，一个transform操作需要发出一些额外的数据在流的结尾。例如，一个zlib压缩流会保存一些有关内部状态的数据来优化压缩输出。当流结束，然而额外的信息需要被发出来，至此整个压缩过程才算结束。

自定义的Transform类实现也许会实现transform._flush()方法。当没有更多的数据需要被写入时会被触发，但是会在end事件发出告诉Readable流结束之前。

实现transform._flush()方法，可能会在内部适当调用readable.push()方法零次或多次。在flush操作结束后必须调用回调函数。

transform._flush()方法有一个前置的下划线，表明这是一个类内部定义的，永远不应该被用户的程序直接调用。

transform._transform(chunk,encoding,callback)

• chunk <Buffer>|<string>|<any> 需要被转换的chunk。在不设置decodeString参数为false或者不为对象模式情况下，用还是buffer.

• encoding <string> 如果chunk是字符串，那么该参数设置其编码类型。如果chunk是buffer，那么这个值是'buffer'，这种情况下会忽略。

• calback <Function> 一个回调函数（带有可选的error和data参数），提供的chunk被处理后会被调用。

注意 这个方法绝对不能应用的代码直接调用。其只应该被继承的子类的Readable类中的方法调用。

所有的Transofrom流的实现都必须提供一个_transform()方法类接受输入，产生输出。transfrom._transform()方法实现处理被写入的字节，计算输出，然后从readable部分使用readable.push()传出。

transform.push()方法可能会被多次调用，根据input chunk来生成output。取决于chunk能产生多少output结果。

有时也可能发生无论什么样的数据chunk也产生不了output。

当前的chunk被消耗完了之后必须调用回调函数。回调函数的第一个参数必须是Error对象，如果发生错误该对象会被传递，否则可以是null。如果有第二个参数，那么它将被推荐到readable.push()方法。简单来说，如下：

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

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

此外，该方法也是一个类的内部方法，不应该被用户程序直接调用。

类：stream.PassThough/Class:stream.PassThrough

stream.PassThough类是一个Transform流的一个琐碎实现，能够简单的传递input字节到output。本来的目的是检验和测试。但是也有一些场景，例如整理小说章节分类。