Introduction to Stream in Node.js_node.js

What is flow?

Speaking of streams, it involves a *nix concept: Pipe - In *nix, streams are implemented in the Shell as data that can be bridged through | (pipe character), a The output of a process (stdout) can be directly used as the input (stdin) of the next process.

In Node, the concept of stream (Stream) is similar, representing the ability of a data stream to be bridged.

pipe

The essence of streaming lies in the .pipe() method. The ability to bridge is that both ends of the data stream (upstream/downstream or read/write stream) are bridged with a .pipe() method.

The expression form of pseudo code is:

Copy code The code is as follows:

//upstream.pipe (downstream)
Readable.pipe(Writable);

Classification of streams

This is not intended to discuss the so-called "classic" flow before Node v0.4. Then, streams are divided into several categories (all abstract interfaces:

1.stream.Readable Readable stream (needs to implement the _read method, the focus is on the details of reading the data stream
2.stream.Writable Writable stream (needs to implement the _write method, the focus is on the details of writing the data stream
3.stream.Duplex Read/write stream (needs to implement the above two interfaces, focus on the details of the above two interfaces
4.stream.Transform Inherited from Duplex (needs to implement the _transform method, the focus is on the processing of data blocks

In short:

1) The owner of .pipe() must have Readable stream (but not limited to) capability. It has a series of 'readable'/'data'/'end'/'close'/'error' events for Subscription also provides a series of methods such as .read()/.pause()/.resume() for calling;
2) The parameters of .pipe() must have Writable stream capabilities (but not limited to). It has 'drain'/'pipe'/'unpipe'/'error'/'finish' events for access, and also provides .write ()/.end() and other methods are available for calling

What the hell

Are you feeling the slightest bit anxious? Don't worry, as a low-level coder who speaks human language, I will break Stream apart and talk to you about it.

The

Stream class, in the Node.js source code , is defined as follows:

Copy code The code is as follows:

var EE = require('events').EventEmitter;
var util = require('util');
util.inherits(Stream, EE);

function Stream() {
EE.call(this);
}

As you can see, essentially, Stream is an EventEmitter, which means that it has event-driven functions (.emit/.on...). As we all know, "Node.js is an event-driven platform based on V8", which implements event-driven streaming programming and has the same asynchronous callback characteristics as Node.

For example, in a Readable stream, there is a readable event. In a paused read-only stream, as long as a data block is ready to be read, it will be sent to the subscriber (what are the Readable streams? Express) req, req.part of ftp or mutli-form upload component, standard input process.stdin in the system, etc.). With the readable event, we can make a tool such as a parser that processes shell command output:

Copy code The code is as follows:

process.stdin.on('readable', function(){
var buf = process.stdin.read();
if(buf){
var data = buf.toString();
// parsing data ...                                                   }
});

Call like this:

Copy code The code is as follows:

head -10 some.txt | node parser.js

For a Readable stream, we can also subscribe to its data and end events to get chunks of data and get notified when the stream is exhausted, as in the classic socket example:

Copy code The code is as follows:

req.on('connect', function(res, socket, head) {
socket.on('data', function(chunk) {
console.log(chunk.toString());
});
socket.on('end', function() {
       proxy.close();
});
});

Readable stream status switching
It should be noted that the Readable stream has two states: flowing mode (torrent) and pause mode (pause). The former cannot stop at all, and will continue to feed whoever is piped; the latter will pause until the downstream explicitly calls Stream.read() request to read the data block. The Readable stream is in pause mode when initialized.

These two states can be switched between each other, among which,

If any of the following behaviors occur, pause will change to flowing:

1. Add a data event subscription to the Readable stream
2. Call .resume() on Readable to explicitly enable flowing
3. Call .pipe(writable) of the Readable stream to bridge to a Writable stream

If any of the following behaviors occurs, flowing will return to pause:

1.Readable stream has not been piped to any stream yet, adjustable .pause() can be used to pause
2. The Readable stream has been piped to the stream. You need to remove all data event subscriptions and call the .unpipe() method to release the relationship with the downstream stream one by one

Wonderful Use

Combined with the asynchronous characteristics of the stream, I can write an application like this: directly bridge the output of user A to the output on the page of user B:

Copy code The code is as follows:

router.post('/post', function(req, res) {
var destination = req.headers['destination']; //Who to send to
cache[destionation] = req;
//Yes, it does not return, so it is best to make an ajax request
});

When user B requests:

Copy code The code is as follows:

router.get('/inbox', function(req, res){
var user = req.headers['user'];
cache.find(user, function(err, previousReq){ //Find the previously saved req
      var form = new multiparty.Form();
        form.parse(previousReq); // There are files for me
       form.on('part', function (part) {
              part.pipe(res); //Streaming method is good:)

              part.on('error', function (err) {
console.log(err);
                 messaging.setRequestDone(uniqueID);
                     return res.end(err);
            });
        });
});
});

Reference

how to write node programs with streams: stream-handbook