Node js pipe error

Streams, Piping, and Their Error Handling in Node.js

Explained with examples

In the last article, we learned the basics of streams and buffers in Node.js. We also learned about readable streams in Node.js.

If you haven’t read the last article on streams and buffers, read it here:

Streams and Buffers in NodeJs

To handle and manipulate streaming data like a video, a large file, etc. we need streams in Node. Streams module in…

In this article, we will learn about the writable streams, streams piping, and the events and the functions available on a writable stream in Node.js.

A Writable Stream

A writable stream is a stream of data that is created to write some streaming data. For example, creating a write stream to write a text file for some streaming data.

Let’s consider the following example for creating a writable stream in Node.js.

I am using TypeScript instead of JavaScript.

In the above code, we created a write stream to write some streaming data to a file named dump.txt .

Running the above two lines creates a file with the named dump.txt but without any data inside it.

To actually write some data to the file, we need to call the write function of the write stream.

Running the following two write functions on the writeStream variable creates a file named dump.txt and inserts the following text into it.

Stream Piping and Unpiping

In an Express application, the req (request) and res (response) for a request handler are streams. A req is a readable stream of data whereas res is a writable stream of data.

Considering this for an HTTP request, we have to serve a really large file, we can do so by using streams.

The following code in Node represents a similar functionality:

When we request the browser to http://localhost:3000, the request handler is triggered. The handler creates a readStream for file data.txt .

For the data event of readStream , we have called the write method of the res writeStream . Also, on the end event of the readStream , we send a 200 OK status code to the client.

All this code works perfectly and solves the problem well. But there is a shortcut to this problem.

Instead of listening to the data and the end events, we can simply pipe these two streams.

With piping, we simply mean that the two streams are connected. The data that is passed to stream 1 is also passed through stream 2 which is piped to stream 1.

A shorter implementation for the get method using piping is:

With stream piping, the code size is reduced to only one line of code.

In the above code, we piped the read stream from the file to the write stream of the response. This simply means the streaming data from the readStream will be piped and passed through the res write stream.

Similar to the pipe function, there is also an unpipe function on a stream. We can simply call source.unpipe(destination) anytime to stop passing of the data from the source stream to the destination stream.

In the code, we unpiped the stream after 10 milliseconds. Only the data from the read stream that was piped to the res write stream will be sent to the client.

After 10 ms , the res stream will be unpiped from the readStream and we send the 200 status to the client.

As a result of running this code, we do not get the complete content of the data.txt file in the browser.

In my case, the data.txt was 16000 lines of text, but I received only approx. 3800 lines of text on the client size as the stream was unpiped in between sending it to the client.

Error Handling for Normal and Piped Streams

In streams, we handle the errors by creating an error event listener on the stream. The listener gets triggered as soon as an error comes up in the stream.

For the case of error handling in piped streams, let us consider the following code snippet:

For an HTTP request, we created a read stream for a file and we piped it to the HTTP response write stream.

In the code, we closed the read stream in between streaming and this resulted in triggering an error on the write stream for reading from a stream that does not exist anymore.

Running the above code and creating an HTTP request to the endpoint generates the following output:

If we are having a series of piped streams in Node.js, we have to do the error handling for each of the streams individually.

In Node, a pipe does not forward error to the next pipe.

To handle errors in the above case of piped streams, we have to add an error handler on each of the streams like this:

As a result, if any of the streams encounters an error, its corresponding error handler will be triggered and the process will not exit due to unhandled errors.

Events Available on a Writable Stream

The following are the events we can listen for on a writable stream.

drain

Consider we are having a scenario where the stream buffer is full and we want to know when the buffer has some space to continue writing. In such a scenario, we listen to the drain event of the stream.

The drain event triggers as soon as it will appropriate for the stream to resume writing the data.

close

The event is triggered as soon as the stream is closed using the stream.close() function.

finish

The event is triggered after the stream has completed streaming.

pipe/unpipe

The events are triggered as soon as the stream is piped or unpiped by a stream.

Running the following code will give the output as:

Functions Available on a Write Stream

cork/uncork

Consider a scenario where the data flows at a really slow speed and we want some data to get buffered in the stream before using it. We can do so using the cork method on a writable stream.

By calling the cork method, the stream will not write data to the destination and will hold the data in the buffer.

To get the buffer data flushed to the destination, we need to call the uncork method.

In the above code, we corked the stream. As a result, the stream will stop the data flow until it is uncorked.

When uncorked in the nextTick ’s callback, we get 1 2 3 in the buffer of the writable stream being passed to the destination.

Note: If we call the cork() function twice to cork a stream, we have to call the uncork() function twice to enable the data flow from the stream.

Источник

