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

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

Основные этапы:

1. Source (Источник): Читает данные из источника (файл1, сокет2, HTTP-запрос3).
2. Transformations (Преобразования): Последовательно применяет операции обработки к данным. Каждый этап получает данные от предыдущего, обрабатывает их и передает дальше.
3. Destination (Назначение): Записывает обработанные данные в конечный узел.

Пример использования pipeline в Node.js:

const { createReadStream, createWriteStream } = require('node:fs');
const { Transform, pipeline } = require('node:stream');

const sourceStream = createReadStream('input.txt');
const destinationStream = createWriteStream('output.txt');

// Преобразование: добавляем префикс к каждой строке
const transformStream = new Transform({
  transform(chunk, encoding, callback) {
    this.push(`Префикс: ${chunk}`);
    callback();
  }
});

pipeline(
  sourceStream, // Источник
  transformStream, // Преобразование
  destinationStream, // Назначение
  (err) => { // Коллбэк по завершению
    if (err) {
      console.error('Pipeline failed.', err);
    } else {
      console.log('Pipeline succeeded.');
    }
  }
);

Преимущества pipeline:

• Эффективность памяти: Обрабатывает данные порциями, не загружая все в память одновременно.
• Композиция: Легко объединять несколько этапов обработки.
• Обработка ошибок: Встроенный механизм обработки ошибок, который распространяется по всем потокам.
• Протоколирование: Позволяет подключать обработчики событий для мониторинга процесса.