建站
咨詢
售后
建站咨詢
新聞中心
穩(wěn)定性: 2 - 不穩(wěn)定流用于處理node.js中的流數(shù)據(jù)的抽象接口，在Node里被不同的對(duì)象實(shí)現(xiàn)。例如，對(duì)HTTP服務(wù)器的請(qǐng)求是流，process.stdout 是流。
創(chuàng)新互聯(lián)長(zhǎng)期為上千余家客戶提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊(duì)從業(yè)經(jīng)驗(yàn)10年，關(guān)注不同地域、不同群體，并針對(duì)不同對(duì)象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺(tái)，與合作伙伴共同營(yíng)造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為菏澤企業(yè)提供專業(yè)的網(wǎng)站建設(shè)、成都網(wǎng)站建設(shè)，菏澤網(wǎng)站改版等技術(shù)服務(wù)。擁有十載豐富建站經(jīng)驗(yàn)和眾多成功案例,為您定制開發(fā)。
流是可讀的，可寫的，或者是可讀寫的，所有的流是EventEmitter的實(shí)例。
Node.js訪問流模塊的方法如下所示：
const stream = require('stream');你可以通過require('stream')加載Stream基類。其中包括了Readable流、Writable流、Duplex流和Transform流的基類。
本文將分為3個(gè)部分進(jìn)行介紹。
第一個(gè)部分解釋了在你的程序中使用流時(shí)候需要了解的內(nèi)容。如果你不用實(shí)現(xiàn)流式API，可以只看這個(gè)部分。
如果你想實(shí)現(xiàn)你自己的流，第二個(gè)部分解釋了這部分API。這些API讓你的實(shí)現(xiàn)更加簡(jiǎn)單。
第三個(gè)部分深入的解釋了流是如何工作的，包括一些內(nèi)部機(jī)制和函數(shù)，這些內(nèi)容不要改動(dòng)，除非你明確知道你要做什么。
面向流消費(fèi)者的API
流可以是可讀（Readable），可寫（Writable），或者是可讀可寫的（Duplex，雙工）。
所有的流都是事件分發(fā)器（EventEmitters），但是也有自己的方法和屬性，這取決于他它們是可讀（Readable），可寫（Writable），或者兼具兩者（Duplex，雙工）的。
如果流是可讀寫的，則它實(shí)現(xiàn)了下面的所有方法和事件。因此，這個(gè)部分API完全闡述了Duplex或Transform流，即便他們的實(shí)現(xiàn)有所不同。
沒有必要為了消費(fèi)流而在你的程序里實(shí)現(xiàn)流的接口。如果你正在你的程序里實(shí)現(xiàn)流接口，請(qǐng)同時(shí)參考下面的流實(shí)現(xiàn)程序API。
基本所有的Node程序，無論多簡(jiǎn)單，都會(huì)使用到流。以下是一個(gè)使用流的例子：
javascript
var http = require('http');
var server = http.createServer(function (req, res) {
// req is an http.IncomingMessage, which is 可讀流（Readable stream）
// res is an http.ServerResponse, which is a Writable Stream
var body = '';
// we want to get the data as utf8 strings
// If you don't set an encoding, then you'll get Buffer objects
req.setEncoding('utf8');
// 可讀流（Readable stream） emit 'data' 事件 once a 監(jiān)聽器（listener） is added
req.on('data', function (chunk) {
body += chunk;
});
// the end 事件 tells you that you have entire body
req.on('end', function () {
try {
var data = JSON.parse(body);
} catch (er) {
// uh oh! bad json!
res.statusCode = 400;
return res.end('error: ' + er.message);
}
// write back something interesting to the user:
res.write(typeof data);
res.end();
});
});
server.listen(1337);
// $ curl localhost:1337 -d '{}'
// object
// $ curl localhost:1337 -d '"foo"'
// string
// $ curl localhost:1337 -d 'not json'
// error: Unexpected token o類： stream.Readable
可讀流（Readable stream）接口是對(duì)你正在讀取的數(shù)據(jù)的來源的抽象。換句話說，數(shù)據(jù)來來自
可讀流（Readable stream）不會(huì)分發(fā)數(shù)據(jù)，直到你表明準(zhǔn)備就緒。
可讀流（Readable stream） 有2種模式:流動(dòng)模式（flowing mode）和暫停模式（paused mode）。流動(dòng)模式（flowing mode）時(shí)，盡快的從底層系統(tǒng)讀取數(shù)據(jù)并提供給你的程序。暫停模式（paused mode）時(shí)，你必須明確的調(diào)用stream.read()來讀取數(shù)據(jù)。暫停模式（paused mode）是默認(rèn)模式。
注意: 如果沒有綁定數(shù)據(jù)處理函數(shù)，并且沒有pipe()目標(biāo)，流會(huì)切換到流動(dòng)模式（flowing mode），并且數(shù)據(jù)會(huì)丟失。
可以通過下面幾個(gè)方法，將流切換到流動(dòng)模式（flowing mode）。
- 添加一個(gè)[
'data'事件][]事件處理器來監(jiān)聽數(shù)據(jù). - 調(diào)用
resume()方法來明確的開啟數(shù)據(jù)流。 - 調(diào)用
pipe()方法來發(fā)送數(shù)據(jù)給Writable.
可以通過以下方法來切換到暫停模式（paused mode）:
- 如果沒有“導(dǎo)流（pipe）”目標(biāo)，調(diào)用
pause()方法. - 如果有“導(dǎo)流（pipe）”目標(biāo)，移除所有的[
'data'事件][]處理函數(shù)，調(diào)用unpipe()方法移除所有的“導(dǎo)流（pipe）”目標(biāo)。
注意：為了向后兼容考慮， 移除'data'事件監(jiān)聽器并不會(huì)自動(dòng)暫停流。同樣的，當(dāng)有導(dǎo)流目標(biāo)時(shí)，調(diào)用pause()并不能保證流在那些目標(biāo)排空后，請(qǐng)求更多數(shù)據(jù)時(shí)保持暫停狀態(tài)。
可讀流（Readable stream）例子包括:
- http responses, on the client
- http requests, on the server
- fs read streams
- zlib streams
- crypto streams
- tcp sockets
- child process stdout and stderr
- process.stdin
事件: 'readable'
當(dāng)一個(gè)數(shù)據(jù)塊可以從流中讀出，將會(huì)觸發(fā)'readable'事件.`
某些情況下, 如果沒有準(zhǔn)備好，監(jiān)聽一個(gè)'readable'事件將會(huì)導(dǎo)致一些數(shù)據(jù)從底層系統(tǒng)讀取到內(nèi)部緩存。
javascript
var readble = getReadableStreamSomehow();
readable.on('readable', function() {
// there is some data to read now
});一旦內(nèi)部緩存排空，一旦有更多數(shù)據(jù)將會(huì)再次觸發(fā)readable事件。
事件: 'data'
chunk{Buffer | String} 數(shù)據(jù)塊
綁定一個(gè)data事件的監(jiān)聽器（listener）到一個(gè)未明確暫停的流，會(huì)將流切換到流動(dòng)模式。數(shù)據(jù)會(huì)盡額能的傳遞。
如果你像盡快的從流中獲取數(shù)據(jù)，以下的方法是最快的：
javascript
var readable = getReadableStreamSomehow();
readable.on('data', function(chunk) {
console.log('got %d bytes of data', chunk.length);
});事件: 'end'
如果沒有更多的可讀數(shù)據(jù)，將會(huì)觸發(fā)這個(gè)事件。
注意：只有數(shù)據(jù)已經(jīng)被完全消費(fèi)，end事件才會(huì)觸發(fā)?？梢酝ㄟ^切換到流動(dòng)模式（flowing mode）來實(shí)現(xiàn)，或者通過調(diào)用重復(fù)調(diào)用read()獲取數(shù)據(jù)，直到結(jié)束。
javascript
var readable = getReadableStreamSomehow();
readable.on('data', function(chunk) {
console.log('got %d bytes of data', chunk.length);
});
readable.on('end', function() {
console.log('there will be no more data.');
}); 事件: 'close'
當(dāng)?shù)讓淤Y源（例如源頭的文件描述符）關(guān)閉時(shí)觸發(fā)。并不是所有流都會(huì)觸發(fā)這個(gè)事件。
事件: 'error'
- {Error Object}
當(dāng)接收數(shù)據(jù)時(shí)發(fā)生錯(cuò)誤觸發(fā)。
readable.read([size])
size{Number} 可選參數(shù)， 需要讀入的數(shù)據(jù)量- 返回 {String | Buffer | null}
read()方法從內(nèi)部緩存中拉取數(shù)據(jù)。如果沒有可用數(shù)據(jù)，將會(huì)返回null
如果傳了size參數(shù)，將會(huì)返回相當(dāng)字節(jié)的數(shù)據(jù)。如果size不可用，將會(huì)返回null。
如果你沒有指定size參數(shù)。將會(huì)返回內(nèi)部緩存的所有數(shù)據(jù)。
這個(gè)方法僅能再暫停模式（paused mode）里調(diào)用。 流動(dòng)模式（flowing mode）下這個(gè)方法會(huì)被自動(dòng)調(diào)用直到內(nèi)存緩存排空。
javascript
var readable = getReadableStreamSomehow();
readable.on('readable', function() {
var chunk;
while (null !== (chunk = readable.read())) {
console.log('got %d bytes of data', chunk.length);
}
});如果這個(gè)方法返回一個(gè)數(shù)據(jù)塊, 它同時(shí)也會(huì)觸發(fā)['data'事件][].
readable.setEncoding(encoding)
encoding{String} 要使用的編碼.- 返回:
this
調(diào)用此函數(shù)會(huì)使得流返回指定編碼的字符串，而不是Buffer對(duì)象。例如，如果你調(diào)用readable.setEncoding('utf8')，輸出數(shù)據(jù)將會(huì)是UTF-8編碼，并且返回字符串。如果你調(diào)用readable.setEncoding('hex')，將會(huì)返回2進(jìn)制編碼的數(shù)據(jù)。
該方法能正確處理多字節(jié)字符。如果不想這么做，僅簡(jiǎn)單的直接拉取緩存并調(diào)buf.toString(encoding)，可能會(huì)導(dǎo)致字節(jié)錯(cuò)位。因此，如果你想以字符串讀取數(shù)據(jù)，請(qǐng)使用下述的方法：
javascript
var readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', function(chunk) {
assert.equal(typeof chunk, 'string');
console.log('got %d characters of string data', chunk.length);
});readable.resume()
- 返回:
this
這個(gè)方法讓可讀流（Readable stream）繼續(xù)觸發(fā)data事件.
這個(gè)方法會(huì)將流切換到流動(dòng)模式（flowing mode）。 如果你不想從流中消費(fèi)數(shù)據(jù)，而想得到end事件，可以調(diào)用readable.resume()來打開數(shù)據(jù)流，如下所示：
javascript
var readable = getReadableStreamSomehow();
readable.resume();
readable.on('end', function(chunk) {
console.log('got to the end, but did not read anything');
});readable.pause()
- 返回:
this
這個(gè)方法會(huì)使得流動(dòng)模式（flowing mode）的流停止觸發(fā)data事件，切換到流動(dòng)模式（flowing mode）。并讓后續(xù)可用數(shù)據(jù)留在內(nèi)部緩沖區(qū)中。
javascript
var readable = getReadableStreamSomehow();
readable.on('data', function(chunk) {
console.log('got %d bytes of data', chunk.length);
readable.pause();
console.log('there will be no more data for 1 second');
setTimeout(function() {
console.log('now data will start flowing again');
readable.resume();
}, 1000);
});readable.isPaused()
- 返回:
Boolean
這個(gè)方法返回readable是否被客戶端代碼明確的暫停（調(diào)用readable.pause()）。
var readable = new stream.Readable
readable.isPaused() // === false
readable.pause()
readable.isPaused() // === true
readable.resume()
readable.isPaused() // === falsereadable.pipe(destination[, options])
destination{Writable Stream} 寫入數(shù)據(jù)的目標(biāo)options{Object} 導(dǎo)流（pipe）選項(xiàng)end{Boolean} 讀取到結(jié)束符時(shí)，結(jié)束寫入者。默認(rèn) =true
這個(gè)方法從可讀流（Readable stream）拉取所有數(shù)據(jù)，并將數(shù)據(jù)寫入到提供的目標(biāo)中。自動(dòng)管理流量，這樣目標(biāo)不會(huì)快速的可讀流（Readable stream）淹沒。
可以導(dǎo)流到多個(gè)目標(biāo)。
javascript
var readable = getReadableStreamSomehow();
var writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt'
readable.pipe(writable);這個(gè)函數(shù)返回目標(biāo)流, 因此你可以建立導(dǎo)流鏈：
javascript
var r = fs.createReadStream('file.txt');
var z = zlib.createGzip();
var w = fs.createWriteStream('file.txt.gz');
r.pipe(z).pipe(w);例如：模擬Unix的cat命令：
javascript
process.stdin.pipe(process.stdout);默認(rèn)情況下，當(dāng)源數(shù)據(jù)流觸發(fā)end的時(shí)候調(diào)用end()，所以destination不可再寫。傳{ end:false }作為options，可以保持目標(biāo)流打開狀態(tài)。
這會(huì)讓writer保持打開狀態(tài)，可以在最后寫入"Goodbye"：
javascript
reader.pipe(writer, { end: false });
reader.on('end', function() {
writer.end('Goodbye\n');
});注意：process.stderr和process.stdout直到進(jìn)程結(jié)束才會(huì)關(guān)閉，無論是否指定它們。
readable.unpipe([destination])
destination{Writable Stream} 可選，指定解除導(dǎo)流的流
這個(gè)方法會(huì)解除之前調(diào)用pipe() 設(shè)置的鉤子（pipe()）。
如果沒有指定destination，則所有的導(dǎo)流（pipe）都會(huì)被移除。
如果指定了destination，但是沒有建立如果沒有指定destination，則什么事情都不會(huì)發(fā)生。
javascript
var readable = getReadableStreamSomehow();
var writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt',
// but only for the first second
readable.pipe(writable);
setTimeout(function() {
console.log('stop writing to file.txt');
readable.unpipe(writable);
console.log('manually close the file stream');
writable.end();
}, 1000);readable.unshift(chunk)
chunk{Buffer | String} 數(shù)據(jù)塊插入到讀隊(duì)列中
這個(gè)方法很有用，當(dāng)一個(gè)流正被一個(gè)解析器消費(fèi)，解析器可能需要將某些剛拉取出的數(shù)據(jù)“逆消費(fèi)”，返回到原來的源，以便流能將它傳遞給其它消費(fèi)者。
如果你在程序中必須經(jīng)常調(diào)用stream.unshift(chunk) ，那你可以考慮實(shí)現(xiàn)Transform來替換（參見下文API for Stream Implementors）。
javascript
// Pull off a header delimited by \n\n
// use unshift() if we get too much
// Call the callback with (error, header, stream)
var StringDecoder = require('string_decoder').StringDecoder;
function parseHeader(stream, callback) {
stream.on('error', callback);
stream.on('readable', onReadable);
var decoder = new StringDecoder('utf8');
var header = '';
function onReadable() {
var chunk;
while (null !== (chunk = stream.read())) {
var str = decoder.write(chunk);
if (str.match(/\n\n/)) {
// found the header boundary
var split = str.split(/\n\n/);
header += split.shift();
var remaining = split.join('\n\n');
var buf = new Buffer(remaining, 'utf8');
if (buf.length)
stream.unshift(buf);
stream.removeListener('error', callback);
stream.removeListener('readable', onReadable);
// now the body of the message can be read from the stream.
callback(null, header, stream);
} else {
// still reading the header.
header += str;
}
}
}
}readable.wrap(stream)
stream{Stream} 一個(gè)舊式的可讀流（Readable stream）
v0.10版本之前的Node流并未實(shí)現(xiàn)現(xiàn)在所有流的API（更多信息詳見下文“兼容性”部分）。
如果你使用的是舊的Node庫，它觸發(fā)'data'事件，并擁有僅做查詢用的pause()方法，那么你能使用wrap()方法來創(chuàng)建一個(gè)Readable流來使用舊版本的流，作為數(shù)據(jù)源。
你應(yīng)該很少需要用到這個(gè)函數(shù)，但它會(huì)留下方便和舊版本的Node程序和庫交互。
例如:
javascript
var OldReader = require('./old-api-module.js').OldReader;
var oreader = new OldReader;
var Readable = require('stream').Readable;
var myReader = new Readable().wrap(oreader);
myReader.on('readable', function() {
myReader.read(); // etc.
});類： stream.Writable
可寫流（Writable stream ）接口是你正把數(shù)據(jù)寫到一個(gè)目標(biāo)的抽象。
可寫流（Writable stream ）的例子包括:
- http requests, on the client
- http responses, on the server
- fs write streams
- zlib streams
- crypto streams
- tcp sockets
- child process stdin
- process.stdout, process.stderr
writable.write(chunk[, encoding][, callback])
chunk{String | Buffer} 準(zhǔn)備寫的數(shù)據(jù)encoding{String} 編碼方式（如果chunk是字符串）callback{Function} 數(shù)據(jù)塊寫入后的回調(diào)- 返回: {Boolean} 如果數(shù)據(jù)已被全部處理返回true
這個(gè)方法向底層系統(tǒng)寫入數(shù)據(jù)，并在數(shù)據(jù)處理完畢后調(diào)用所給的回調(diào)。
返回值表示你是否應(yīng)該繼續(xù)立即寫入。如果數(shù)據(jù)要緩存在內(nèi)部，將會(huì)返回false。否則返回true。
返回值僅供參考。即使返回false，你也可能繼續(xù)寫。但是寫會(huì)緩存在內(nèi)存里，所以不要做的太過分。最好的辦法是等待drain事件后，再寫入數(shù)據(jù)。
事件: 'drain'
如果調(diào)用writable.write(chunk)返回 false，drain事件會(huì)告訴你什么時(shí)候?qū)⒏嗟臄?shù)據(jù)寫入到流中。
javascript
// Write the data to the supplied 可寫流（Writable stream ） 1MM times.
// Be attentive to back-pressure.
function writeOneMillionTimes(writer, data, encoding, callback) {
var i = 1000000;
write();
function write() {
var ok = true;
do {
i -= 1;
if (i === 0) {
// last time!
writer.write(data, encoding, callback);
} else {
// see if we should continue, or wait
// don't pass the callback, because we're not done yet.
ok = writer.write(data, encoding);
}
} while (i > 0 && ok);
if (i > 0) {
// had to stop early!
// write some more once it drains
writer.once('drain', write);
}
}
}writable.cork()
強(qiáng)制緩存所有寫入。
調(diào)用.uncork()或.end()后，會(huì)把緩存數(shù)據(jù)寫入。
writable.uncork()
寫入所有.cork()調(diào)用之后緩存的數(shù)據(jù)。
writable.setDefaultEncoding(encoding)
encoding{String} 新的默認(rèn)編碼- 返回:
Boolean
給寫數(shù)據(jù)流設(shè)置默認(rèn)編碼方式，如編碼有效，則返回true ，否則返回false。
writable.end([chunk][, encoding][, callback])
chunk{String | Buffer} 可選，要寫入的數(shù)據(jù)encoding{String} 編碼方式（如果chunk是字符串）callback{Function} 可選， stream結(jié)束時(shí)的回調(diào)函數(shù)
當(dāng)沒有更多的數(shù)據(jù)寫入的時(shí)候調(diào)用這個(gè)方法。如果給出，回調(diào)會(huì)被用作finish事件的監(jiān)聽器。
調(diào)用end()后調(diào)用write()會(huì)產(chǎn)生錯(cuò)誤。
javascript
// write 'hello, ' and then end with 'world!'
var file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// writing more now is not allowed!事件: 'finish'
調(diào)用end()方法后，并且所有的數(shù)據(jù)已經(jīng)寫入到底層系統(tǒng)，將會(huì)觸發(fā)這個(gè)事件。
javascript
var writer = getWritableStreamSomehow();
for (var i = 0; i < 100; i ++) {
writer.write('hello, #' + i + '!\n');
}
writer.end('this is the end\n');
writer.on('finish', function() {
console.error('all writes are now complete.');
});事件: 'pipe'
src{Readable Stream} 是導(dǎo)流（pipe）到可寫流的源流。
無論何時(shí)在可寫流（Writable stream ）上調(diào)用pipe()方法，都會(huì)觸發(fā)'pipe'事件，添加這個(gè)流到目標(biāo)。
javascript
var writer = getWritableStreamSomehow();
var reader = getReadableStreamSomehow();
writer.on('pipe', function(src) {
console.error('something is piping into the writer');
assert.equal(src, reader);
});
reader.pipe(writer);事件: 'unpipe'
src{Readable Stream}未寫入此可寫的源流。
無論何時(shí)在可寫流（Writable stream ）上調(diào)用unpipe()方法，都會(huì)觸發(fā)'unpipe'事件，將這個(gè)流從目標(biāo)上移除。
javascript
var writer = getWritableStreamSomehow();
var reader = getReadableStreamSomehow();
writer.on('unpipe', function(src) {
console.error('something has stopped piping into the writer');
assert.equal(src, reader);
});
reader.pipe(writer);
reader.unpipe(writer);事件: 'error'
- {Error object}
寫或?qū)Я鳎╬ipe）數(shù)據(jù)時(shí)，如果有錯(cuò)誤會(huì)觸發(fā)。
類： stream.Duplex
雙工流（Duplex streams）是同時(shí)實(shí)現(xiàn)了Readable和Writable 接口。用法詳見下文。
雙工流（Duplex streams） 的例子包括:
- tcp sockets
- zlib streams
- crypto streams
類： stream.Transform
轉(zhuǎn)換流（Transform streams）是雙工Duplex流，它的輸出是從輸入計(jì)算得來。 它實(shí)現(xiàn)了Readable和Writable接口. 用法詳見下文.
轉(zhuǎn)換流（Transform streams）的例子包括:
- zlib streams
- crypto streams
流實(shí)現(xiàn)程序API
無論實(shí)現(xiàn)什么形式的流，模式都是一樣的:
- 在你的子類中擴(kuò)展適合的父類。 (
util.inherits方法很有幫助) - 在你的構(gòu)造函數(shù)中調(diào)用父類的構(gòu)造函數(shù)，以確保內(nèi)部的機(jī)制初始化正確。
- 實(shí)現(xiàn)一個(gè)或多個(gè)方法，如下所列。
所擴(kuò)展的類和要實(shí)現(xiàn)的方法取決于你要編寫的流類。
|
Use-case |
Class |
方法(s) to implement |
|---|---|---|
|
Reading only |
[Readable](#stream_class_stream_readable_1) |
|
|
Writing only |
[Writable](#stream_class_stream_writable_1) |
|
|
Reading and writing |
[Duplex](#stream_class_stream_duplex_1) |
|
|
Operate on written data, then read the result |
[Transform](#stream_class_stream_transform_1) |
|
在你的代碼里，千萬不要調(diào)用流實(shí)現(xiàn)程序API里的方法。否則可能會(huì)引起消費(fèi)流的程序副作用。
類： stream.Readable
stream.Readable是一個(gè)可被擴(kuò)充的、實(shí)現(xiàn)了底層_read(size)方法的抽象類。
參照之前的流實(shí)現(xiàn)程序API查看如何在你的程序里消費(fèi)流。以下內(nèi)容解釋了在你的程序里如何實(shí)現(xiàn)可讀流（Readable stream）。
Example: 計(jì)數(shù)流
這是可讀流（Readable stream）的基礎(chǔ)例子，它將從1至1,000,000遞增地觸發(fā)數(shù)字，然后結(jié)束：
javascript
var Readable = require('stream').Readable;
var util = require('util');
util.inherits(Counter, Readable);
function Counter(opt) {
Readable.call(this, opt);
this._max = 1000000;
this._index = 1;
}
Counter.prototype._read = function() {
var i = this._index++;
if (i > this._max)
this.push(null);
else {
var str = '' + i;
var buf = new Buffer(str, 'ascii');
this.push(buf);
}
};Example: 簡(jiǎn)單協(xié)議 v1 (初始版)
和之前描述的parseHeader函數(shù)類似，但它被實(shí)現(xiàn)為自定義流。注意這個(gè)實(shí)現(xiàn)不會(huì)將輸入數(shù)據(jù)轉(zhuǎn)換為字符串。
實(shí)際上，更好的辦法是將他實(shí)現(xiàn)為Transform流，使用下面的實(shí)現(xiàn)方法會(huì)更好：
javascript
// A parser for a simple data protocol.
// "header" is a JSON object, followed by 2 \n characters, and
// then a message body.
//
// 注意: This can be done more simply as a Transform stream!
// Using Readable directly for this is sub-optimal. See the
// alternative example below under Transform section.
var Readable = require('stream').Readable;
var util = require('util');
util.inherits(SimpleProtocol, Readable);
function SimpleProtocol(source, options) {
if (!(this instanceof SimpleProtocol))
return new SimpleProtocol(source, options);
Readable.call(this, options;
this._inBody = false;
this._sawFirstCr = false;
// source is 可讀流（Readable stream）, such as a socket or file
this._source = source;
var self = this;
source.on('end', function() {
self.push(null);
});
// give it a kick whenever the source is readable
// read(0) will not consume any bytes
source.on('readable', function() {
self.read(0);
});
this._rawHeader = [];
this.header = null;
}
SimpleProtocol.prototype._read = function(n) {
if (!this._inBody) {
var chunk = this._source.read();
// if the source doesn't have data, we don't have data yet.
if (chunk === null)
return this.push('');
// check if the chunk has a \n\n
var split = -1;
for (var i = 0; i < chunk.length; i++) {
if (chunk[i] === 10) { // '\n'
if (this._sawFirstCr) {
split = i;
break;
} else {
this._sawFirstCr = true;
}
} else {
this._sawFirstCr = false;
}
}
if (split === -1) {
// still waiting for the \n\n
// stash the chunk, and try again.
this._rawHeader.push(chunk);
this.push('');
} else {
this._inBody = true;
var h = chunk.slice(0, split);
this._rawHeader.push(h);
var header = Buffer.concat(this._rawHeader).toString();
try {
this.header = JSON.parse(header);
} catch (er) {
this.emit('error', new Error('invalid simple protocol data'));
return;
}
// now, because we got some extra data, unshift the rest
// back into the 讀取隊(duì)列 so that our consumer will see it.
var b = chunk.slice(split);
this.unshift(b);
// and let them know that we are done parsing the header.
this.emit('header', this.header);
}
} else {
// from there on, just provide the data to our consumer.
// careful not to push(null), since that would indicate EOF.
var chunk = this._source.read();
if (chunk) this.push(chunk);
}
};
// Usage:
// var parser = new SimpleProtocol(source);
// Now parser is 可讀流（Readable stream） that will emit 'header'
// with the parsed header data.new stream.Readable([options])
options{Object}highWaterMark{Number} 停止從底層資源讀取數(shù)據(jù)前，存儲(chǔ)在內(nèi)部緩存的最大字節(jié)數(shù)；默認(rèn)=16kb，objectMode流是16.encoding{String} 若指定，則Buffer會(huì)被解碼成所給編碼的字符串，默認(rèn)為null。objectMode{Boolean} 該流是否為對(duì)象的流。意思是說stream.read(n)返回一個(gè)單獨(dú)的值，而不是大小為n的Buffer。
Readable的擴(kuò)展類中，確保調(diào)用了Readable的構(gòu)造函數(shù)，這樣才能正確初始化。
readable._read(size)
size{Number} 異步讀取的字節(jié)數(shù)
注意：實(shí)現(xiàn)這個(gè)函數(shù)，但不要直接調(diào)用。
這個(gè)函數(shù)不要直接調(diào)用。在子類里實(shí)現(xiàn)，僅能被內(nèi)部的Readable類調(diào)用。
所有可讀流（Readable stream） 的實(shí)現(xiàn)必須停供一個(gè)_read方法，從底層資源里獲取數(shù)據(jù)。
這個(gè)方法以下劃線開頭，是因?yàn)閷?duì)于定義它的類是內(nèi)部的，不會(huì)被用戶程序直接調(diào)用。你可以在自己的擴(kuò)展類中實(shí)現(xiàn)。
當(dāng)數(shù)據(jù)可用時(shí)，通過調(diào)用readable.push(chunk)將之放到讀取隊(duì)列中。再次調(diào)用_read，需要繼續(xù)推出更多數(shù)據(jù)。
size參數(shù)僅供參考。調(diào)用“read”可以知道知道應(yīng)當(dāng)抓取多少數(shù)據(jù)；其余與之無關(guān)的實(shí)現(xiàn)，比如TCP或TLS，則可忽略這個(gè)參數(shù)，并在可用時(shí)返回?cái)?shù)據(jù)。例如，沒有必要“等到”size個(gè)字節(jié)可用時(shí)才調(diào)用stream.push(chunk)。
readable.push(chunk[, encoding])
chunk{Buffer | null | String} 推入到讀取隊(duì)列的數(shù)據(jù)塊encoding{String} 字符串塊的編碼。必須是有效的Buffer編碼，比如utf8或ascii。- 返回{Boolean}是否應(yīng)該繼續(xù)推入
注意:這個(gè)函數(shù)必須被 Readable 實(shí)現(xiàn)者調(diào)用, 而不是可讀流（Readable stream）的消費(fèi)者.
_read()函數(shù)直到調(diào)用push(chunk)后才能被再次調(diào)用。
Readable類將數(shù)據(jù)放到讀取隊(duì)列，當(dāng)'readable'事件觸發(fā)后，被read()方法取出。push()方法會(huì)插入數(shù)據(jù)到讀取隊(duì)列中。如果調(diào)用了null，會(huì)觸發(fā)數(shù)據(jù)結(jié)束信號(hào) (EOF)。
這個(gè)API被設(shè)計(jì)成盡可能地靈活。比如說，你可以包裝一個(gè)低級(jí)別的，具備某種暫停/恢復(fù)機(jī)制，和數(shù)據(jù)回調(diào)的數(shù)據(jù)源。這種情況下，你可以通過這種方式包裝低級(jí)別來源對(duì)象：
javascript
// source is an object with readStop() and readStart() 方法s,
// and an `ondata` member that gets called when it has data, and
// an `onend` member that gets called when the data is over.
util.inherits(SourceWrapper, Readable);
function SourceWrapper(options) {
Readable.call(this, options);
this._source = getLowlevelSourceObject();
var self = this;
// Every time there's data, we push it into the internal buffer.
this._source.ondata = function(chunk) {
// if push() 返回 false, then we need to stop reading from source
if (!self.push(chunk))
self._source.readStop();
};
// When the source ends, we push the EOF-signaling `null` chunk
this._source.onend = function() {
self.push(null);
};
}
// _read will be called when the stream wants to pull more data in
// the advisory size 參數(shù) is ignored in this case.
SourceWrapper.prototype._read = function(size) {
this._source.readStart();
};類： stream.Writable
stream.Writable是個(gè)抽象類，它擴(kuò)展了一個(gè)底層的實(shí)現(xiàn)_write(chunk, encoding, callback)方法.
參考上面的流實(shí)現(xiàn)程序API，來了解在你的程序里如何消費(fèi)可寫流。下面內(nèi)容介紹了如何在你的程序里實(shí)現(xiàn)可寫流。
new stream.Writable([options])
options{Object}highWaterMark{Number} 當(dāng)write()返回false時(shí)的緩存級(jí)別。默認(rèn)=16kb，objectMode流是16。decodeStrings{Boolean} 傳給_write()前是否解碼為字符串。默認(rèn)=trueobjectMode{Boolean}write(anyObj)是否是有效操作；如果為true，可以寫任意數(shù)據(jù)，而不僅僅是Buffer/String。默認(rèn)=false
請(qǐng)確保Writable類的擴(kuò)展類中，調(diào)用構(gòu)造函數(shù)以便緩沖設(shè)定能被正確初始化。
writable._write(chunk, encoding, callback)
chunk{Buffer | String} 要寫入的數(shù)據(jù)塊?？偸莃uffer， 除非decodeStrings選項(xiàng)為false。encoding{String} 如果數(shù)據(jù)塊是字符串，這個(gè)參數(shù)就是編碼方式。如果是緩存，則忽略。注意，除非decodeStrings被設(shè)置為false，否則這個(gè)數(shù)據(jù)塊一直是buffer。callback{函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個(gè)函數(shù) (錯(cuò)誤參數(shù)為可選參數(shù))。
所以可寫流（Writable stream ） 實(shí)現(xiàn)必須提供一個(gè)_write()方法，來發(fā)送數(shù)據(jù)給底層資源。
注意:這個(gè)函數(shù)不能直接調(diào)用，由子類實(shí)現(xiàn)，僅內(nèi)部可寫方法可以調(diào)用。
使用標(biāo)準(zhǔn)的callback(error)方法調(diào)用回調(diào)函數(shù)，來表明寫入完成或遇到錯(cuò)誤。
如果構(gòu)造函數(shù)選項(xiàng)中設(shè)定了decodeStrings標(biāo)識(shí)，則chunk可能會(huì)是字符串而不是Buffer，encoding表明了字符串的格式。這種設(shè)計(jì)是為了支持對(duì)某些字符串?dāng)?shù)據(jù)編碼提供優(yōu)化處理的實(shí)現(xiàn)。如果你沒有明確的設(shè)置decodeStrings為false，這樣你就可以安不管encoding參數(shù)，并假定chunk一直是一個(gè)緩存。
該方法以下劃線開頭，是因?yàn)閷?duì)于定義它的類來說，這個(gè)方法是內(nèi)部的，并且不應(yīng)該被用戶程序直接調(diào)用。你應(yīng)當(dāng)在你的擴(kuò)充類中重寫這個(gè)方法。
writable._writev(chunks, callback)
chunks{Array} 準(zhǔn)備寫入的數(shù)據(jù)塊，每個(gè)塊格式如下:{ chunk: ..., encoding: ... }.callback{函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個(gè)函數(shù) (錯(cuò)誤參數(shù)為可選參數(shù))。
注意:這個(gè)函數(shù)不能直接調(diào)用。由子類實(shí)現(xiàn)，僅內(nèi)部可寫方法可以調(diào)用.
這個(gè)函數(shù)的實(shí)現(xiàn)是可選的。多數(shù)情況下，沒有必要實(shí)現(xiàn)。如果實(shí)現(xiàn)，將會(huì)在所有數(shù)據(jù)塊緩存到寫隊(duì)列后調(diào)用。
類： stream.Duplex
雙工流（duplex stream）同時(shí)兼具可讀和可寫特性，比如一個(gè)TCP socket連接。
注意stream.Duplex可以像Readable或Writable一樣被擴(kuò)充，實(shí)現(xiàn)了底層_read(sise) 和_write(chunk, encoding, callback) 方法的抽象類。
由于JavaScript并沒有多重繼承能力，因此這個(gè)類繼承自Readable，寄生自Writable.從而讓用戶在雙工擴(kuò)展類中同時(shí)實(shí)現(xiàn)低級(jí)別的_read(n)方法和低級(jí)別的_write(chunk, encoding, callback)方法。
new stream.Duplex(options)
options{Object} 傳遞Writable and Readable構(gòu)造函數(shù)，有以下的內(nèi)容：allowHalfOpen{Boolean} 默認(rèn)=true。 如果設(shè)置為false，當(dāng)寫端結(jié)束的時(shí)候，流會(huì)自動(dòng)的結(jié)束讀端，反之亦然。readableObjectMode{Boolean} 默認(rèn)=false。將objectMode設(shè)為讀端的流，如果為true，將沒有效果。writableObjectMode{Boolean} 默認(rèn)=false。將objectMode設(shè)為寫端的流，如果為true，將沒有效果。
擴(kuò)展自Duplex的類，確保調(diào)用了父親的構(gòu)造函數(shù)，保證緩存設(shè)置能正確初始化。
類： stream.Transform
轉(zhuǎn)換流（transform class) 是雙工流（duplex stream），輸入輸出端有因果關(guān)系，比如，zlib流或crypto流。
輸入輸出沒有要求大小相同，塊數(shù)量相同，到達(dá)時(shí)間相同。例如，一個(gè)Hash流只會(huì)在輸入結(jié)束時(shí)產(chǎn)生一個(gè)數(shù)據(jù)塊的輸出；一個(gè)zlib流會(huì)產(chǎn)生比輸入小得多或大得多的輸出。
轉(zhuǎn)換流（transform class) 必須實(shí)現(xiàn)_transform()方法，而不是_read()和_write()方法，也可以實(shí)現(xiàn)_flush()方法（參見如下）。
new stream.Transform([options])
options{Object} 傳遞給Writable和Readable構(gòu)造函數(shù)。
擴(kuò)展自轉(zhuǎn)換流（transform class) 的類，確保調(diào)用了父親的構(gòu)造函數(shù)，保證緩存設(shè)置能正確初始化。
transform._transform(chunk, encoding, callback)
chunk{Buffer | String} 準(zhǔn)備轉(zhuǎn)換的數(shù)據(jù)塊。是buffer，除非decodeStrings選項(xiàng)設(shè)置為false。encoding{String} 如果數(shù)據(jù)塊是字符串, 這個(gè)參數(shù)就是編碼方式，否則就忽略這個(gè)參數(shù)callback{函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個(gè)函數(shù) (錯(cuò)誤參數(shù)為可選參數(shù))。
注意：這個(gè)函數(shù)不能直接調(diào)用。由子類實(shí)現(xiàn)，僅內(nèi)部可寫方法可以調(diào)用.
所有的轉(zhuǎn)換流（transform class) 實(shí)現(xiàn)必須提供 _transform方法來接收輸入，并生產(chǎn)輸出。
_transform可以做轉(zhuǎn)換流（transform class)里的任何事，處理寫入的字節(jié)，傳給接口的寫端，異步I/O，處理事情等等。
調(diào)用transform.push(outputChunk)0次或多次，從這個(gè)輸入塊里產(chǎn)生輸出，依賴于你想要多少數(shù)據(jù)作為輸出。
僅在當(dāng)前數(shù)據(jù)塊完全消費(fèi)后調(diào)用這個(gè)回調(diào)。
注意，輸入塊可能有，也可能沒有對(duì)應(yīng)的輸出塊。如果你提供了第二個(gè)參數(shù)，將會(huì)傳給push方法。如下述的例子：
javascript
transform.prototype._transform = function (data, encoding, callback) {
this.push(data);
callback();
}
transform.prototype._transform = function (data, encoding, callback) {
callback(null, data);
}該方法以下劃線開頭，是因?yàn)閷?duì)于定義它的類來說，這個(gè)方法是內(nèi)部的，并且不應(yīng)該被用戶程序直接調(diào)用。你應(yīng)當(dāng)在你的擴(kuò)充類中重寫這個(gè)方法。
transform._flush(callback)
callback{函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個(gè)函數(shù) (錯(cuò)誤參數(shù)為可選參數(shù))
注意：這個(gè)函數(shù)不能直接調(diào)用。由子類實(shí)現(xiàn)，僅內(nèi)部可寫方法可以調(diào)用.
某些情況下，轉(zhuǎn)換操作可能需要分發(fā)一點(diǎn)流最后的數(shù)據(jù)。例如，Zlib流會(huì)存儲(chǔ)一些內(nèi)部狀態(tài)，以便優(yōu)化壓縮輸出。
有些時(shí)候，你可以實(shí)現(xiàn)_flush方法，它可以在最后面調(diào)用，當(dāng)所有的寫入數(shù)據(jù)被消費(fèi)后，分發(fā)end告訴讀端。和_transform一樣，當(dāng)刷新操作完畢， transform.push(chunk)為0次或更多次數(shù)。
該方法以下劃線開頭，是因?yàn)閷?duì)于定義它的類來說，這個(gè)方法是內(nèi)部的，并且不應(yīng)該被用戶程序直接調(diào)用。你應(yīng)當(dāng)在你的擴(kuò)充類中重寫這個(gè)方法。
事件: 'finish' and 'end'
finish和end事件 分別來自Writable和Readable類。.end()事件結(jié)束后調(diào)用finish事件，所有的數(shù)據(jù)已經(jīng)被_transform處理完畢，調(diào)用_flush后，所有的數(shù)據(jù)輸出完畢，觸發(fā)end。
Example:SimpleProtocolparser v2
上面的簡(jiǎn)單協(xié)議分析例子列子可以通過使用高級(jí)別的Transform流來實(shí)現(xiàn)，和parseHeader ，SimpleProtocol v1列子類似。
在這個(gè)示例中，輸入會(huì)被導(dǎo)流到解析器中，而不是作為參數(shù)提供。這種做法更符合Node流的慣例。
javascript
var util = require('util');
var Transform = require('stream').Transform;
util.inherits(SimpleProtocol, Transform);
function SimpleProtocol(options) {
if (!(this instanceof SimpleProtocol))
return new SimpleProtocol(options);
Transform.call(this, options);
this._inBody = false;
this._sawFirstCr = false;
this._rawHeader = [];
this.header = null;
}
SimpleProtocol.prototype._transform = function(chunk, encoding, done) {
if (!this._inBody) {
// check if the chunk has a \n\n
var split = -1;
for (var i = 0; i < chunk.length; i++) {
if (chunk[i] === 10) { // '\n'
if (this._sawFirstCr) {
split = i;
break;
} else {
this._sawFirstCr = true;
}
} else {
this._sawFirstCr = false;
}
}
if (split === -1) {
// still waiting for the \n\n
// stash the chunk, and try again.
this._rawHeader.push(chunk);
} else {
this._inBody = true;
var h = chunk.slice(0, split);
this._rawHeader.push(h);
var header = Buffer.concat(this._rawHeader).toString();
try {
this.header = JSON.parse(header);
} catch (er) {
this.emit('error', new Error('invalid simple protocol data'));
return;
}
// and let them know that we are done parsing the header.
this.emit('header', this.header);
// now, because we got some extra data, emit this first.
this.push(chunk.slice(split));
}
} else {
// from there on, just provide the data to our consumer as-is.
this.push(chunk);
}
done();
};
// Usage:
// var parser = new SimpleProtocol();
// source.pipe(parser)
// Now parser is 可讀流（Readable stream） that will emit 'header'
// with the parsed header data.類： stream.PassThrough
這是Transform流的簡(jiǎn)單實(shí)現(xiàn)，將輸入的字節(jié)簡(jiǎn)單的傳遞給輸出。它的主要用途是測(cè)試和演示。偶爾要構(gòu)建某種特殊流時(shí)也會(huì)用到。
流: 內(nèi)部細(xì)節(jié)
緩沖
可寫流（Writable streams ）和可讀流（Readable stream）都會(huì)緩存數(shù)據(jù)到內(nèi)部對(duì)象上，叫做_writableState.buffer或_readableState.buffer。
緩存的數(shù)據(jù)量，取決于構(gòu)造函數(shù)是傳入的highWaterMark參數(shù)。
調(diào)用stream.push(chunk)時(shí)，緩存數(shù)據(jù)到可讀流（Readable stream）。在數(shù)據(jù)消費(fèi)者調(diào)用stream.read()前，數(shù)據(jù)會(huì)一直緩存在內(nèi)部隊(duì)列中。
調(diào)用stream.write(chunk)時(shí)，緩存數(shù)據(jù)到可寫流（Writable stream）。即使write()返回false。
流（尤其是pipe()方法）得目的是限制數(shù)據(jù)的緩存量到一個(gè)可接受的水平，使得不同速度的源和目的不會(huì)淹沒可用內(nèi)存。
stream.read(0)
某些時(shí)候，你可能想不消費(fèi)數(shù)據(jù)的情況下，觸發(fā)底層可讀流（Readable stream）機(jī)制的刷新。這種情況下可以調(diào)用stream.read(0)，它總會(huì)返回null。
如果內(nèi)部讀取緩沖低于highWaterMark，并且流當(dāng)前不在讀取狀態(tài)，那么調(diào)用read(0)會(huì)觸發(fā)一個(gè)低級(jí)_read調(diào)用。
雖然基本上沒有必要這么做。但你在Node內(nèi)部的某些地方看到它確實(shí)這么做了，尤其是在Readable流類的內(nèi)部。
stream.push('')
推一個(gè)0字節(jié)的字符串或緩存 (不在Object mode時(shí))會(huì)發(fā)送有趣的副作用。 因?yàn)樗且粋€(gè)對(duì)stream.push()的調(diào)用，它將會(huì)結(jié)束reading進(jìn)程。然而，它沒有添加任何數(shù)據(jù)到可讀緩沖區(qū)中，所以沒有東西可供用戶消費(fèi)。
少數(shù)情況下，你當(dāng)時(shí)沒有提供數(shù)據(jù)，但你的流的消費(fèi)者（或你的代碼的其它部分）會(huì)通過調(diào)用stream.read(0)得知何時(shí)再次檢查。在這種情況下，你可以調(diào)用 stream.push('')。
到目前為止，這個(gè)功能唯一一個(gè)使用情景是在tls.CryptoStream類中，但它將在Node v0.12中被廢棄。如果你發(fā)現(xiàn)你不得不使用stream.push('')，請(qǐng)考慮另一種方式。
和老版本的兼容性
v0.10版本前，可讀流（Readable stream）接口比較簡(jiǎn)單，因此功能和用處也小。
'data'事件會(huì)立即開始觸發(fā)，而不會(huì)等待你調(diào)用read()方法。如果你需要進(jìn)行某些I/O來決定如何處理數(shù)據(jù)，那么你只能將數(shù)據(jù)塊儲(chǔ)存到某種緩沖區(qū)中以防它們流失。pause()方法僅供參考，而不保證生效。這意味著，即便流處于暫停狀態(tài)時(shí)，你仍然需要準(zhǔn)備接收'data'事件。
在Node v0.10中, 加入了下文所述的Readable類。為了考慮向后兼容，添加了'data'事件監(jiān)聽器或resume()方法被調(diào)用時(shí)，可讀流（Readable stream）會(huì)切換到 "流動(dòng)模式（flowing mode）"。其作用是，即便你不使用新的read()方法和'readable'事件，你也不必?fù)?dān)心丟失'data'數(shù)據(jù)塊。
大多數(shù)程序會(huì)維持正常功能。然而，下列條件下也會(huì)引入邊界情況：
- 沒有添加 [
'data'事件][]處理器 - 從來沒有調(diào)用
resume()方法 - 流從來沒有被倒流（pipe）到任何可寫目標(biāo)上、
例如：
javascript
// WARNING! BROKEN!
net.createServer(function(socket) {
// we add an 'end' 方法, but never consume the data
socket.on('end', function() {
// It will never get here.
socket.end('I got your message (but didnt read it)\n');
});
}).listen(1337);v0.10版本前的Node，流入的消息數(shù)據(jù)會(huì)被簡(jiǎn)單的拋棄。之后的版本，socket會(huì)一直保持暫停。
這種情形下，調(diào)用resume()方法來開始工作：
javascript
// Workaround
net.createServer(function(socket) {
socket.on('end', function() {
socket.end('I got your message (but didnt read it)\n');
});
// start the flow of data, discarding it.
socket.resume();
}).listen(1337);可讀流（Readable stream）切換到流動(dòng)模式（flowing mode），v0.10 版本前，可以使用wrap()方法將風(fēng)格流包含在一個(gè)可讀類里。
Object Mode
通常情況下，流僅操作字符串和緩存。
處于object mode的流，除了緩存和字符串，還可以可以讀出普通JavaScript值。
在對(duì)象模式里，可讀流（Readable stream） 調(diào)用stream.read(size)總會(huì)返回單個(gè)項(xiàng)目，無論是什么參數(shù)。
在對(duì)象模式里， 可寫流（Writable stream ） 總會(huì)忽略傳給stream.write(data, encoding)的encoding參數(shù)。
特殊值null在對(duì)象模式里，依舊保持它的特殊性。也就說，對(duì)于對(duì)象模式的可讀流（Readable stream），stream.read()返回null意味著沒有更多數(shù)據(jù)，同時(shí)stream.push(null)會(huì)告知流數(shù)據(jù)結(jié)束（EOF）。
Node核心不存在對(duì)象模式的流，這種設(shè)計(jì)只被某些用戶態(tài)流式庫所使用。
應(yīng)該在你的子類構(gòu)造函數(shù)里，設(shè)置objectMode。在過程中設(shè)置不安全。
對(duì)于雙工流（Duplex streams），objectMode可以用readableObjectMode和writableObjectMode分別為讀寫端分別設(shè)置。這些選項(xiàng)，被轉(zhuǎn)換流（Transform streams）用來實(shí)現(xiàn)解析和序列化。
javascript
var util = require('util');
var StringDecoder = require('string_decoder').StringDecoder;
var Transform = require('stream').Transform;
util.inherits(JSONParseStream, Transform);
// Gets \n-delimited JSON string data, and emits the parsed objects
function JSONParseStream() {
if (!(this instanceof JSONParseStream))
return new JSONParseStream();
Transform.call(this, { readableObjectMode : true });
this._buffer = '';
this._decoder = new StringDecoder('utf8');
}
JSONParseStream.prototype._transform = function(chunk, encoding, cb) {
this._buffer += this._decoder.write(chunk);
// split on newlines
var lines = this._buffer.split(/\r?\n/);
// keep the last partial line buffered
this._buffer = lines.pop();
for (var l = 0; l < lines.length; l++) {
var line = lines[l];
try {
var obj = JSON.parse(line);
} catch (er) {
this.emit('error', er);
return;
}
// push the parsed object out to the readable consumer
this.push(obj);
}
cb();
};
JSONParseStream.prototype._flush = function(cb) {
// Just handle any leftover
var rem = this._buffer.trim();
if (rem) {
try {
var obj = JSON.parse(rem);
} catch (er) {
this.emit('error', er);
return;
}
// push the parsed object out to the readable consumer
this.push(obj);
}
cb();
};網(wǎng)站題目：創(chuàng)新互聯(lián)Node.js教程：Node.js流
分享鏈接：http://fisionsoft.com.cn/article/ccohegc.html