How Error Events Affect Piped Streams In Node.js

Over the weekend, I demonstrated that «error» events don’t have any inherent affect on how individual Streams work in Node.js. In that post, I stressed that I was talking about «individual» streams because multi-stream workflows, that use .pipe(), are somewhat affected by «error» events. The «error» still doesn’t affect the individual streams; but, Node.js will unpipe the streams depending on the source of the error.

If you .pipe() one stream into another, error events emitted from the source stream have no bearing on the workflow (unless handled explicitly by the developer). The only error events that have any affect are those emitted by the target / destination stream. If the target stream emits an error, the source stream will disconnect from it (ie, .unpipe() itself from the target stream).

That said, as we’ve seen before, other than the unpipe response, the error event has no bearing on either stream. This means that after the streams are disconnected, they continue to function normally on their own. To see this in action, I’m going to take one source stream and pipe it into two different target streams. One of the target streams will emit an error event which will cause an .unpipe(); but, the code will demonstrate that all three streams (source + 2 targets) still work like healthy streams.

Since the code is all event-driven, it’s a little hard to follow — I suggest you watch the video. But, when we run the above code, we get the following terminal output:

bens-imac:pipe ben$ node test.js
Source error: StreamError
Unpiped source: true
Target error: StreamError (What)
Target error: StreamError (Written after pipe-break.)
Target error: StreamError (Ended.)
Unsafe Target Buffer: What Written after pipe-break. Ended.
Safe Target Buffer: it be like?

Let’s try to break this down, line by line, so we can see how the error events are affecting the individual streams as well as the stream interactions.

Source error: StreamError

This is the source stream emitting an error when populating the underlying stream buffer. This has no affect at all — not on the source stream and not on the stream pipes.

Unpiped source: true

This is the source stream reacting to the error event in the unsafeTarget .write() method. It [the source] is unpiping itself from the unsafeTarget. However, the first chunk of data was still written to the unsafeTarget buffer since the error was emitted as part of the write-action.

Target error: StreamError (What)

This is that first chunk getting written to the unsafeTarget buffer, before the .unpipe() call has any affect.

Target error: StreamError (Written after pipe-break.)
Target error: StreamError (Ended.)

Inside the «unpipe» event on the unsafeTarget, we make two more explicit writes to the unsafeTarget stream. This is to demonstrate that the unsafeTarget stream continues to function properly even after the «error» event and the «unpipe» event.

Unsafe Target Buffer: What Written after pipe-break. Ended.
Safe Target Buffer: it be like?

This is the «debug» event on both target streams that outputs the aggregated buffer. As you can see, the source continued to pipe data into the safeTarget even after it [source] was disconnected from the unsafeTarget. Furthermore, the unsafeTarget was able to continue accepting writes after being disconnected from the source.

What we’re seeing here is that «error» events, in Node.js, will disconnect piped-streams. However, the streams in question will continue to work properly. Or perhaps more importantly, will remain open. This means that after streams are unpiped, you probably have to end them explicitly in your error handling.

Want to use code from this post? Check out the license.

Источник

Working with Node.js streams

September 14, 2021 4 min read 1182

Introduction

Streams are one of the major features that most Node.js applications rely on, especially when handling HTTP requests, reading/writing files, and making socket communications. Streams are very predictable since we can always expect data, error, and end events when using streams.

This article will teach Node developers how to use streams to efficiently handle large amounts of data. This is a typical real-world challenge faced by Node developers when they have to deal with a large data source, and it may not be feasible to process this data all at once.

This article will cover the following topics:

Types of streams

The following are four main types of streams in Node.js:

• Readable streams: The readable stream is responsible for reading data from a source file
• Writable streams: The writable stream is responsible for writing data in specific formats to files
• Duplex streams: Duplex streams are streams that implement both readable and writable stream interfaces
• Transform streams: The transform stream is a type of duplex stream that reads data, transforms the data, and then writes the transformed data in a specified format

When to use Node.js streams

Streams come in handy when we are working with files that are too large to read into memory and process as a whole.

For example, consider Node.js streams a good choice if you are working on a video conference/streaming application that would require the transfer of data in smaller chunks to enable high-volume web streaming while avoiding network latency.

The batching process

Batching is a common pattern for data optimization which involves the collection of data in chunks, storing these data in memory, and writing them to disk once all the data are stored in memory.

Let’s take a look at a typical batching process:

Here, all of the data is pushed into an array. When the data event is triggered and once the “end” event is triggered, indicating that we are done receiving the data, we proceed to write the data to a file using the fs.writeFile and Buffer.concat methods.

The major downside with batching is insufficient memory allocation because all the data is stored in memory before writing to disk.

