1 //自定義文件讀取流readFile.js
  2 const fs = require('fs');
  3 const EventEmitter = require('events').EventEmitter;
  4 const {Queue} = require('./linked.js');
  5 
  6 class MyFileReadStream extends EventEmitter {
  7     constructor(path, optons={}){
  8         super();
  9         this.path = path;                   //綁定要讀取的文件
 10         this.flags = optons.flags || 'r';   //文件件操作模式：讀取模式
 11         this.mode = optons.mode || 438;     //文件操作權限
 12         this.autoClose = optons.autoClose || true;  //是否關閉（銷毀）當前文件
 13         this.start = optons.start || 0;     //開始讀取的位置
 14         this.end = optons.end;              //結束讀取的位置
 15         this.highWaterMark = optons.highWaterMark || 64 * 1024; //可讀流最大可緩存的位元組數
 16         
 17         this.readOffset = 0;        //從什麼位置讀出
 18         this.cache = new Queue();   //用於緩存數據的隊列
 19         this.readableFlowing = null;    //當前流的狀態：null(不提供數據狀態)、true(持續輸出數據)、false(暫停輸出數據)
 20 
 21         this.open();
 22         this.on('newListener',(type)=>{ //當在MyFileReadStream實例上添加監聽事件時觸發該事件，並將當前監聽事件的名稱傳遞給回調
 23             if(type === 'data'){
 24                 this.read();
 25             }
 26         });
 27     }
 28     open(){
 29         //原生open方法打開指定位置上的文件
 30         fs.open(this.path, this.flags, this.mode, (err, fd)=>{
 31             if(err){
 32                 this.emit('error', err);
 33             }
 34             this.fd = fd;
 35             this.readableFlowing = true;    //當文件正常打開時將流的狀態置為持續數據數據的狀態
 36             this.emit('open',fd);
 37         });
 38     }
 39     read(){
 40         //負責將要讀取的數據通過data事件輸出
 41         if(typeof this.fd !== 'number'){   
 42              //當第一次觸發事件
 43             return this.once('open',this.read);     //在通過open事件獲取到文件標識符後，重新調用數據讀取操作
 44         }
 45         if(!this.readableFlowing){  //當流處於暫停模式時，需要通過resume()重啟流
 46             return ;
 47         }
 48         if(this.cache.size > 0){    //如果可讀流的中間緩存中有數據，就從中間緩存中拿數據通過data事件響應出去
 49             this.emit("data",this.cache.deQueue());
 50             return this.read();
 51         }
 52         //實現數據讀取操作：        
 53         let buf = Buffer.alloc(this.highWaterMark);
 54         let howMuchToRead ;
 55         if(this.end){
 56             howMuchToRead = Math.min(this.end - this.readOffset + 1, this.highWaterMark);
 57         }else{
 58             howMuchToRead = this.highWaterMark;
 59         }
 60         //調用原生文件讀取方法fs.read獲取數據
 61         fs.read(this.fd, buf, 0, howMuchToRead, this.readOffset,(err,readBytes)=>{
 62             if(readBytes){
 63                 this.readOffset += readBytes;
 64                 //這裡需要判斷是否通過data事件輸出數據，還是將數據緩存到中間緩存中
 65                 if(this.readableFlowing){
 66                     //當流處於數據持續輸出狀態，直接將數據通過data事件輸出，並繼續調用read()讀取數據
 67                     this.emit('data',buf.slice(0,readBytes));
 68                     this.read();
 69                 }else{
 70                     //當流處於數據暫停輸出狀態，將數據緩存到中間緩存(即緩存隊列)
 71                    this.cache.enQueue(buf.slice(0,readBytes));
 72                 }
 73             }else{
 74                 //當沒有數據可讀時觸發end事件，並關閉文件標識符
 75                 this.emit('end');
 76                 this.close();
 77             }
 78         });
 79     }
 80     close(){
 81         //關閉文件流操作
 82         fs.close(this.fd,()=>{
 83             this.autoClose= false; //表示文件已被關閉
 84             this.emit('close');
 85         });
 86     }
 87     pause(){
 88         //暫停流操作
 89         this.readableFlowing = false;
 90     }
 91     resume(){
 92         //重啟流操作
 93         this.readableFlowing = true;
 94         this.read();
 95     }
 96     pipe(ws){
 97         this.on('data',(data)=>{
 98             let flag = ws.write(data);
 99             if(!flag){
100                 this.pause();
101             }
102         });
103         this.on("end",()=>{
104             ws.end();
105         });
106         this.on("close",()=>{
107             console.log("文件讀取完成，正常關閉");
108         });
109         ws.on('drain',()=>{
110             this.resume();
111         });
112     }
113 }
114 
115 module.exports = MyFileReadStream;

  1 //自定文件寫入流writeFile.js
  2 const fs = require('fs');
  3 const {EventEmitter} = require('events');
  4 const {Queue} = require('./linked.js');
  5 
  6 class MyFileWriteStream extends EventEmitter{
  7     constructor(path,options={}){
  8         super();
  9         this.path = path;
 10         this.flags = options.flags || 'w';
 11         this.mode = options.mode || 438;
 12         this.autoClose = options.autoClose || true;
 13         this.start = options.start || 0;
 14         this.encoding = options.encoding || 'utf8';
 15         this.highWaterMark = options.highWaterMark || 16*1024;
 16 
 17         this.writeoffset = this.start;  //從什麼位置寫入
 18         this.writing = false;   //是否正在寫入
 19         this.writLen = 0;       //當前MyWriteStream實例要寫入的位元組長度，也就在當的寫入操作中間緩存上有多少個位元組的數據
 20         this.needDrain = false; //中間緩存是否排空了，如果排空了就觸發drain事件
 21         this.cache = new Queue();
 22         this.upstream = false;  //上游狀態是否關閉。比如由上可讀流通過pipe傳輸到當前可寫流上的數據已經到達末尾，可讀流讀取的文件要關閉被打開的文件時，通過可寫流的end()通知可寫流它已經關閉
 23 
 24         this.open();
 25     };
 26     open(){
 27         //原生fs.open
 28         fs.open(this.path, this.flags, (err, fd)=>{
 29             if(err){
 30                 this.emit('error', err);
 31             }
 32             //正常打開文件
 33             this.fd = fd;
 34             this.emit('open', fd);
 35         });
 36     };
 37     //寫入操作的對外接口
 38     write(chunk, encoding, cb){
 39         if(chunk !== null){
 40             chunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
 41             this.writLen += chunk.length;   //累計當前要寫入的位元組數（即表示寫入流緩存中的數據位元組長度）
 42             let flag = this.writLen < this.highWaterMark;   //檢查緩存中的數據節點長度是否超出流的最大緩存空間。flag為tuer時表示緩存沒有超出，反之則超出了
 43             this.needDrain = !flag;
 44             if(this.writing){
 45                 //當前正在執行寫入，即將內容緩存到隊列
 46                 this.cache.enQueue({
 47                     chunk:chunk,
 48                     encoding:encoding,
 49                     cb:cb
 50                 });
 51             }else{
 52                 //當前不是正在寫入，即立可以即執行寫入操作
 53                 this._write(chunk,encoding,()=>{
 54                     cb && cb();   //執行寫入回調
 55                     //清空排隊的內容
 56                     this._clearBuffer();
 57                 });
 58             }
 59             return flag;
 60         }else{
 61             cb && cb();
 62             //清空排隊的內容
 63             this._clearBuffer();
 64         }
 65     };
 66     //實現寫入
 67     _write(chunk, encoding, cb){
 68         this.writing = true;    //將寫入狀態置為正在寫入
 69         if(typeof this.fd !== 'number'){
 70             //第一次寫入時，writeFile 對象上沒有fd，這事因為_write是同步任務，而this.fd獲取文件描述符的邏輯在open()執行後的異步回調上
 71             //在open()的異步回調任務中綁定this.fd以後會觸發'open'事件，也就是說這個當'open'事件觸發時就會fd了
 72             //所以在第一次寫入時在'open'事件上添加一個一次性的任務，在這個任務內真正的實現寫入操作。
 73             return this.once('open',()=>{
 74                 this._write(chunk, encoding, cb);
 75             });
 76         }
 77         fs.write(this.fd, chunk, this.start, chunk.length, this.writeoffset,(err,writen)=>{
 78             this.writeoffset += writen; //將寫入的位元組數累計到寫入位置上，為下一次寫入提供寫入位置定位
 79             this.writLen -= writen; //將緩存位元組數記錄減掉寫入的位元組數
 80             this.writing = false;
 81             cb && cb();
 82         });
 83     };
 84     _clearBuffer(){
 85         let data = this.cache.deQueue();
 86         if(data){
 87             //當有數據時持續迭代鏈表節點中的緩存，從而實現將緩存中的數據寫入到磁盤中
 88             this._write(data.element.chunk, data.element.encoding, ()=>{
 89                 data.element.cb && data.element.cb();
 90                 this._clearBuffer();
 91             });
 92         }else{
 93             //當緩存中沒有數據
 94             if(this.upstream){  //當上游調用可寫流的程序已經關閉，說明不會再有數據傳入，這時候應該關閉當前的可寫流
 95                 return this.close();
 96             }
 97             if(this.needDrain){ 
 98                 this.needDrain = false;
 99                 this.emit('drain');
100             }
101         }
102     }
103     end(){
104         this.write(null);
105         this.upstream = true;
106     }
107     close(){
108         fs.close(this.fd,()=>{
109             this.emit("close");
110             console.log("文件寫入完成，正常關閉")
111         });
112     }
113 }
114 module.exports = MyFileWriteStream;

  1 //鏈表結構
  2 //node節點 + head + null
  3 //head頭 -> null
  4 //size鏈表長度
  5 //next下一個節點 element
  6  //增加、刪除、修改、查詢、清空
  7 
  8  //構造節點
  9  class Node{
 10      constructor(element, next){
 11          this.element = element;
 12          this.next = next;
 13      }
 14  }
 15 //構造鏈表
 16 class LinkedList{
 17     constructor(head, size){
 18         this.head = null;
 19         this.size = 0;
 20     }
 21     //獲取指定節點
 22     _getNode(index){
 23         if(index < 0 || index >= this.size){
 24             throw new Error('cross the border');    //拋出越界錯誤
 25         }
 26         let currentNode = this.head;
 27         for(let i = 0; i < index; i++){
 28             currentNode = currentNode.next;
 29         }
 30         return currentNode;
 31     }
 32     //添加節點：可以在指定的位置添加（即插入節點：傳入索引+節點元素兩個參數），如果只傳入節點元素就默認在鏈表的末尾添加節點
 33     add(index, element){
 34        if(arguments.length === 1){
 35            element = index;
 36            index = this.size;   //當沒有傳入插入位置時，將插入位置默認未鏈表的末尾
 37        }
 38        if(index < 0 || index > this.size){
 39             throw new Error('cross the border');    //拋出越界錯誤
 40        }
 41        if(index === 0){
 42            let head = this.head;
 43            this.head = new Node(element,head);
 44        }else{
 45             let prevNode = this._getNode(index -1);
 46             prevNode.next = new Node(element,prevNode.next);
 47        }
 48        this.size++;
 49     }
 50     //刪除節點
 51     remove(index){
 52         let rmNode = null;  //刪除的節點
 53         if(index === 0){
 54             rmNode = this.head;
 55             if(!rmNode){
 56                 return undefined;
 57             }
 58             this.head = rmNode.next;
 59         }else{
 60             let prevNode = this._getNode(index-1);
 61             rmNode = prevNode.next; 
 62             prevNode.next = prevNode.next.next;
 63         }
 64         this.size --;
 65         return rmNode;
 66     }
 67     //修改鏈表節點
 68     set(index,element){
 69         let node = this._getNode(index);
 70         node.element = element;
 71     }
 72     //查詢鏈表節點
 73     get(index){
 74         return this._getNode(index);
 75     }
 76     //清空鏈表
 77     clear(){
 78         this.head = null;
 79         this.size = 0;
 80     }
 81 }
 82 
 83 //構造鏈表隊列
 84 class Queue{
 85     constructor(){
 86         this.linkedList = new LinkedList();
 87     }
 88     //入隊列
 89     enQueue(data){
 90         this.linkedList.add(data);
 91     }
 92     //出隊列
 93     deQueue(){
 94         return this.linkedList.remove(0);
 95     }
 96 }
 97 
 98 module.exports = {
 99     Node:Node,
100     LinkedList:LinkedList,
101     Queue:Queue
102 };

 1 const fs = require('fs');
 2 const myFileReadStream = require('./readFile.js');
 3 const myWriteStream = require('./writeFile.js');
 4 
 5 // const rs = fs.createReadStream('./筆記(副本).txt',{
 6 //     highWaterMark:4
 7 // });
 8 // const rs = fs.createReadStream('./tst.txt',{
 9 //     highWaterMark:1
10 // });
11 const rs = new myFileReadStream('./筆記(副本).txt',{
12     highWaterMark:4
13 });
14 
15 // const ws = fs.createWriteStream('./筆記.txt',{
16 //     highWaterMark:12
17 // });
18 const ws = new myWriteStream('./筆記.txt',{
19     highWaterMark:12
20 });
21 rs.pipe(ws);