How is data processing organized in a pipeline?

The data processing process in a pipeline (for example, in the context of Node.js streams) is based on the principles of chaining and redirecting.

1. Sources: It starts with a source stream (readable stream) that generates data.
2. Transforms: Data from the source can be passed through one or more intermediate transform streams. Each transformer performs some operation on the data (e.g., filtering, format conversion).
3. Destinations: The processed data is sent to a destination stream (writable stream), which consumes it (e.g., writes to a file, sends over the network).

pipe() Method:

The central method for building a pipeline is pipe(). It connects the output of one stream to the input of another:

// sourceStream - Readable stream
// transformStream1 - Transform stream
// transformStream2 - Transform stream
// destinationStream - Writable stream

sourceStream.pipe(transformStream1).pipe(transformStream2).pipe(destinationStream);

Event Flow:

• When the source is ready to emit data, it triggers the data event, passing a chunk of data.
• This chunk is passed to the next stream in the pipeline (or directly to the destination if there are no transformers).
• Intermediate streams process the chunk and, if ready, trigger their own data event with transformed data.
• This process repeats until reaching the destination stream.
• When the source stops generating data, it triggers the end event, signaling the end of the stream. This event also propagates through the pipeline, and the last destination stream also triggers end.

Error Handling:

Errors in any stream of the pipeline can interrupt its execution. It is important to handle the error event on each stream or use a library to manage pipelines that simplifies error handling.

sourceStream.on('error', handleError);
transformStream1.on('error', handleError);
transformStream2.on('error', handleError);
destinationStream.on('error', handleError);

Example:

const { createReadStream, createWriteStream } = require('fs');
const { createGzip } = require('zlib');

const readable = createReadStream('input.txt');
const writable = createWriteStream('output.txt.gz');
const gzip = createGzip();

// Read from 'input.txt', compress, and write to 'output.txt.gz'
readable.pipe(gzip).pipe(writable);

readable.on('error', (err) => console.error('Read error:', err));
gzip.on('error', (err) => console.error('Compression error:', err));
writable.on('error', (err) => console.error('Write error:', err));

writable.on('finish', () => console.log('File successfully compressed.'));