【问题标题】：How to implement a writable stream如何实现可写流
【发布时间】：2014-02-24 19:50:00
【问题描述】：

我想将数据从 amazon kinesis 流传输到 s3 日志或 bunyan 日志。

该示例适用于文件写入流或标准输出。我将如何实现自己的可写流？

//this works
var file = fs.createWriteStream('my.log')
kinesisSource.pipe(file)

说它没有“on”方法是行不通的

var stream = {}; //process.stdout works however
stream.writable = true;
stream.write =function(data){
    console.log(data);
};
kinesisSource.pipe(stream);

我必须为自己的自定义可写流实现哪些方法，文档似乎表明我需要实现“写入”而不是“开启”

【问题讨论】：

标签： node.js node.js-stream

【解决方案1】：

这是一个直接来自 nodejs 文档的示例
https://nodejs.org/api/stream.html#an-example-writable-stream

const { Writable } = require('stream');
class MyWritable extends Writable {
  _write(chunk, encoding, callback) {
    if (chunk.toString().indexOf('a') >= 0) {
      callback(new Error('chunk is invalid'));
    } else {
      callback();
    }
  }
}

【讨论】：

【解决方案2】：

要创建自己的可写流，您有三种可能性。

创建自己的类

为此，您需要 1) 扩展 Writable 类 2) 在您自己的构造函数中调用 Writable 构造函数 3) 在流对象的原型中定义 _write() 方法。

这是一个例子：

var stream = require('stream');
var util = require('util');

function EchoStream () { // step 2
  stream.Writable.call(this);
};
util.inherits(EchoStream, stream.Writable); // step 1
EchoStream.prototype._write = function (chunk, encoding, done) { // step 3
  console.log(chunk.toString());
  done();
}

var myStream = new EchoStream(); // instanciate your brand new stream
process.stdin.pipe(myStream);

扩展一个空的 Writable 对象

您可以实例化一个空的Writable 对象并实现_write() 方法，而不是定义一个新的对象类型：

var stream = require('stream');
var echoStream = new stream.Writable();
echoStream._write = function (chunk, encoding, done) {
  console.log(chunk.toString());
  done();
};

process.stdin.pipe(echoStream);

使用简化的构造函数 API

如果你使用 io.js，你可以使用simplified constructor API:

var writable = new stream.Writable({
  write: function(chunk, encoding, next) {
    console.log(chunk.toString());
    next();
  }
});

在 Node 4+ 中使用 ES6 类

class EchoStream extends stream.Writable {
  _write(chunk, enc, next) {
    console.log(chunk.toString());
    next();
  }
}

【讨论】：

支持对象模式替换chunk.toString() per chunk.toString ? chunk.toString() : chunk
这一行是做什么的："util.inherits(EchoStream, stream.Writable); // step 1 "?
它使EchoStream 成为stream.Writable 的“子类”：它的原型方法继承自它，并且stream.Writable 可以使用super_ 属性访问。请参阅the documentation 了解更多信息。
“创建自己的类”比“扩展一个空的可写对象”有什么好处？

【解决方案3】：

其实创建一个可写流是很简单的。这是一个例子：

var fs = require('fs');
var Stream = require('stream');

var ws = new Stream;
ws.writable = true;
ws.bytes = 0;

ws.write = function(buf) {
   ws.bytes += buf.length;
}

ws.end = function(buf) {
   if(arguments.length) ws.write(buf);
   ws.writable = false;

   console.log('bytes length: ' + ws.bytes);
}

fs.createReadStream('file path').pipe(ws);

另外，如果您想创建自己的课程，@Paul 给出了一个很好的答案。

【讨论】：

Cannot read property 'length' of undefined... at the buf.length in ws.write function definition
应该是 buf.length
这曾经可能是正确的，但现在不再正确了。 Node 文档现在说：“stream.Writable 类被扩展为实现 Writable 流。自定义 Writable 流必须调用新的 stream.Writable([options]) 构造函数并实现 writable._write() 和/或 writable._writev （）方法。” nodejs.org/api/…