nodejs stream

简单的流读写实现

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

var rs = Readable();
var ws = Writable();

var c = 97 - 1;

rs._read = function () {
if (c >= 'z'.charCodeAt(0)) return rs.push(null);

setTimeout(function () {
rs.push(String.fromCharCode(++c));
}, 1000);
};

rs.pipe(ws);

ws._write = function (chunk, enc, next) {
process.stdout.write(chunk);
next();
};

process.on('exit', function () {
console.error('\n_read() called ' + (c - 97) + ' times');
});

process.stdout.on('error', process.exit);

Transform Stream 实现字符过滤

const Transform = require('stream').Transform;
const util = require('util');
const fs= require('fs');

function MyTransform(filter) {
if (!(this instanceof MyTransform))
return new MyTransform(null);
Transform.call(this, null);
this.temp = new Buffer(6);//utf-8最多6个字节
this.filter = filter;//要过滤的字符
}
util.inherits(MyTransform, Transform);

MyTransform.prototype._transform = function (data, encoding, callback) {
let len=0;//数据长度（单位字节） 用于递减
let blen=0;//数据长度（单位字节）最终从temp提取的长度
let c=null;//解析出的字符

for (let b of data) {
//如果只有一个字节则其最高二进制位为0；如果是多字节，其第一个字节从最高位开始，连续的二进制位值为1的个数决定了其编码的位数
if(len == 0){
if(b<128){
len=1;
}else if(b<224){
len=2;
}else if(b<240){
len=3;
}else if(b<248){
len=4;
}else if(b<252){
len=5;
}else if(b<254){
len=6;
}else{
continue;
}
blen=len;
}else{
//其余各字节均以10开头,这边是为了排错
if( b<128 || 191<b){
len=0;
blen=0;
continue;
}
}

len--;
this.temp[blen-len-1]=b;

if(len < 1){
c = this.temp.slice(0, blen);
if(c.toString()!=this.filter){
this.push(c);
}
}
}
callback();
};

//test

const is = fs.createReadStream('a.txt');

//过滤字符‘a’
is.pipe(new MyTransform('a')).pipe(process.stdout);