为了实现可读流,引用Readable接口并用它构造新对象
var stream = require('stream');
var util = require('util');
util.inherits(Counter, stream.Readable);
function Counter(options) {
stream.Readable.call(this, options);
this._index = 0;
}
Counter.prototype._read = function() {
if(this._index++<3){
this.push(this._index+'');
}else{
this.push(null);
}
};
var counter = new Counter();
counter.on('data', function(data){
console.log("读到数据: " + data.toString());//no maybe
});
counter.on('end', function(data){
console.log("读完了");
});
为了实现可写流,我们需要使用流模块中的Writable构造函数。 我们只需给Writable构造函数传递一些选项并创建一个对象。唯一需要的选项是write函数,该函数揭露数据块要往哪里写。
var stream = require('stream');
var util = require('util');
util.inherits(Writer, stream.Writable);
let stock = [];
function Writer(opt) {
stream.Writable.call(this, opt);
}
Writer.prototype._write = function(chunk, encoding, callback) {
setTimeout(()=>{
stock.push(chunk.toString('utf8'));
console.log("增加: " + chunk);
callback();
},500)
};
var w = new Writer();
for (var i=1; i<=5; i++){
w.write("项目:" + i, 'utf8');
}
w.end("结束写入",function(){
console.log(stock);
});
const stream = require('stream')
var index = 0;
const readable = stream.Readable({
highWaterMark: 2,
read: function () {
process.nextTick(() => {
console.log('push', ++index)
this.push(index+'');
})
}
})
const writable = stream.Writable({
highWaterMark: 2,
write: function (chunk, encoding, next) {
console.log('写入:', chunk.toString())
}
})
readable.pipe(writable);
有了双工流,我们可以在同一个对象上同时实现可读和可写,就好像同时继承这两个接口。 重要的是双工流的可读性和可写性操作完全独立于彼此。这仅仅是将两个特性组合成一个对象。
const {Duplex} = require('stream');
const inoutStream = new Duplex({
write(chunk, encoding, callback) {
console.log(chunk.toString());
callback();
},
read(size) {
this.push((++this.index)+'');
if (this.index > 3) {
this.push(null);
}
}
});
inoutStream.index = 0;
process.stdin.pipe(inoutStream).pipe(process.stdout);
const {Transform} = require('stream');
const upperCase = new Transform({
transform(chunk, encoding, callback) {
this.push(chunk.toString().toUpperCase());
callback();
}
});
process.stdin.pipe(upperCase).pipe(process.stdout);
默认情况下,流处理的数据是Buffer/String类型的值。有一个objectMode标志,我们可以设置它让流可以接受任何JavaScript对象。
const {Transform} = require('stream');
let fs = require('fs');
let rs = fs.createReadStream('./users.json');
rs.setEncoding('utf8');
let toJson = Transform({
readableObjectMode: true,
transform(chunk, encoding, callback) {
this.push(JSON.parse(chunk));
callback();
}
});
let jsonOut = Transform({
writableObjectMode: true,
transform(chunk, encoding, callback) {
console.log(chunk);
callback();
}
});
rs.pipe(toJson).pipe(jsonOut)
[
{"name":"zfpx1","age":8},
{"name":"zfpx2","age":9}
]
readable.unshift() 方法会把一块数据压回到Buffer内部。 这在如下特定情形下有用: 代码正在消费一个数据流,已经"乐观地"拉取了数据。 又需要"反悔-消费"一些数据,以便这些数据可以传给其他人用。
const {Transform} = require('stream');
const { StringDecoder } = require('string_decoder');
let decoder = new StringDecoder('utf8');
let fs = require('fs');
let rs = fs.createReadStream('./req.txt');
function parseHeader(stream, callback) {
let header = '';
rs.on('readable',onReadable);
function onReadable() {
let chunk;
while(null != (chunk = rs.read())){
const str = decoder.write(chunk);
if(str.match(/\r\n\r\n/)){
const split = str.split(/\r\n\r\n/);
console.log(split);
header+=split.shift();
const remaining = split.join('\r\n\r\n');
const buf = Buffer.from(remaining,'utf8');
rs.removeListener('readable', onReadable);
if(buf.length){
stream.unshift(buf);
}
callback(null,header,rs);
}else{
header += str;
}
}
}
}
parseHeader(rs,function(err,header,stream){
console.log(header);
stream.setEncoding('utf8');
stream.on('data',function (data) {
console.log('data',data);
});
});
Host: www.baidu.com
User-Agent: curl/7.53.0
Accept: */*
name=zfpx&age=9