Writing data as we receive it is a more efficient approach to handling large files. This is where streams come in handy.

Composing streams in Node.js

The Node.js fs module exposes some of the native Node Stream API, which can be used to compose streams.

We’ll be covering readable, writable, and transform streams. You can read our blog post about duplex streams in Node.js if you want to learn more about them.

Composing writable streams

A writeable stream is created using the createWriteStream() method, which requires the path of the file to write to as a parameter.
Running the above snippet will create a file named file.txt in your current directory with 20,000 lines of Hello world welcome to Node.js in it.

Composing readable streams

Here, the data event handler will execute each time a chunk of data has been read, while the end event handler will execute once there is no more data.
Running the above snippet will log 20,000 lines of the Hello world welcome to Node.js string from ./file.txt to the console.

Over 200k developers use LogRocket to create better digital experiences

Composing transform streams

Transform streams have both readable and writable features. It allows the processing of input data followed by outputting data in the processed format.

To create a transform stream, we need to import the Transform class from the Node.js stream module. The transform stream constructor accepts a function containing the data processing/transformation logic:

Here, we create a new transform stream containing a function that expects three arguments: the first being the chunk of data, the second is encoding (which comes in handy if the chunk is a string), followed by a callback which gets called with the transformed results.

Running the above snippet will transform all the text in ./file.txt to uppercase then write it to transformedData.txt .
If we run this script and we open the resulting file, we’ll see that all the text has been transformed to uppercase.

Piping streams

Piping streams is a vital technique used to connect multiple streams together. It comes in handy when we need to break down complex processing into smaller tasks and execute them sequentially. Node.js provides a native pipe method for this purpose:

Refer to the code snippet under Composing transform streams for more detail on the above snippet.

Error handling Node.js streams

Error handling using pipelines

Node 10 introduced the Pipeline API to enhance error handling with Node.js streams. The pipeline method accepts any number of streams followed by a callback function that handles any errors in our pipeline and will be executed once the pipeline has been completed:

When using pipeline , the series of streams should be passed sequentially in the order in which they need to be executed.

Error handling using pipes

We can also handle stream errors using pipes as follows:

As seen in the above snippet, we have to create an error event handler for each pipe created. With this, we can keep track of the context for errors, which becomes useful when debugging. The drawback with this technique is its verbosity.

Conclusion

In this article, we’ve explored Node.js streams, when to use them, and how to implement them.

Knowledge of Node.js streams is essential because they are a great tool to rely on when handling large sets of data. Check out the Node.js API docs for more information about streams.

200’s only Monitor failed and slow network requests in production

LogRocket is like a DVR for web and mobile apps, recording literally everything that happens while a user interacts with your app. Instead of guessing why problems happen, you can aggregate and report on problematic network requests to quickly understand the root cause.

LogRocket instruments your app to record baseline performance timings such as page load time, time to first byte, slow network requests, and also logs Redux, NgRx, and Vuex actions/state. Start monitoring for free.

Источник

Have you been using Node.js for a while and only until now have you heard about the word pipe? You tried checking the documentation but still can’t figure out what it means or does? In this article, I will clarify those doubts by explaining what .pipe or piping is and how to use it in Node.js. To ensure you understand the article, previous knowledge about streams is strongly recommended.

What Does .pipe() Method Do?

The method .pipe was added in v0.9.4 of Node.js and its purpose is to attach a writeable stream to a readable stream allowing to pass the readable stream data to the writeable stream. One good way to understand this concept is by thinking about PVC pipes and connecting two pipes.

For the sake of explanation, let’s assume the first PVC pipe is a readable stream and the second pipe is a writeable stream. The method .pipe will be the orange pipe fitting which will connect both pipes allow for the water, or the data, to flow from one pipe to another.

How to Use The .pipe() Method?

In this pipe implementation, we are going to create a simple HTTP server that will read data from a file and send the response to the client.

1. Let’s start by creating the HTTP server using the http package that returns some data.

const http = require('http');

http.createServer(function(req, res) {
  res.write('hello!');
  res.end();
}).listen(8080);

Let’s make sure it works, by making a request to our server using curl.

curl localhost:8080

Or another option is to open a new tab http://localhost:8080/. Once you make the request, you should receive “hello!”.

We are going to pause for a second. Let’s recall the anatomy of an HTTP transaction. An HTTP transaction is made of a server, created by the method createServer which in itself is an EventEmitter. When an HTTP request hits the server, node calls the request handler using the req and res objects, which are request and response respectively, for dealing with the transaction.

The req or request object is an instance of the IncomingMessage object. The IncomingMessage object is a child object of a ReadableStream.

The res or response object is an instance of the ServerResponse object. The ServerResponse object is a child object of a WriteableStream.

