Asynchronous Loops and Stream APIs in Node.js 10

This month comes the tenth version of Node.js, in which we are waiting for a change in the behavior of streams ( readable-stream ), caused by the appearance of asynchronous for-await-of loops. Let's see what it is and what we need to prepare for.

For-await-of construction.

To begin with, let's understand how asynchronous loops work with a simple example. For clarity, add completed promises.

const promises = [ Promise.resolve(1), Promise.resolve(2), Promise.resolve(3), ]; 

A normal loop will go through the promises array and return the values ​​themselves:

 for (const value of promises) { console.log(value); } // > Promise({resolved: 1}) // > Promise({resolved: 2}) // > Promise({resolved: 3}) 

The asynchronous cycle will wait for the permission of the promise and return the value returned by the promis:

 for await (const value of promises) { console.log(value); } // > 1 // > 2 // > 3 

To make asynchronous loops work in earlier versions of Node.js, use the --harmony_async_iteration flag.

ReadableStream and for-await-of

The ReadableStream object received the Symbol.asyncIterator property, which allows it to also be passed to the for-await-of loop. Take fs.createReadableStream for example:

 const readStream = fs.createReadStream(file); const chunks = []; for await (const chunk of readStream) { chunks.push(chunk); } console.log(Buffer.concat(chunks)); 

As you can see from the example, now we got rid of the on('data', ... and on('end', ... calls, and the code itself began to look clearer and more predictable.

Asynchronous generators

In some cases, you may need additional processing of the received data, for this purpose asynchronous generators are used. We can implement a search for a regular expression file:

 async function * search(needle, chunks) { let pos = 0; for await (const chunk of chunks) { let string = chunk.toString(); while (string.length) { const match = string.match(needle); if (! match) { pos += string.length; break; } yield { index: pos + match.index, value: match[0], }; string = string.slice(match.index + match[0].length); pos += match.index; } } } 

Let's see what happened:

 const stream = fs.createReadStream(file); for await (const {index, value} of search(/(a|b)c/, stream)) { console.log('found "%s" at %s', value, index); } 

You have to agree that it’s quite convenient, we turned the lines into objects on the fly and we didn’t need to use the TransformStream and think about how to intercept errors that can occur in two different streams, etc.

Unix-like thread example

The task of reading a file is quite common, but not exhaustive. Let's take a look at when streaming output is required, like unix pipelines. To do this, we use asynchronous generators, through which we will pass the result of the ls .

First we will create a child process const subproc = spawn('ls') and then we will read the standard output:

 for await (const chunk of subproc.stdout) { // ... } 

And since stdout generates output as Buffer objects, the first thing to do is add a generator that will output the output from the Buffer type to String:

 async function *toString(chunks) { for await (const chunk of chunks) { yield chunk.toString(); } } 

Next, we make a simple generator that will break the output line by line. It is important to note here that the portion of data transferred from createReadStream has a limited maximum length, which means that we can receive either a whole string or a piece of a very long string, or several rows at a time:

 async function *chunksToLines(chunks) { let previous = ''; for await (const chunk of chunks) { previous += chunk; while (true) { const i = previous.indexOf('\n'); if (i < 0) { break; } yield previous.slice(0, i + 1); previous = previous.slice(i + 1); } } if (previous.length > 0) { yield previous; } } 

Since each found value still contains a line break, create a generator to clear the value of hanging whitespace characters:

 async function *trim(values) { for await (const value of values) { yield value.trim(); } } 

The final action will be direct line output to the console:

 async function print(values) { for await (const value of values) { console.log(value); } } 

Combine the resulting code:

 async function main() { const subproc = spawn('ls'); await print(trim(chunksToLines(toString(subproc.stdout)))); console.log('DONE'); } 

As you can see, the code is somewhat difficult to read. If we want to add a few more calls or parameters, we end up with a mess. To avoid and make the code more linear, let's add the pipe function:

 function pipe(value, ...fns) { let result = value; for (const fn of fns) { result = fn(result); } return result; } 

Now the call can be brought to the following form:

 async function main() { const subproc = spawn('ls'); await pipe( subproc.stdout, toString, chunksToLines, trim, print, ); console.log('DONE'); } 

Operator |>

It should be borne in mind that soon the JS standard should include the new pipeline operator |> allowing you to do the same thing that the pipe is doing now:

 async function main() { const subproc = spawn('ls'); await subproc.stdout |> toString |> chunksToLines |> trim |> print; console.log('DONE'); } 

Conclusion

As you can see, asynchronous loops and iterators made the language even more expressive, capacious and understandable. Hell from kollbekov goes farther into the past and will soon become a scarecrow, which we will scare grandchildren. And the generators seem to take their place in the JS hierarchy and will be used for their intended purpose.

The basis for this article was Axel Rauschmeier’s material. Using async iteration natively in Node.js.

Continuing the theme

• Iterables and Iterators [EN].
• Asynchronous iteration [en].