fs.open(path.resolve(__dirname, "name.txt"), "r", function (err, fd) {
fs.open(path.resolve(__dirname, "copy.txt"), "w", 438, function (err, wfd) {
let readOffset = 0;
let writeOffset = 0;
// ------
~(function next() {
fs.read(fd, buf, 0, buf.length, readOffset, function (err, bytesRead) {
readOffset += bytesRead;
fs.write(wfd, buf, 0, bytesRead, writeOffset, function (err, written) {
// 写入的个数
writeOffset += written;
if (bytesRead < buf.length) {
// 每次判断一下读取的个数 是否小于 buffer的长度
return console.log("拷贝完毕");
}
next();
});
});
})();
// ------
});
});
// 读写需要做分离， 以上代码是耦合在一起的，没有通用性， 如果我只希望对文件做读取操作？ 只希望写入内容？
// 解开耦合？ 通过发布订阅模式解耦合
class MiniEventEmitter {
constructor() {
// 存储事件 -> 回调列表
this._events = Object.create(null);
}
on(event, listener) {
if (!this._events[event]) {
this._events[event] = [];
}
this._events[event].push(listener);
}
emit(event, ...args) {
const listeners = this._events[event];
if (!listeners) return;
// 同步依次执行
listeners.forEach(fn => fn(...args));
}
}
const fs = require("fs");
const path = require("path");
const WriteStream = require("./WriteStream");
const ReadStream = require("./ReadStream");
const rs = new ReadStream(path.resolve(__dirname, "copy.txt"), {
flags: "r", // 默认不写就是 r 表示我要读取
autoClose: true,
emitClose: true,
start: 0,
highWaterMark: 4, // 写入 不代表一次写几个，而是表示我的预期
encoding: "utf8",
});
const ws = new WriteStream(path.resolve(__dirname, "copy1.txt"), {
flags: "w", // 默认不写就是 r 表示我要读取
autoClose: true,
emitClose: true,
start: 0,
highWaterMark: 1, // 写入 不代表一次写几个，而是表示我的预期
encoding: "utf8",
});
rs.pipe(ws);
// 此方法是异步的可以是实现文件的拷贝，缺点就是无法看到中间过程
const EventEmitter = require("events");
const fs = require("fs");
class ReadStream extends EventEmitter {
constructor(path, options = {}) {
super();
this.path = path;
this.flags = options.flags || "r";
this.autoClose = options.autoClose || true;
this.emitClose = options.emitClose || true;
this.start = options.start || 0;
this.offset = this.start;
this.highWaterMark = options.highWaterMark || 64 * 1024;
this.flowing = false; // 默认为非流动模式， 用户没有读取数据
this.open(); // 这个打开操作是异步的
this.on("newListener", function (eventName) {
if (eventName === "data") {
// 用户绑定了data事件
this.flowing = true;
this.read(); // 需要用到打开的fd
}
});
}
destroy(err) {
if (typeof this.fd == "number") {
if (this.autoClose) {
fs.close(this.fd, () => {
if (this.emitClose) {
this.emit("close");
}
});
}
}
if (err) {
this.emit("error", err);
}
}
pause() {
this.flowing = false;
}
resume() {
if (!this.flowing) {
this.flowing = true;
this.read();
}
}
read() {
if (typeof this.fd !== "number") {
return this.once("open", this.read);
}
this.buffer = Buffer.alloc(this.highWaterMark);
fs.read(
this.fd,
this.buffer,
0,
this.highWaterMark,
this.offset,
(err, bytesRead) => {
if (err) {
return this.destroy(err);
}
if (bytesRead) {
this.offset += bytesRead;
this.emit("data", this.buffer.slice(0, bytesRead));
if (this.flowing) {
this.read(); // 递归读取
}
} else {
this.emit("end");
this.destroy();
}
}
);
}
open(ws) {
fs.open(this.path, this.flags, (err, fd) => {
if (err) {
return this.destroy(err);
}
this.fd = fd; // 打开成功后就有了fd
this.emit("open", fd);
});
}
pipe(ws) {
this.on("data", (chunk) => {
let flag = ws.write(chunk); // 写入的返回值来控制读取的速率， 实现了 读写间的分离
if (!flag) {
this.pause();
}
});
ws.on("drain", () => {
this.resume();
});
this.on("end", function () {
ws.end();
});
}
}
module.exports = ReadStream;
const EventEmitter = require("events");
const fs = require("fs");
class WriteStream extends EventEmitter {
constructor(path, options = {}) {
super();
this.path = path;
this.flags = options.flags || "w";
this.autoClose = options.autoClose || true;
this.emitClose = options.emitClose || true;
this.start = options.start || 0;
this.highWaterMark = options.highWaterMark || 16 * 1024;
this.encoding = options.encoding || "utf8";
this.offset = this.start; // offset 是写入的偏移量，这个参数可变
this.writing = false; // 是否正在写入，如果是正在写入，后续的写入操作要当到队列中
this.cache = []; // 缓存取， 队列
this.needDrain = false; // 只有写入的个数 达到了highWaterMark 并且清空后才会将这个值变成true
this.len = 0; // 每次写入的时候统计写入的个数
this.open();
}
open() {
fs.open(this.path, this.flags, (err, fd) => {
this.fd = fd;
this.emit("open", this.fd); // 发射open事件
});
}
// 写入的数据只能是是字符串或者buffer
clear() {
let buffer = this.cache.shift(); // 拿出缓存区中的第一个来
// 这个浪费性能，内部用的是链表
if (buffer) {
let { chunk, encoding, clearBuffer } = buffer;
this._write(chunk, encoding, clearBuffer);
} else {
this.writing = false;
if (this.needDrain) {
this.emit("drain");
}
}
}
end(chunk = "", encoding = this.encoding, callback = () => {}) {
this.write(chunk, encoding, () => {
callback();
fs.close(this.fd, () => {
if (this.emitClose) {
this.emit("close");
}
});
});
}
write(chunk, encoding = this.encoding, callback = () => {}) {
chunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
this.len += chunk.length; // 每次写入的个数都要记录，看能不能拿到highWaterMark
this.needDrain = this.len >= this.highWaterMark; // 说明写入的个数达到了触发drain条件
const clearBuffer = () => {
callback(); // 当前本次写入完成后回调,在调用后续的写入操作
this.clear();
};
if (this.writing) {
this.cache.push({
chunk,
encoding,
clearBuffer,
});
} else {
// 第一次做的写入操作， 这个时候需要像文件中写入
this.writing = true;
this._write(chunk, encoding, clearBuffer);
}
return !this.needDrain;
}
_write(chunk, encoding, callback) {
if (typeof this.fd !== "number") {
return this.once("open", () => this._write(chunk, encoding, callback));
}
fs.write(this.fd, chunk, 0, chunk.length, this.offset, (err, written) => {
this.offset += written;
this.len -= written;
callback();
});
}
}
module.exports = WriteStream;
浙公网安备 33010602011771号