Therefore, we know we have a writeable and a readable stream.

2. We are going to create a data.txt file in the same directory folder, and save some information. For the sake of making things clear, I will save the following text: “This is data from the data.txt file”.

3. Remove the existing logic from the event handler.

4. We are going to read the content of the data.txt file using the fs package using fs.createReadStream. The fs.createReadStream will return a ReadableStream. We are going to use that ReadableStream to pipe or pass the data from data.txt file to the response object, which is a WriteableStream.

const http = require('http');
const fs = require('fs');

http.createServer(function(req, res) {
  // generete readable stream to read content of data.txt
  const readStream = fs.createReadStream(__dirname + '/data.txt');
  
  // pass readable stream data, which are the content of data.txt, to the 
  // response object, which is a writeable stream
  readStream.pipe(res);
}).listen(8080);

Once updated the event handler’s logic, make a request to http://localhost:8080/ and you should see data.txt data.

Only Works With Readable Streams

Remember, the pipe method can only be used in readable streams. Don’t let yourself fool by your IDE in case it suggests the pipe method in a writeable stream.

In case you attempt using the .pipe method using a writeable stream, like in the example below:

At the moment of executing this code, it will throw the following error.

Error [ERR_STREAM_CANNOT_PIPE]: Cannot pipe, not readable

Pushing Data No Matter Readable Stream’s Flowing Mode

If you are familiar with readable streams, you will know there are two modes in which data flows, flowing and paused mode. You can use the pause() or resume() method to update the flowing mode.

const http = require('http');
const fs = require('fs');

http.createServer(function(req, res) {
  const readStream = fs.createReadStream(__dirname + '/data.txt');
  
  readStream.on('data', function(chunk) {
    console.log('this is the data from file', chunk);
  });

  readStream.pause();
  console.log('on pause: readable flowing', readStream.readableFlowing);

  readStream.resume();
  console.log('on resume: readable flowing', readStream.readableFlowing);

  res.write('Hello!')
  res.end();
}).listen(8080);

If you run the example above, you will only read data from the data.txt file whenever the readable stream flowing mode is set to true which is enabled by using the resume() method. If the flowing mode is set to false, it will never read the content of the data.txt file.

However, when using the pipe method the flowing mode will automatically be set to true ensuring the data is passed from one stream to another. We can confirm this if we try to pause the flowing mode prior to piping both streams.

const http = require('http');
const fs = require('fs');

http.createServer(function(req, res) {
  // generete readable stream to read content of data.txt
  const readStream = fs.createReadStream(__dirname + '/data.txt');
  
  readStream.on('data', function(chunk) {
    console.log('this is the data from file', chunk);
  });

  readStream.pause();
  console.log('on pause: readable flowing', readStream.readableFlowing);

  readStream.pipe(res);

}).listen(8080);

After making a request to the server, we will still receive the content from the data.txt file.

Don’t Confuse The pipe Method With The Event pipe

If you have never heard of the word “pipe” or “piping” when working with streams, there is a slight chance you could find the wrong information if you go to Node.js documentation and start finding for the word “pipe”. When you do a quick search, you will find two options.

1. Event “pipe”
2. Readable.pipe

If you find the first option, you will notice it is an event listener that writeable streams can set when a readable stream uses the pipe method to pass the data from one stream to another. The event pipe is only available on writeable streams. We are going to use our simple server API to demonstrate the event pipes.

const http = require('http');
const fs = require('fs');

http.createServer(function(req, res) {
  const readStream = fs.createReadStream(__dirname + '/data.txt');

  // setting pipe event listener before triggering the pipe method in the readable stream
  // otherwise, the pipe event listener won't be triggered if set after triggering the pipe method
  res.on('pipe', function(src) {
    console.log('Triggered the pipe event listener whenever a source readable stream pipes the writeable stream');
  });

  readStream.pipe(res);

}).listen(8080);

In other words, calling the pipe method on the readable stream causes the pipe event listener to be triggered on the writeable stream.

It is important to mention to define the pipe event listener prior to calling the pipe method from the readable stream. Attempting to call the pipe method prior to setting the event listener in the writeable stream won’t work.

Why You Should Use .pipe Method?

Streams are one of the most powerful and fundamental concepts of Node.js applications. They allow us to handle data in a more efficient way as pieces of data can be transported in smaller chunks preventing you from running out of memory and maintaining good performance in your applications.

Therefore, using the pipe method is an effective and easy solution to push data between streams. In that way, we avoid storing too much data that needs to be manipulated or modified all at the same time. Also, the code will be shorter, elegant, and easy to follow.

Was this article helpful?

Share your thoughts by replying on Twitter of Become A Better Programmer or to personal my Twitter account.