5 A stream is an abstract interface implemented by various objects in
6 io.js. For example a [request to an HTTP
7 server](https://iojs.org/dist/v2.3.0/doc/api/http.html#http_http_incomingmessage) is a stream, as is
8 [stdout][]. Streams are readable, writable, or both. All streams are
9 instances of [EventEmitter][]
11 You can load the Stream base classes by doing `require('stream')`.
12 There are base classes provided for [Readable][] streams, [Writable][]
13 streams, [Duplex][] streams, and [Transform][] streams.
15 This document is split up into 3 sections. The first explains the
16 parts of the API that you need to be aware of to use streams in your
17 programs. If you never implement a streaming API yourself, you can
20 The second section explains the parts of the API that you need to use
21 if you implement your own custom streams yourself. The API is
22 designed to make this easy for you to do.
24 The third section goes into more depth about how streams work,
25 including some of the internal mechanisms and functions that you
26 should probably not modify unless you definitely know what you are
30 ## API for Stream Consumers
34 Streams can be either [Readable][], [Writable][], or both ([Duplex][]).
36 All streams are EventEmitters, but they also have other custom methods
37 and properties depending on whether they are Readable, Writable, or
40 If a stream is both Readable and Writable, then it implements all of
41 the methods and events below. So, a [Duplex][] or [Transform][] stream is
42 fully described by this API, though their implementation may be
45 It is not necessary to implement Stream interfaces in order to consume
46 streams in your programs. If you **are** implementing streaming
47 interfaces in your own program, please also refer to
48 [API for Stream Implementors][] below.
50 Almost all io.js programs, no matter how simple, use Streams in some
51 way. Here is an example of using Streams in an io.js program:
54 var http = require('http');
56 var server = http.createServer(function (req, res) {
57 // req is an http.IncomingMessage, which is a Readable Stream
58 // res is an http.ServerResponse, which is a Writable Stream
61 // we want to get the data as utf8 strings
62 // If you don't set an encoding, then you'll get Buffer objects
63 req.setEncoding('utf8');
65 // Readable streams emit 'data' events once a listener is added
66 req.on('data', function (chunk) {
70 // the end event tells you that you have entire body
71 req.on('end', function () {
73 var data = JSON.parse(body);
77 return res.end('error: ' + er.message);
80 // write back something interesting to the user:
81 res.write(typeof data);
88 // $ curl localhost:1337 -d '{}'
90 // $ curl localhost:1337 -d '"foo"'
92 // $ curl localhost:1337 -d 'not json'
93 // error: Unexpected token o
96 ### Class: stream.Readable
100 The Readable stream interface is the abstraction for a *source* of
101 data that you are reading from. In other words, data comes *out* of a
104 A Readable stream will not start emitting data until you indicate that
105 you are ready to receive it.
107 Readable streams have two "modes": a **flowing mode** and a **paused
108 mode**. When in flowing mode, data is read from the underlying system
109 and provided to your program as fast as possible. In paused mode, you
110 must explicitly call `stream.read()` to get chunks of data out.
111 Streams start out in paused mode.
113 **Note**: If no data event handlers are attached, and there are no
114 [`pipe()`][] destinations, and the stream is switched into flowing
115 mode, then data will be lost.
117 You can switch to flowing mode by doing any of the following:
119 * Adding a [`'data'` event][] handler to listen for data.
120 * Calling the [`resume()`][] method to explicitly open the flow.
121 * Calling the [`pipe()`][] method to send the data to a [Writable][].
123 You can switch back to paused mode by doing either of the following:
125 * If there are no pipe destinations, by calling the [`pause()`][]
127 * If there are pipe destinations, by removing any [`'data'` event][]
128 handlers, and removing all pipe destinations by calling the
129 [`unpipe()`][] method.
131 Note that, for backwards compatibility reasons, removing `'data'`
132 event handlers will **not** automatically pause the stream. Also, if
133 there are piped destinations, then calling `pause()` will not
134 guarantee that the stream will *remain* paused once those
135 destinations drain and ask for more data.
137 Examples of readable streams include:
139 * [http responses, on the client](https://iojs.org/dist/v2.3.0/doc/api/http.html#http_http_incomingmessage)
140 * [http requests, on the server](https://iojs.org/dist/v2.3.0/doc/api/http.html#http_http_incomingmessage)
141 * [fs read streams](https://iojs.org/dist/v2.3.0/doc/api/fs.html#fs_class_fs_readstream)
145 * [child process stdout and stderr][]
148 #### Event: 'readable'
150 When a chunk of data can be read from the stream, it will emit a
153 In some cases, listening for a `'readable'` event will cause some data
154 to be read into the internal buffer from the underlying system, if it
158 var readable = getReadableStreamSomehow();
159 readable.on('readable', function() {
160 // there is some data to read now
164 Once the internal buffer is drained, a `readable` event will fire
165 again when more data is available.
169 * `chunk` {Buffer | String} The chunk of data.
171 Attaching a `data` event listener to a stream that has not been
172 explicitly paused will switch the stream into flowing mode. Data will
173 then be passed as soon as it is available.
175 If you just want to get all the data out of the stream as fast as
176 possible, this is the best way to do so.
179 var readable = getReadableStreamSomehow();
180 readable.on('data', function(chunk) {
181 console.log('got %d bytes of data', chunk.length);
187 This event fires when there will be no more data to read.
189 Note that the `end` event **will not fire** unless the data is
190 completely consumed. This can be done by switching into flowing mode,
191 or by calling `read()` repeatedly until you get to the end.
194 var readable = getReadableStreamSomehow();
195 readable.on('data', function(chunk) {
196 console.log('got %d bytes of data', chunk.length);
198 readable.on('end', function() {
199 console.log('there will be no more data.');
205 Emitted when the underlying resource (for example, the backing file
206 descriptor) has been closed. Not all streams will emit this.
212 Emitted if there was an error receiving data.
214 #### readable.read([size])
216 * `size` {Number} Optional argument to specify how much data to read.
217 * Return {String | Buffer | null}
219 The `read()` method pulls some data out of the internal buffer and
220 returns it. If there is no data available, then it will return
223 If you pass in a `size` argument, then it will return that many
224 bytes. If `size` bytes are not available, then it will return `null`.
226 If you do not specify a `size` argument, then it will return all the
227 data in the internal buffer.
229 This method should only be called in paused mode. In flowing mode,
230 this method is called automatically until the internal buffer is
234 var readable = getReadableStreamSomehow();
235 readable.on('readable', function() {
237 while (null !== (chunk = readable.read())) {
238 console.log('got %d bytes of data', chunk.length);
243 If this method returns a data chunk, then it will also trigger the
244 emission of a [`'data'` event][].
246 #### readable.setEncoding(encoding)
248 * `encoding` {String} The encoding to use.
251 Call this function to cause the stream to return strings of the
252 specified encoding instead of Buffer objects. For example, if you do
253 `readable.setEncoding('utf8')`, then the output data will be
254 interpreted as UTF-8 data, and returned as strings. If you do
255 `readable.setEncoding('hex')`, then the data will be encoded in
256 hexadecimal string format.
258 This properly handles multi-byte characters that would otherwise be
259 potentially mangled if you simply pulled the Buffers directly and
260 called `buf.toString(encoding)` on them. If you want to read the data
261 as strings, always use this method.
264 var readable = getReadableStreamSomehow();
265 readable.setEncoding('utf8');
266 readable.on('data', function(chunk) {
267 assert.equal(typeof chunk, 'string');
268 console.log('got %d characters of string data', chunk.length);
272 #### readable.resume()
276 This method will cause the readable stream to resume emitting `data`
279 This method will switch the stream into flowing mode. If you do *not*
280 want to consume the data from a stream, but you *do* want to get to
281 its `end` event, you can call [`readable.resume()`][] to open the flow of
285 var readable = getReadableStreamSomehow();
287 readable.on('end', function() {
288 console.log('got to the end, but did not read anything');
292 #### readable.pause()
296 This method will cause a stream in flowing mode to stop emitting
297 `data` events, switching out of flowing mode. Any data that becomes
298 available will remain in the internal buffer.
301 var readable = getReadableStreamSomehow();
302 readable.on('data', function(chunk) {
303 console.log('got %d bytes of data', chunk.length);
305 console.log('there will be no more data for 1 second');
306 setTimeout(function() {
307 console.log('now data will start flowing again');
313 #### readable.isPaused()
317 This method returns whether or not the `readable` has been **explicitly**
318 paused by client code (using `readable.pause()` without a corresponding
319 `readable.resume()`).
322 var readable = new stream.Readable
324 readable.isPaused() // === false
326 readable.isPaused() // === true
328 readable.isPaused() // === false
331 #### readable.pipe(destination[, options])
333 * `destination` {[Writable][] Stream} The destination for writing data
334 * `options` {Object} Pipe options
335 * `end` {Boolean} End the writer when the reader ends. Default = `true`
337 This method pulls all the data out of a readable stream, and writes it
338 to the supplied destination, automatically managing the flow so that
339 the destination is not overwhelmed by a fast readable stream.
341 Multiple destinations can be piped to safely.
344 var readable = getReadableStreamSomehow();
345 var writable = fs.createWriteStream('file.txt');
346 // All the data from readable goes into 'file.txt'
347 readable.pipe(writable);
350 This function returns the destination stream, so you can set up pipe
354 var r = fs.createReadStream('file.txt');
355 var z = zlib.createGzip();
356 var w = fs.createWriteStream('file.txt.gz');
360 For example, emulating the Unix `cat` command:
363 process.stdin.pipe(process.stdout);
366 By default [`end()`][] is called on the destination when the source stream
367 emits `end`, so that `destination` is no longer writable. Pass `{ end:
368 false }` as `options` to keep the destination stream open.
370 This keeps `writer` open so that "Goodbye" can be written at the
374 reader.pipe(writer, { end: false });
375 reader.on('end', function() {
376 writer.end('Goodbye\n');
380 Note that `process.stderr` and `process.stdout` are never closed until
381 the process exits, regardless of the specified options.
383 #### readable.unpipe([destination])
385 * `destination` {[Writable][] Stream} Optional specific stream to unpipe
387 This method will remove the hooks set up for a previous `pipe()` call.
389 If the destination is not specified, then all pipes are removed.
391 If the destination is specified, but no pipe is set up for it, then
395 var readable = getReadableStreamSomehow();
396 var writable = fs.createWriteStream('file.txt');
397 // All the data from readable goes into 'file.txt',
398 // but only for the first second
399 readable.pipe(writable);
400 setTimeout(function() {
401 console.log('stop writing to file.txt');
402 readable.unpipe(writable);
403 console.log('manually close the file stream');
408 #### readable.unshift(chunk)
410 * `chunk` {Buffer | String} Chunk of data to unshift onto the read queue
412 This is useful in certain cases where a stream is being consumed by a
413 parser, which needs to "un-consume" some data that it has
414 optimistically pulled out of the source, so that the stream can be
415 passed on to some other party.
417 If you find that you must often call `stream.unshift(chunk)` in your
418 programs, consider implementing a [Transform][] stream instead. (See API
419 for Stream Implementors, below.)
422 // Pull off a header delimited by \n\n
423 // use unshift() if we get too much
424 // Call the callback with (error, header, stream)
425 var StringDecoder = require('string_decoder').StringDecoder;
426 function parseHeader(stream, callback) {
427 stream.on('error', callback);
428 stream.on('readable', onReadable);
429 var decoder = new StringDecoder('utf8');
431 function onReadable() {
433 while (null !== (chunk = stream.read())) {
434 var str = decoder.write(chunk);
435 if (str.match(/\n\n/)) {
436 // found the header boundary
437 var split = str.split(/\n\n/);
438 header += split.shift();
439 var remaining = split.join('\n\n');
440 var buf = new Buffer(remaining, 'utf8');
443 stream.removeListener('error', callback);
444 stream.removeListener('readable', onReadable);
445 // now the body of the message can be read from the stream.
446 callback(null, header, stream);
448 // still reading the header.
456 #### readable.wrap(stream)
458 * `stream` {Stream} An "old style" readable stream
460 Versions of Node.js prior to v0.10 had streams that did not implement the
461 entire Streams API as it is today. (See "Compatibility" below for
464 If you are using an older io.js library that emits `'data'` events and
465 has a [`pause()`][] method that is advisory only, then you can use the
466 `wrap()` method to create a [Readable][] stream that uses the old stream
469 You will very rarely ever need to call this function, but it exists
470 as a convenience for interacting with old io.js programs and libraries.
475 var OldReader = require('./old-api-module.js').OldReader;
476 var oreader = new OldReader;
477 var Readable = require('stream').Readable;
478 var myReader = new Readable().wrap(oreader);
480 myReader.on('readable', function() {
481 myReader.read(); // etc.
486 ### Class: stream.Writable
490 The Writable stream interface is an abstraction for a *destination*
491 that you are writing data *to*.
493 Examples of writable streams include:
495 * [http requests, on the client](https://iojs.org/dist/v2.3.0/doc/api/http.html#http_class_http_clientrequest)
496 * [http responses, on the server](https://iojs.org/dist/v2.3.0/doc/api/http.html#http_class_http_serverresponse)
497 * [fs write streams](https://iojs.org/dist/v2.3.0/doc/api/fs.html#fs_class_fs_writestream)
501 * [child process stdin](https://iojs.org/dist/v2.3.0/doc/api/child_process.html#child_process_child_stdin)
502 * [process.stdout][], [process.stderr][]
504 #### writable.write(chunk[, encoding][, callback])
506 * `chunk` {String | Buffer} The data to write
507 * `encoding` {String} The encoding, if `chunk` is a String
508 * `callback` {Function} Callback for when this chunk of data is flushed
509 * Returns: {Boolean} True if the data was handled completely.
511 This method writes some data to the underlying system, and calls the
512 supplied callback once the data has been fully handled.
514 The return value indicates if you should continue writing right now.
515 If the data had to be buffered internally, then it will return
516 `false`. Otherwise, it will return `true`.
518 This return value is strictly advisory. You MAY continue to write,
519 even if it returns `false`. However, writes will be buffered in
520 memory, so it is best not to do this excessively. Instead, wait for
521 the `drain` event before writing more data.
525 If a [`writable.write(chunk)`][] call returns false, then the `drain`
526 event will indicate when it is appropriate to begin writing more data
530 // Write the data to the supplied writable stream 1MM times.
531 // Be attentive to back-pressure.
532 function writeOneMillionTimes(writer, data, encoding, callback) {
541 writer.write(data, encoding, callback);
543 // see if we should continue, or wait
544 // don't pass the callback, because we're not done yet.
545 ok = writer.write(data, encoding);
547 } while (i > 0 && ok);
549 // had to stop early!
550 // write some more once it drains
551 writer.once('drain', write);
559 Forces buffering of all writes.
561 Buffered data will be flushed either at `.uncork()` or at `.end()` call.
563 #### writable.uncork()
565 Flush all data, buffered since `.cork()` call.
567 #### writable.setDefaultEncoding(encoding)
569 * `encoding` {String} The new default encoding
571 Sets the default encoding for a writable stream.
573 #### writable.end([chunk][, encoding][, callback])
575 * `chunk` {String | Buffer} Optional data to write
576 * `encoding` {String} The encoding, if `chunk` is a String
577 * `callback` {Function} Optional callback for when the stream is finished
579 Call this method when no more data will be written to the stream. If
580 supplied, the callback is attached as a listener on the `finish` event.
582 Calling [`write()`][] after calling [`end()`][] will raise an error.
585 // write 'hello, ' and then end with 'world!'
586 var file = fs.createWriteStream('example.txt');
587 file.write('hello, ');
589 // writing more now is not allowed!
594 When the [`end()`][] method has been called, and all data has been flushed
595 to the underlying system, this event is emitted.
598 var writer = getWritableStreamSomehow();
599 for (var i = 0; i < 100; i ++) {
600 writer.write('hello, #' + i + '!\n');
602 writer.end('this is the end\n');
603 writer.on('finish', function() {
604 console.error('all writes are now complete.');
610 * `src` {[Readable][] Stream} source stream that is piping to this writable
612 This is emitted whenever the `pipe()` method is called on a readable
613 stream, adding this writable to its set of destinations.
616 var writer = getWritableStreamSomehow();
617 var reader = getReadableStreamSomehow();
618 writer.on('pipe', function(src) {
619 console.error('something is piping into the writer');
620 assert.equal(src, reader);
627 * `src` {[Readable][] Stream} The source stream that [unpiped][] this writable
629 This is emitted whenever the [`unpipe()`][] method is called on a
630 readable stream, removing this writable from its set of destinations.
633 var writer = getWritableStreamSomehow();
634 var reader = getReadableStreamSomehow();
635 writer.on('unpipe', function(src) {
636 console.error('something has stopped piping into the writer');
637 assert.equal(src, reader);
640 reader.unpipe(writer);
647 Emitted if there was an error when writing or piping data.
649 ### Class: stream.Duplex
651 Duplex streams are streams that implement both the [Readable][] and
652 [Writable][] interfaces. See above for usage.
654 Examples of Duplex streams include:
661 ### Class: stream.Transform
663 Transform streams are [Duplex][] streams where the output is in some way
664 computed from the input. They implement both the [Readable][] and
665 [Writable][] interfaces. See above for usage.
667 Examples of Transform streams include:
673 ## API for Stream Implementors
677 To implement any sort of stream, the pattern is the same:
679 1. Extend the appropriate parent class in your own subclass. (The
680 [`util.inherits`][] method is particularly helpful for this.)
681 2. Call the appropriate parent class constructor in your constructor,
682 to be sure that the internal mechanisms are set up properly.
683 2. Implement one or more specific methods, as detailed below.
685 The class to extend and the method(s) to implement depend on the sort
686 of stream class you are writing:
698 <p>Method(s) to implement</p>
707 <p>[Readable](#stream_class_stream_readable_1)</p>
710 <p><code>[_read][]</code></p>
718 <p>[Writable](#stream_class_stream_writable_1)</p>
721 <p><code>[_write][]</code>, <code>_writev</code></p>
726 <p>Reading and writing</p>
729 <p>[Duplex](#stream_class_stream_duplex_1)</p>
732 <p><code>[_read][]</code>, <code>[_write][]</code>, <code>_writev</code></p>
737 <p>Operate on written data, then read the result</p>
740 <p>[Transform](#stream_class_stream_transform_1)</p>
743 <p><code>_transform</code>, <code>_flush</code></p>
748 In your implementation code, it is very important to never call the
749 methods described in [API for Stream Consumers][] above. Otherwise, you
750 can potentially cause adverse side effects in programs that consume
751 your streaming interfaces.
753 ### Class: stream.Readable
757 `stream.Readable` is an abstract class designed to be extended with an
758 underlying implementation of the [`_read(size)`][] method.
760 Please see above under [API for Stream Consumers][] for how to consume
761 streams in your programs. What follows is an explanation of how to
762 implement Readable streams in your programs.
764 #### Example: A Counting Stream
768 This is a basic example of a Readable stream. It emits the numerals
769 from 1 to 1,000,000 in ascending order, and then ends.
772 var Readable = require('stream').Readable;
773 var util = require('util');
774 util.inherits(Counter, Readable);
776 function Counter(opt) {
777 Readable.call(this, opt);
782 Counter.prototype._read = function() {
783 var i = this._index++;
788 var buf = new Buffer(str, 'ascii');
794 #### Example: SimpleProtocol v1 (Sub-optimal)
796 This is similar to the `parseHeader` function described above, but
797 implemented as a custom stream. Also, note that this implementation
798 does not convert the incoming data to a string.
800 However, this would be better implemented as a [Transform][] stream. See
801 below for a better implementation.
804 // A parser for a simple data protocol.
805 // The "header" is a JSON object, followed by 2 \n characters, and
806 // then a message body.
808 // NOTE: This can be done more simply as a Transform stream!
809 // Using Readable directly for this is sub-optimal. See the
810 // alternative example below under the Transform section.
812 var Readable = require('stream').Readable;
813 var util = require('util');
815 util.inherits(SimpleProtocol, Readable);
817 function SimpleProtocol(source, options) {
818 if (!(this instanceof SimpleProtocol))
819 return new SimpleProtocol(source, options);
821 Readable.call(this, options);
822 this._inBody = false;
823 this._sawFirstCr = false;
825 // source is a readable stream, such as a socket or file
826 this._source = source;
829 source.on('end', function() {
833 // give it a kick whenever the source is readable
834 // read(0) will not consume any bytes
835 source.on('readable', function() {
839 this._rawHeader = [];
843 SimpleProtocol.prototype._read = function(n) {
845 var chunk = this._source.read();
847 // if the source doesn't have data, we don't have data yet.
849 return this.push('');
851 // check if the chunk has a \n\n
853 for (var i = 0; i < chunk.length; i++) {
854 if (chunk[i] === 10) { // '\n'
855 if (this._sawFirstCr) {
859 this._sawFirstCr = true;
862 this._sawFirstCr = false;
867 // still waiting for the \n\n
868 // stash the chunk, and try again.
869 this._rawHeader.push(chunk);
873 var h = chunk.slice(0, split);
874 this._rawHeader.push(h);
875 var header = Buffer.concat(this._rawHeader).toString();
877 this.header = JSON.parse(header);
879 this.emit('error', new Error('invalid simple protocol data'));
882 // now, because we got some extra data, unshift the rest
883 // back into the read queue so that our consumer will see it.
884 var b = chunk.slice(split);
887 // and let them know that we are done parsing the header.
888 this.emit('header', this.header);
891 // from there on, just provide the data to our consumer.
892 // careful not to push(null), since that would indicate EOF.
893 var chunk = this._source.read();
894 if (chunk) this.push(chunk);
899 // var parser = new SimpleProtocol(source);
900 // Now parser is a readable stream that will emit 'header'
901 // with the parsed header data.
905 #### new stream.Readable([options])
908 * `highWaterMark` {Number} The maximum number of bytes to store in
909 the internal buffer before ceasing to read from the underlying
910 resource. Default=16kb, or 16 for `objectMode` streams
911 * `encoding` {String} If specified, then buffers will be decoded to
912 strings using the specified encoding. Default=null
913 * `objectMode` {Boolean} Whether this stream should behave
914 as a stream of objects. Meaning that stream.read(n) returns
915 a single value instead of a Buffer of size n. Default=false
917 In classes that extend the Readable class, make sure to call the
918 Readable constructor so that the buffering settings can be properly
921 #### readable.\_read(size)
923 * `size` {Number} Number of bytes to read asynchronously
925 Note: **Implement this function, but do NOT call it directly.**
927 This function should NOT be called directly. It should be implemented
928 by child classes, and only called by the internal Readable class
931 All Readable stream implementations must provide a `_read` method to
932 fetch data from the underlying resource.
934 This method is prefixed with an underscore because it is internal to
935 the class that defines it, and should not be called directly by user
936 programs. However, you **are** expected to override this method in
937 your own extension classes.
939 When data is available, put it into the read queue by calling
940 `readable.push(chunk)`. If `push` returns false, then you should stop
941 reading. When `_read` is called again, you should start pushing more
944 The `size` argument is advisory. Implementations where a "read" is a
945 single call that returns data can use this to know how much data to
946 fetch. Implementations where that is not relevant, such as TCP or
947 TLS, may ignore this argument, and simply provide data whenever it
948 becomes available. There is no need, for example to "wait" until
949 `size` bytes are available before calling [`stream.push(chunk)`][].
951 #### readable.push(chunk[, encoding])
953 * `chunk` {Buffer | null | String} Chunk of data to push into the read queue
954 * `encoding` {String} Encoding of String chunks. Must be a valid
955 Buffer encoding, such as `'utf8'` or `'ascii'`
956 * return {Boolean} Whether or not more pushes should be performed
958 Note: **This function should be called by Readable implementors, NOT
959 by consumers of Readable streams.**
961 The `_read()` function will not be called again until at least one
962 `push(chunk)` call is made.
964 The `Readable` class works by putting data into a read queue to be
965 pulled out later by calling the `read()` method when the `'readable'`
968 The `push()` method will explicitly insert some data into the read
969 queue. If it is called with `null` then it will signal the end of the
972 This API is designed to be as flexible as possible. For example,
973 you may be wrapping a lower-level source which has some sort of
974 pause/resume mechanism, and a data callback. In those cases, you
975 could wrap the low-level source object by doing something like this:
978 // source is an object with readStop() and readStart() methods,
979 // and an `ondata` member that gets called when it has data, and
980 // an `onend` member that gets called when the data is over.
982 util.inherits(SourceWrapper, Readable);
984 function SourceWrapper(options) {
985 Readable.call(this, options);
987 this._source = getLowlevelSourceObject();
990 // Every time there's data, we push it into the internal buffer.
991 this._source.ondata = function(chunk) {
992 // if push() returns false, then we need to stop reading from source
993 if (!self.push(chunk))
994 self._source.readStop();
997 // When the source ends, we push the EOF-signaling `null` chunk
998 this._source.onend = function() {
1003 // _read will be called when the stream wants to pull more data in
1004 // the advisory size argument is ignored in this case.
1005 SourceWrapper.prototype._read = function(size) {
1006 this._source.readStart();
1011 ### Class: stream.Writable
1015 `stream.Writable` is an abstract class designed to be extended with an
1016 underlying implementation of the [`_write(chunk, encoding, callback)`][] method.
1018 Please see above under [API for Stream Consumers][] for how to consume
1019 writable streams in your programs. What follows is an explanation of
1020 how to implement Writable streams in your programs.
1022 #### new stream.Writable([options])
1024 * `options` {Object}
1025 * `highWaterMark` {Number} Buffer level when [`write()`][] starts
1026 returning false. Default=16kb, or 16 for `objectMode` streams
1027 * `decodeStrings` {Boolean} Whether or not to decode strings into
1028 Buffers before passing them to [`_write()`][]. Default=true
1029 * `objectMode` {Boolean} Whether or not the `write(anyObj)` is
1030 a valid operation. If set you can write arbitrary data instead
1031 of only `Buffer` / `String` data. Default=false
1033 In classes that extend the Writable class, make sure to call the
1034 constructor so that the buffering settings can be properly
1037 #### writable.\_write(chunk, encoding, callback)
1039 * `chunk` {Buffer | String} The chunk to be written. Will **always**
1040 be a buffer unless the `decodeStrings` option was set to `false`.
1041 * `encoding` {String} If the chunk is a string, then this is the
1042 encoding type. If chunk is a buffer, then this is the special
1043 value - 'buffer', ignore it in this case.
1044 * `callback` {Function} Call this function (optionally with an error
1045 argument) when you are done processing the supplied chunk.
1047 All Writable stream implementations must provide a [`_write()`][]
1048 method to send data to the underlying resource.
1050 Note: **This function MUST NOT be called directly.** It should be
1051 implemented by child classes, and called by the internal Writable
1054 Call the callback using the standard `callback(error)` pattern to
1055 signal that the write completed successfully or with an error.
1057 If the `decodeStrings` flag is set in the constructor options, then
1058 `chunk` may be a string rather than a Buffer, and `encoding` will
1059 indicate the sort of string that it is. This is to support
1060 implementations that have an optimized handling for certain string
1061 data encodings. If you do not explicitly set the `decodeStrings`
1062 option to `false`, then you can safely ignore the `encoding` argument,
1063 and assume that `chunk` will always be a Buffer.
1065 This method is prefixed with an underscore because it is internal to
1066 the class that defines it, and should not be called directly by user
1067 programs. However, you **are** expected to override this method in
1068 your own extension classes.
1070 #### writable.\_writev(chunks, callback)
1072 * `chunks` {Array} The chunks to be written. Each chunk has following
1073 format: `{ chunk: ..., encoding: ... }`.
1074 * `callback` {Function} Call this function (optionally with an error
1075 argument) when you are done processing the supplied chunks.
1077 Note: **This function MUST NOT be called directly.** It may be
1078 implemented by child classes, and called by the internal Writable
1081 This function is completely optional to implement. In most cases it is
1082 unnecessary. If implemented, it will be called with all the chunks
1083 that are buffered in the write queue.
1086 ### Class: stream.Duplex
1090 A "duplex" stream is one that is both Readable and Writable, such as a
1091 TCP socket connection.
1093 Note that `stream.Duplex` is an abstract class designed to be extended
1094 with an underlying implementation of the `_read(size)` and
1095 [`_write(chunk, encoding, callback)`][] methods as you would with a
1096 Readable or Writable stream class.
1098 Since JavaScript doesn't have multiple prototypal inheritance, this
1099 class prototypally inherits from Readable, and then parasitically from
1100 Writable. It is thus up to the user to implement both the lowlevel
1101 `_read(n)` method as well as the lowlevel
1102 [`_write(chunk, encoding, callback)`][] method on extension duplex classes.
1104 #### new stream.Duplex(options)
1106 * `options` {Object} Passed to both Writable and Readable
1107 constructors. Also has the following fields:
1108 * `allowHalfOpen` {Boolean} Default=true. If set to `false`, then
1109 the stream will automatically end the readable side when the
1110 writable side ends and vice versa.
1111 * `readableObjectMode` {Boolean} Default=false. Sets `objectMode`
1112 for readable side of the stream. Has no effect if `objectMode`
1114 * `writableObjectMode` {Boolean} Default=false. Sets `objectMode`
1115 for writable side of the stream. Has no effect if `objectMode`
1118 In classes that extend the Duplex class, make sure to call the
1119 constructor so that the buffering settings can be properly
1123 ### Class: stream.Transform
1125 A "transform" stream is a duplex stream where the output is causally
1126 connected in some way to the input, such as a [zlib][] stream or a
1129 There is no requirement that the output be the same size as the input,
1130 the same number of chunks, or arrive at the same time. For example, a
1131 Hash stream will only ever have a single chunk of output which is
1132 provided when the input is ended. A zlib stream will produce output
1133 that is either much smaller or much larger than its input.
1135 Rather than implement the [`_read()`][] and [`_write()`][] methods, Transform
1136 classes must implement the `_transform()` method, and may optionally
1137 also implement the `_flush()` method. (See below.)
1139 #### new stream.Transform([options])
1141 * `options` {Object} Passed to both Writable and Readable
1144 In classes that extend the Transform class, make sure to call the
1145 constructor so that the buffering settings can be properly
1148 #### transform.\_transform(chunk, encoding, callback)
1150 * `chunk` {Buffer | String} The chunk to be transformed. Will **always**
1151 be a buffer unless the `decodeStrings` option was set to `false`.
1152 * `encoding` {String} If the chunk is a string, then this is the
1153 encoding type. If chunk is a buffer, then this is the special
1154 value - 'buffer', ignore it in this case.
1155 * `callback` {Function} Call this function (optionally with an error
1156 argument and data) when you are done processing the supplied chunk.
1158 Note: **This function MUST NOT be called directly.** It should be
1159 implemented by child classes, and called by the internal Transform
1162 All Transform stream implementations must provide a `_transform`
1163 method to accept input and produce output.
1165 `_transform` should do whatever has to be done in this specific
1166 Transform class, to handle the bytes being written, and pass them off
1167 to the readable portion of the interface. Do asynchronous I/O,
1168 process things, and so on.
1170 Call `transform.push(outputChunk)` 0 or more times to generate output
1171 from this input chunk, depending on how much data you want to output
1172 as a result of this chunk.
1174 Call the callback function only when the current chunk is completely
1175 consumed. Note that there may or may not be output as a result of any
1176 particular input chunk. If you supply output as the second argument to the
1177 callback, it will be passed to push method, in other words the following are
1181 transform.prototype._transform = function (data, encoding, callback) {
1186 transform.prototype._transform = function (data, encoding, callback) {
1187 callback(null, data);
1191 This method is prefixed with an underscore because it is internal to
1192 the class that defines it, and should not be called directly by user
1193 programs. However, you **are** expected to override this method in
1194 your own extension classes.
1196 #### transform.\_flush(callback)
1198 * `callback` {Function} Call this function (optionally with an error
1199 argument) when you are done flushing any remaining data.
1201 Note: **This function MUST NOT be called directly.** It MAY be implemented
1202 by child classes, and if so, will be called by the internal Transform
1205 In some cases, your transform operation may need to emit a bit more
1206 data at the end of the stream. For example, a `Zlib` compression
1207 stream will store up some internal state so that it can optimally
1208 compress the output. At the end, however, it needs to do the best it
1209 can with what is left, so that the data will be complete.
1211 In those cases, you can implement a `_flush` method, which will be
1212 called at the very end, after all the written data is consumed, but
1213 before emitting `end` to signal the end of the readable side. Just
1214 like with `_transform`, call `transform.push(chunk)` zero or more
1215 times, as appropriate, and call `callback` when the flush operation is
1218 This method is prefixed with an underscore because it is internal to
1219 the class that defines it, and should not be called directly by user
1220 programs. However, you **are** expected to override this method in
1221 your own extension classes.
1223 #### Events: 'finish' and 'end'
1225 The [`finish`][] and [`end`][] events are from the parent Writable
1226 and Readable classes respectively. The `finish` event is fired after
1227 `.end()` is called and all chunks have been processed by `_transform`,
1228 `end` is fired after all data has been output which is after the callback
1229 in `_flush` has been called.
1231 #### Example: `SimpleProtocol` parser v2
1233 The example above of a simple protocol parser can be implemented
1234 simply by using the higher level [Transform][] stream class, similar to
1235 the `parseHeader` and `SimpleProtocol v1` examples above.
1237 In this example, rather than providing the input as an argument, it
1238 would be piped into the parser, which is a more idiomatic io.js stream
1242 var util = require('util');
1243 var Transform = require('stream').Transform;
1244 util.inherits(SimpleProtocol, Transform);
1246 function SimpleProtocol(options) {
1247 if (!(this instanceof SimpleProtocol))
1248 return new SimpleProtocol(options);
1250 Transform.call(this, options);
1251 this._inBody = false;
1252 this._sawFirstCr = false;
1253 this._rawHeader = [];
1257 SimpleProtocol.prototype._transform = function(chunk, encoding, done) {
1258 if (!this._inBody) {
1259 // check if the chunk has a \n\n
1261 for (var i = 0; i < chunk.length; i++) {
1262 if (chunk[i] === 10) { // '\n'
1263 if (this._sawFirstCr) {
1267 this._sawFirstCr = true;
1270 this._sawFirstCr = false;
1275 // still waiting for the \n\n
1276 // stash the chunk, and try again.
1277 this._rawHeader.push(chunk);
1279 this._inBody = true;
1280 var h = chunk.slice(0, split);
1281 this._rawHeader.push(h);
1282 var header = Buffer.concat(this._rawHeader).toString();
1284 this.header = JSON.parse(header);
1286 this.emit('error', new Error('invalid simple protocol data'));
1289 // and let them know that we are done parsing the header.
1290 this.emit('header', this.header);
1292 // now, because we got some extra data, emit this first.
1293 this.push(chunk.slice(split));
1296 // from there on, just provide the data to our consumer as-is.
1303 // var parser = new SimpleProtocol();
1304 // source.pipe(parser)
1305 // Now parser is a readable stream that will emit 'header'
1306 // with the parsed header data.
1310 ### Class: stream.PassThrough
1312 This is a trivial implementation of a [Transform][] stream that simply
1313 passes the input bytes across to the output. Its purpose is mainly
1314 for examples and testing, but there are occasionally use cases where
1315 it can come in handy as a building block for novel sorts of streams.
1318 ## Simplified Constructor API
1322 In simple cases there is now the added benefit of being able to construct a stream without inheritance.
1324 This can be done by passing the appropriate methods as constructor options:
1330 var readable = new stream.Readable({
1332 // sets this._read under the hood
1339 var writable = new stream.Writable({
1340 write: function(chunk, encoding, next) {
1341 // sets this._write under the hood
1347 var writable = new stream.Writable({
1348 writev: function(chunks, next) {
1349 // sets this._writev under the hood
1356 var duplex = new stream.Duplex({
1358 // sets this._read under the hood
1360 write: function(chunk, encoding, next) {
1361 // sets this._write under the hood
1367 var duplex = new stream.Duplex({
1369 // sets this._read under the hood
1371 writev: function(chunks, next) {
1372 // sets this._writev under the hood
1379 var transform = new stream.Transform({
1380 transform: function(chunk, encoding, next) {
1381 // sets this._transform under the hood
1383 flush: function(done) {
1384 // sets this._flush under the hood
1389 ## Streams: Under the Hood
1397 Both Writable and Readable streams will buffer data on an internal
1398 object called `_writableState.buffer` or `_readableState.buffer`,
1401 The amount of data that will potentially be buffered depends on the
1402 `highWaterMark` option which is passed into the constructor.
1404 Buffering in Readable streams happens when the implementation calls
1405 [`stream.push(chunk)`][]. If the consumer of the Stream does not call
1406 `stream.read()`, then the data will sit in the internal queue until it
1409 Buffering in Writable streams happens when the user calls
1410 [`stream.write(chunk)`][] repeatedly, even when `write()` returns `false`.
1412 The purpose of streams, especially with the `pipe()` method, is to
1413 limit the buffering of data to acceptable levels, so that sources and
1414 destinations of varying speed will not overwhelm the available memory.
1416 ### `stream.read(0)`
1418 There are some cases where you want to trigger a refresh of the
1419 underlying readable stream mechanisms, without actually consuming any
1420 data. In that case, you can call `stream.read(0)`, which will always
1423 If the internal read buffer is below the `highWaterMark`, and the
1424 stream is not currently reading, then calling `read(0)` will trigger
1425 a low-level `_read` call.
1427 There is almost never a need to do this. However, you will see some
1428 cases in io.js's internals where this is done, particularly in the
1429 Readable stream class internals.
1431 ### `stream.push('')`
1433 Pushing a zero-byte string or Buffer (when not in [Object mode][]) has an
1434 interesting side effect. Because it *is* a call to
1435 [`stream.push()`][], it will end the `reading` process. However, it
1436 does *not* add any data to the readable buffer, so there's nothing for
1439 Very rarely, there are cases where you have no data to provide now,
1440 but the consumer of your stream (or, perhaps, another bit of your own
1441 code) will know when to check again, by calling `stream.read(0)`. In
1442 those cases, you *may* call `stream.push('')`.
1444 So far, the only use case for this functionality is in the
1445 [tls.CryptoStream][] class, which is deprecated in io.js v1.0. If you
1446 find that you have to use `stream.push('')`, please consider another
1447 approach, because it almost certainly indicates that something is
1450 ### Compatibility with Older Node.js Versions
1454 In versions of Node.js prior to v0.10, the Readable stream interface was
1455 simpler, but also less powerful and less useful.
1457 * Rather than waiting for you to call the `read()` method, `'data'`
1458 events would start emitting immediately. If you needed to do some
1459 I/O to decide how to handle data, then you had to store the chunks
1460 in some kind of buffer so that they would not be lost.
1461 * The [`pause()`][] method was advisory, rather than guaranteed. This
1462 meant that you still had to be prepared to receive `'data'` events
1463 even when the stream was in a paused state.
1465 In io.js v1.0 and Node.js v0.10, the Readable class described below was added.
1466 For backwards compatibility with older Node.js programs, Readable streams
1467 switch into "flowing mode" when a `'data'` event handler is added, or
1468 when the [`resume()`][] method is called. The effect is that, even if
1469 you are not using the new `read()` method and `'readable'` event, you
1470 no longer have to worry about losing `'data'` chunks.
1472 Most programs will continue to function normally. However, this
1473 introduces an edge case in the following conditions:
1475 * No [`'data'` event][] handler is added.
1476 * The [`resume()`][] method is never called.
1477 * The stream is not piped to any writable destination.
1479 For example, consider the following code:
1483 net.createServer(function(socket) {
1485 // we add an 'end' method, but never consume the data
1486 socket.on('end', function() {
1487 // It will never get here.
1488 socket.end('I got your message (but didnt read it)\n');
1494 In versions of Node.js prior to v0.10, the incoming message data would be
1495 simply discarded. However, in io.js v1.0 and Node.js v0.10 and beyond,
1496 the socket will remain paused forever.
1498 The workaround in this situation is to call the `resume()` method to
1499 start the flow of data:
1503 net.createServer(function(socket) {
1505 socket.on('end', function() {
1506 socket.end('I got your message (but didnt read it)\n');
1509 // start the flow of data, discarding it.
1515 In addition to new Readable streams switching into flowing mode,
1516 pre-v0.10 style streams can be wrapped in a Readable class using the
1524 Normally, Streams operate on Strings and Buffers exclusively.
1526 Streams that are in **object mode** can emit generic JavaScript values
1527 other than Buffers and Strings.
1529 A Readable stream in object mode will always return a single item from
1530 a call to `stream.read(size)`, regardless of what the size argument
1533 A Writable stream in object mode will always ignore the `encoding`
1534 argument to `stream.write(data, encoding)`.
1536 The special value `null` still retains its special value for object
1537 mode streams. That is, for object mode readable streams, `null` as a
1538 return value from `stream.read()` indicates that there is no more
1539 data, and [`stream.push(null)`][] will signal the end of stream data
1542 No streams in io.js core are object mode streams. This pattern is only
1543 used by userland streaming libraries.
1545 You should set `objectMode` in your stream child class constructor on
1546 the options object. Setting `objectMode` mid-stream is not safe.
1548 For Duplex streams `objectMode` can be set exclusively for readable or
1549 writable side with `readableObjectMode` and `writableObjectMode`
1550 respectively. These options can be used to implement parsers and
1551 serializers with Transform streams.
1554 var util = require('util');
1555 var StringDecoder = require('string_decoder').StringDecoder;
1556 var Transform = require('stream').Transform;
1557 util.inherits(JSONParseStream, Transform);
1559 // Gets \n-delimited JSON string data, and emits the parsed objects
1560 function JSONParseStream() {
1561 if (!(this instanceof JSONParseStream))
1562 return new JSONParseStream();
1564 Transform.call(this, { readableObjectMode : true });
1567 this._decoder = new StringDecoder('utf8');
1570 JSONParseStream.prototype._transform = function(chunk, encoding, cb) {
1571 this._buffer += this._decoder.write(chunk);
1572 // split on newlines
1573 var lines = this._buffer.split(/\r?\n/);
1574 // keep the last partial line buffered
1575 this._buffer = lines.pop();
1576 for (var l = 0; l < lines.length; l++) {
1577 var line = lines[l];
1579 var obj = JSON.parse(line);
1581 this.emit('error', er);
1584 // push the parsed object out to the readable consumer
1590 JSONParseStream.prototype._flush = function(cb) {
1591 // Just handle any leftover
1592 var rem = this._buffer.trim();
1595 var obj = JSON.parse(rem);
1597 this.emit('error', er);
1600 // push the parsed object out to the readable consumer
1608 [EventEmitter]: https://iojs.org/dist/v2.3.0/doc/api/events.html#events_class_events_eventemitter
1609 [Object mode]: #stream_object_mode
1610 [`stream.push(chunk)`]: #stream_readable_push_chunk_encoding
1611 [`stream.push(null)`]: #stream_readable_push_chunk_encoding
1612 [`stream.push()`]: #stream_readable_push_chunk_encoding
1613 [`unpipe()`]: #stream_readable_unpipe_destination
1614 [unpiped]: #stream_readable_unpipe_destination
1615 [tcp sockets]: https://iojs.org/dist/v2.3.0/doc/api/net.html#net_class_net_socket
1616 [zlib streams]: zlib.html
1618 [crypto streams]: crypto.html
1619 [crypto]: crypto.html
1620 [tls.CryptoStream]: https://iojs.org/dist/v2.3.0/doc/api/tls.html#tls_class_cryptostream
1621 [process.stdin]: https://iojs.org/dist/v2.3.0/doc/api/process.html#process_process_stdin
1622 [stdout]: https://iojs.org/dist/v2.3.0/doc/api/process.html#process_process_stdout
1623 [process.stdout]: https://iojs.org/dist/v2.3.0/doc/api/process.html#process_process_stdout
1624 [process.stderr]: https://iojs.org/dist/v2.3.0/doc/api/process.html#process_process_stderr
1625 [child process stdout and stderr]: https://iojs.org/dist/v2.3.0/doc/api/child_process.html#child_process_child_stdout
1626 [API for Stream Consumers]: #stream_api_for_stream_consumers
1627 [API for Stream Implementors]: #stream_api_for_stream_implementors
1628 [Readable]: #stream_class_stream_readable
1629 [Writable]: #stream_class_stream_writable
1630 [Duplex]: #stream_class_stream_duplex
1631 [Transform]: #stream_class_stream_transform
1632 [`end`]: #stream_event_end
1633 [`finish`]: #stream_event_finish
1634 [`_read(size)`]: #stream_readable_read_size_1
1635 [`_read()`]: #stream_readable_read_size_1
1636 [_read]: #stream_readable_read_size_1
1637 [`writable.write(chunk)`]: #stream_writable_write_chunk_encoding_callback
1638 [`write(chunk, encoding, callback)`]: #stream_writable_write_chunk_encoding_callback
1639 [`write()`]: #stream_writable_write_chunk_encoding_callback
1640 [`stream.write(chunk)`]: #stream_writable_write_chunk_encoding_callback
1641 [`_write(chunk, encoding, callback)`]: #stream_writable_write_chunk_encoding_callback_1
1642 [`_write()`]: #stream_writable_write_chunk_encoding_callback_1
1643 [_write]: #stream_writable_write_chunk_encoding_callback_1
1644 [`util.inherits`]: https://iojs.org/dist/v2.3.0/doc/api/util.html#util_util_inherits_constructor_superconstructor
1645 [`end()`]: #stream_writable_end_chunk_encoding_callback
1646 [`'data'` event]: #stream_event_data
1647 [`resume()`]: #stream_readable_resume
1648 [`readable.resume()`]: #stream_readable_resume
1649 [`pause()`]: #stream_readable_pause
1650 [`unpipe()`]: #stream_readable_unpipe_destination
1651 [`pipe()`]: #stream_readable_pipe_destination_options