ノードストリーム - 読み取り可能なストリームでのunpipeのリッスン

Question

これで、SFTPに最初に接続し、ファイルからの読み取りを開始する読み取りストリームを作成しました。いつでも、私のコードはそのストリームをアンペアして何か他のことをすることができます。たとえば、これを使用してCSVの最初の数行を取得し、読み上げを停止することがあります。

問題は、正確にSFTP接続を閉じることができるように、私のreadStreamコンストラクタのunpipeイベントをリスンする方法がわかりません。私は書き込みストリームでflushメソッドを使用しますが、ストリームのようなものがありますか？

ここに私のreadStreamコンストラクタの単純化の部分です：

const Client = require('ssh2').Client, 
     nom = require('noms'); 

function getStream (get) { 
    const self = this; 
    const conn = new Client(); 

    let client, 
     fileData, 
     buffer, 
     totalBytes = 0, 
     bytesRead = 0; 

    let read = function(size,next) { 
     const read = this; 
     // Read each chunk of the file 
     client.read(fileData, buffer, bytesRead, size, bytesRead, 
      function (err, byteCount, buff, pos) { 
       bytesRead += byteCount; 
       read.push(buff); 
       next(); 
      } 
     ); 
    }; 

    let before = function(start) { 
     // setup the connection BEFORE we start _read 
     conn.on('ready', function(){ 
      conn.sftp(function(err,sftp) { 
       sftp.open(get, 'r', function(err, fd){ 
        sftp.fstat(fd, function(err, stats) { 
         client = sftp; 
         fileData = fd; 
         totalBytes = stats.size; 
         buffer = new Buffer(totalBytes); 

         start(); 
        }); 
       }); 
      }); 
     }).connect(credentials); 
    }; 

    return nom(read,before); 
} 

その後、私はmyStream.pipe(writeStream)を呼び出し、その後myStream.unpipe()かもしれません。しかし、私はそのunpipeイベントをリッスンする方法がないため、読み取りは中止されますが、SFTP接続は開いたままになり、最終的にタイムアウトします。

アイデア？

Answer 1

さらに研究を重ねた結果、readStream.unpipe(writeStream)に電話すると、ReadStreamsがunpipeイベントに渡されないことがわかりました。このイベントはwriteStreamだけに渡されます。

readStream.emit('unpipe'); 

あなたはどこでも、内部のか、本当に便利です、あなたのストリームコンストラクタ、外でこのイベントをリッスンすることができますunpipeをリッスンするためには、そのように、明示的readStreamにイベントを放出する必要があります。あなたが発すると箱の外のカスタムイベントをリッスンすることができますので、Event Emitter class methodsを持って、すでに、物語の

function getStream (get) { 
    /** 
    * ... stuff 
    * ... read() 
    * ... before() 
    * ... etc 
    */ 

    let readStream = nom(read,before); 

    readStream.on('unpipe', function(){ 
     console.log('called unpipe on read stream'); 
    }); 

    return readStream; 
} 

道徳ストリーム：だから、それは次のようになり、上記のコードになるだろう。