fs.readStreamからのカスタムプロセスストリームチャンクをインターセプトすることは可能ですか？

Question

fs.readStreamからデータの中間部分を処理することは可能ですか？例：ファイルを読み込み、ファイル全体をメモリに保存せずに文字が何回出現するかを数えます。単純にストリームから各チャンクを解析し、数値を集めます。.pipe（）に渡す関数を作成しようとしましたが、fnには.on()が定義され、失敗している必要がありました。

私はストリーミングで初心者です。私は、グーグル、読書、実験の束をやったが、何も助けていない。私はまた、配管がどのように機能するか、独自のカスタムパイプ受信機能を作成できるかどうかについての情報を見つけることはできません。あなたができる

おかげ

Answer 1

easiesの事は、単に「data」イベントにサブスクライブし、単にこのような発生をカウントすることです：

'use strict'; 
 

 
const fs = require('fs'); 
 

 
const countCharsInString = function (st, char) { 
 
\t //Use a regex to calculate this 
 
\t let regex = new RegExp(char, 'g'); 
 
\t return (st.match(regex) || []).length 
 
}; 
 

 
const CHAR_TO_COUNT = '1'; 
 
let total_count = 0; 
 

 
let fileStream = fs.createReadStream('./log.txt'); 
 

 
//We'll calculate the occurrences as the stream emits data event 
 
//As we don't keep any references to 'chunk' and we don't collect the data anywhere 
 
//The garbage collector will clean the memory and we'll not run out of the RAM. 
 
fileStream.on('data', chunk => { 
 
\t let string = chunk.toString(); 
 
\t total_count += countCharsInString(string, CHAR_TO_COUNT); 
 
}); 
 
fileStream.on('error', err => console.error(err.stack)); 
 
//Log the count when everything is done. 
 
fileStream.on('end',() => console.log(`All done. There are ${total_count} occurrences of character '${CHAR_TO_COUNT}'.`));

しかし、あなたは何したい場合ストリーム経由でデータを処理し、そのデータをその場で集計して、別の場所にパイプしてください。

'use strict'; 
 

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

 
//We inherit from the Transform stream class 
 
class OcurrenceCountingStream extends Transform { 
 
\t constructor(options) { 
 
\t \t super(options); 
 

 
\t \t //Allowing to pass an option here 
 
\t \t this.charToCount = options.charToCount; 
 
\t \t this.totalCount = 0; 
 
\t } 
 

 
\t //This is now a static method as it's a pure function 
 
\t //That does not depend on the object state 
 
\t static countCharsInString(st, char) { 
 
\t \t //Use a regex to calculate this 
 
\t \t let regex = new RegExp(char, 'g'); 
 
\t \t return (st.match(regex) || []).length 
 
\t } 
 

 
\t //We should implement _transform function 
 
\t //in order to make all piping mechanisms working 
 
\t _transform(chunk, encoding, callback) { 
 
\t \t //Get our string, process and update totalCount 
 
\t \t let string = chunk.toString(); 
 
\t \t this.totalCount += OcurrenceCountingStream.countCharsInString(string, this.charToCount); 
 

 
\t \t //Pass the data further 
 
\t \t callback(null, chunk); 
 
\t } 
 
} 
 

 
let fileStream = fs.createReadStream('./log.txt'); 
 
let countingStream = new OcurrenceCountingStream({charToCount: '1'}); 
 

 
//So now we can pipe 
 
fileStream.pipe(countingStream); 
 
/* 
 
Here is a little moment. 
 
The stream should be in a flowing mode. This means that is started reading the data 
 
From the writable that was piped to it and will keep reading until the writer is ended 
 
So nothing basically happens if we just pipe it like this fileStream.pipe(countingStream); 
 
There are some ways to make countingStream flowing: 
 
1) Subscribe on its 'data' event. 
 
2) Pipe it somewhere 
 
3) Call .resume() is we don't really care of the data that's comming out. 
 
*/ 
 
countingStream.resume(); 
 

 
countingStream.on('error', err => console.error(err.stack)); 
 
countingStream.on('end',() => console.log(`All done. There are ${countingStream.totalCount} occurrences of character '${countingStream.charToCount}'.`));