Как устроен процесс обработки данных в pipeline?

Процесс обработки данных в pipeline (например, в контексте потоков Node.js) основан на принципах чейннинга и перенаправления.

1. Источники (Sources): Начинается с потока-источника (readable stream), который генерирует данные.
2. Преобразователи (Transforms): Данные из источника могут передаваться через один или несколько промежуточных потоков-преобразователей (transform streams). Каждый преобразователь выполняет какую-то операцию над данными (например, фильтрация, преобразование формата).
3. Назначения (Destinations): Обработанные данные поступают в поток-назначение (writable stream), который их потребляет (например, записывает в файл, отправляет по сети).

Метод pipe():

Центральным методом для построения pipeline является pipe(). Он связывает выход одного потока с входом другого:

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

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

Поток событий:

• Когда источник готов выдать данные, он инициирует событие data, передавая чанк данных.
• Этот чанк передается в следующий поток в pipeline (или напрямую в назначение, если преобразователей нет).
• Промежуточные потоки обрабатывают чанк и, если готовы, инициируют свое событие data с преобразованными данными.
• Этот процесс повторяется до достижения потока-назначения.
• Когда источник перестает генерировать данные, он инициирует событие end, сигнализируя об окончании потока. Это событие также проходит по pipeline, и последний поток-назначение также инициирует end.

Обработка ошибок:

Ошибки в любом потоке pipeline могут прервать его выполнение. Важно обрабатывать событие error на каждом потоке или использовать библиотеку для управления pipeline'ами, которая упрощает обработку ошибок.

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

Пример:

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

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

// Читаем из 'input.txt', сжимаем и записываем в 'output.txt.gz'
readable.pipe(gzip).pipe(writable);

readable.on('error', (err) => console.error('Ошибка чтения:', err));
gzip.on('error', (err) => console.error('Ошибка сжатия:', err));
writable.on('error', (err) => console.error('Ошибка записи:', err));

writable.on('finish', () => console.log('Файл успешно сжат.'));