怎样利用node做出可读流动模式

let fs = require("fs");
let rs = fs.createReadStream("./a.js", {
 highWaterMark: 3,
 encoding: "utf8",
 autoClose: true,
 start: 0,
 end: 9
});
rs.on("open", () => {console.log("open");});
rs.on("close", () => {console.log("close");});
rs.on("data", data => {
 console.log(data);
 rs.pause();//暂停读取 此时流动模式为暂停模式
});
setInterval(() => {
 rs.resume();//重新设置为流动模式，开始读取数据
}, 1000);
rs.on("end", () => { console.log("end"); });
rs.on("error", err => { console.log(err); });

let fs = require('fs');
let EventEmitter = require('events');
class ReadStream extends EventEmitter {
}

let rs = fs.createReadStream("./a.js", {
 highWaterMark: 3,
 encoding: "utf8",
 autoClose: true,
 start: 0,
 end: 9
});

constructor(path, options = {}) {
 super();
 this.path = path; //指定要读取的文件地址
 this.highWaterMark = options.highWaterMark || 64 * 1024;
 this.autoClose = options.autoClose || true; //是否自动关闭文件
 this.start = options.start || 0; //从文件哪个位置开始读取
 this.pos = this.start; // pos会随着读取的位置改变
 this.end = options.end || null; // null表示没传递
 this.encoding = options.encoding || null;// buffer编码
 this.flags = options.flags || 'r';
 this.flowing = null; // 模式开关
 this.buffer = Buffer.alloc(this.highWaterMark);// 根据设置创建一个buffer存储读出来的数
 this.open();
}

fs.open(filename,flags,[mode],callback);
//实例
fs.open('./1,txt','r',function(err,fd){});

open() {
 fs.open(this.path, this.flags, (err, fd) => { 
 //fd标识的就是当前this.path这个文件，从3开始(number类型)
 if (err) {
 if (this.autoClose) { // 如果需要自动关闭则去关闭文件
 this.destroy(); // 销毁(关闭文件，触发关闭事件)
 }
 this.emit('error', err); // 如果有错误触发error事件
 return;
 }
 this.fd = fd; // 保存文件描述符
 this.emit('open', this.fd); // 触发文件的打开的方法
 });
}

rs.on("data", data => {
 console.log(data);
});

// 监听newListener事件，看是否监听了data事件，如果监听了data事件的话，就开始启动流动模式，读取文件中的内容
this.on("newListener", type => {
 if (type === "data") {
 // 开启流动模式，开始读取文件中的内容
 this.flowing = true;
 this.read();
 }
});

read() {
 // 第一次读取文件的话，有可能文件是还没有打开的，此时this.fd可能还没有值
 if (typeof this.fd !== "number") {
 // 如果此时文件还是没有打开的话，就触发一次open事件，这样文件就真的打开了，然后再读取
 return this.once("open", () => this.read());
 }
 // 具体每次读取多少个字符，需要进行计算,因为最后一次读取倒的可能比highWaterMark小
 let howMuchRead = this.end ? Math.min(this.end - this.pos + 1, this.highWaterMark) : this.highWaterMark;
 fs.read(this.fd, this.buffer, 0, howMuchRead, this.pos, (err, byteRead) => {
 // this.pos 是每次读取文件读取的位置，是一个偏移量，每次读取会发生变化
 this.pos += byteRead;
 // 将读取到的内容转换成字符串串，然后通过data事件，将内容发布出去
 let srr = this.encoding ? this.buffer.slice(0, byteRead).toString(this.encoding) : this.buffer.slice(0, byteRead);
 // 将内容通过data事件发布出去
 this.emit("data", srr);
 // 当读取到到内容长度和设置的highWaterMark一致的话，并且还是流动模式的话，就继续读取
 if ((byteRead === this.highWaterMark) && this.flowing) {
 return this.read();
 }
 // 没有更多的内容了，此时表示文件中的内容已经读取完毕
 if (byteRead < this.highWaterMark) {
 // 读取完成，发布end方法，并关闭文件
 this.emit("end");
 this.destory();
 }
 });
}

destory() {
 if (typeof this.fd !== "number") {
 // 发布close事件
 return this.emit("close");
 }
 // 将文件关闭，发布close事件
 fs.close(this.fd, () => {
 this.emit("close");
 });
}

pause() {
 this.flowing = false;// 将流动模式设置成暂停模式，不会读取文件
}
resume() {
 this.flowing = true;//将模式设置成流动模式，可以读取文件
 this.read();// 重新开始读取文件
}