Структура этого документа

Этот документ состоит из двух основных разделов и третьего раздела с примечаниями. В первом разделе объясняется, как использовать существующие потоки в приложении. Во втором разделе объясняется, как создавать потоки новых типов.

Типы потоков

В Node.js есть четыре основных типа потоков:

• Writable: потоки, в которые можно записывать данные (например, fs.createWriteStream()).
• Readable: потоки, из которых можно читать данные (например, fs.createReadStream()).
• Duplex: потоки, которые одновременно являются Readable и Writable (например, net.Socket).
• Transform: Duplex потоки, которые могут изменять или преобразовывать данные при записи и чтении (например, zlib.createDeflate()).

Кроме того, этот модуль включает вспомогательные функции stream.duplexPair(), stream.pipeline(), stream.finished(), stream.Readable.from() и stream.addAbortSignal().

API потоков на основе Promise

API stream/promises предоставляет альтернативный набор асинхронных вспомогательных функций для потоков, которые возвращают объекты Promise вместо использования обратных вызовов. Доступ к API осуществляется через require('node:stream/promises') или require('node:stream').promises.

stream.pipeline(source[, ...transforms], destination[, options])

stream.pipeline(streams[, options])

• streams <Stream[]> | <Iterable[]> | <AsyncIterable[]> | <Function[]>
• source <Stream> | <Iterable> | <AsyncIterable> | <Function>
  • Возвращает: <Promise> | <AsyncIterable>
• ...transforms <Stream> | <Function>
  • source <AsyncIterable>
  • Возвращает: <Promise> | <AsyncIterable>
• destination <Stream> | <Function>
  • source <AsyncIterable>
  • Возвращает: <Promise> | <AsyncIterable>
• options <Object> Параметры конвейера
  • signal <AbortSignal>
  • end <boolean> Завершать поток назначения после завершения потока источника. Потоки преобразования завершаются всегда, даже если этому значению задано false. По умолчанию: true.
• Возвращает: <Promise> Выполняется после завершения конвейера.

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

await pipeline(
  createReadStream('archive.tar'),
  createGzip(),
  createWriteStream('archive.tar.gz'),
);
console.log('Pipeline succeeded.');

Чтобы использовать AbortSignal, передайте его последним аргументом внутри объекта параметров. При прерывании сигнала для базового конвейера будет вызван destroy с объектом AbortError.

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  const ac = new AbortController();
  const signal = ac.signal;

  setImmediate(() => ac.abort());
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
    { signal },
  );
}

run().catch(console.error); // AbortError

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

const ac = new AbortController();
const { signal } = ac;
setImmediate(() => ac.abort());
try {
  await pipeline(
    createReadStream('archive.tar'),
    createGzip(),
    createWriteStream('archive.tar.gz'),
    { signal },
  );
} catch (err) {
  console.error(err); // AbortError
}

API pipeline также поддерживает асинхронные генераторы:

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
  await pipeline(
    fs.createReadStream('lowercase.txt'),
    async function* (source, { signal }) {
      source.setEncoding('utf8');  // Work with strings rather than `Buffer`s.
      for await (const chunk of source) {
        yield await processChunk(chunk, { signal });
      }
    },
    fs.createWriteStream('uppercase.txt'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';

await pipeline(
  createReadStream('lowercase.txt'),
  async function* (source, { signal }) {
    source.setEncoding('utf8');  // Work with strings rather than `Buffer`s.
    for await (const chunk of source) {
      yield await processChunk(chunk, { signal });
    }
  },
  createWriteStream('uppercase.txt'),
);
console.log('Pipeline succeeded.');

Не забудьте обработать аргумент signal, переданный асинхронному генератору. Это особенно важно, если асинхронный генератор является источником конвейера (то есть первым аргументом), иначе конвейер никогда не завершится.

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
  await pipeline(
    async function* ({ signal }) {
      await someLongRunningfn({ signal });
      yield 'asd';
    },
    fs.createWriteStream('uppercase.txt'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);

import { pipeline } from 'node:stream/promises';
import fs from 'node:fs';
await pipeline(
  async function* ({ signal }) {
    await someLongRunningfn({ signal });
    yield 'asd';
  },
  fs.createWriteStream('uppercase.txt'),
);
console.log('Pipeline succeeded.');

API pipeline предоставляет версию с обратным вызовом:

stream.finished(stream[, options])

• stream <Stream> | <ReadableStream> | <WritableStream> Читаемый и/или записываемый поток/веб-поток.
• options <Object>
  • error <boolean> | <undefined>
  • readable <boolean> | <undefined>
  • writable <boolean> | <undefined>
  • signal: <AbortSignal> | <undefined>
• Возвращает: <Promise> Выполняется, когда поток становится нечитаемым или недоступным для записи.

const { finished } = require('node:stream/promises');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

async function run() {
  await finished(rs);
  console.log('Stream is done reading.');
}

run().catch(console.error);
rs.resume(); // Drain the stream.

import { finished } from 'node:stream/promises';
import { createReadStream } from 'node:fs';

const rs = createReadStream('archive.tar');

async function run() {
  await finished(rs);
  console.log('Stream is done reading.');
}

run().catch(console.error);
rs.resume(); // Drain the stream.

API finished также предоставляет версию с обратным вызовом.

Режим объектов

Все потоки, созданные API Node.js, работают исключительно со строками, объектами <Buffer>, <TypedArray> и <DataView>:

• Strings и Buffers — наиболее распространённые типы, используемые в потоках.
• TypedArray и DataView позволяют обрабатывать двоичные данные с помощью таких типов, как Int32Array или Uint8Array. При записи TypedArray или DataView в поток Node.js обрабатывает исходные байты.

Однако реализации потоков могут работать и с другими значениями JavaScript (за исключением null, которое выполняет в потоках особую функцию). Такие потоки считаются работающими в «режиме объектов».

Экземпляры потоков переводятся в режим объектов с помощью параметра objectMode при создании потока. Переключать уже существующий поток в режим объектов небезопасно.

Буферизация

Потоки Writable и Readable хранят данные во внутреннем буфере.

Количество данных, которое может быть буферизовано, зависит от параметра highWaterMark, переданного конструктору потока. Для обычных потоков параметр highWaterMark задаёт общее количество байтов. Для потоков, работающих в режиме объектов, параметр highWaterMark задаёт общее количество объектов. Для потоков, работающих со строками (но не декодирующих их), параметр highWaterMark задаёт общее количество кодовых единиц UTF-16.

Данные буферизуются в потоках Readable, когда реализация вызывает stream.push(chunk). Если потребитель потока не вызывает stream.read(), данные будут оставаться во внутренней очереди, пока их не прочитают.

Когда общий размер внутреннего буфера чтения достигает порогового значения, заданного параметром highWaterMark, поток временно прекращает чтение данных из базового ресурса, пока уже буферизованные данные не будут обработаны (то есть поток перестаёт вызывать внутренний метод readable._read(), который используется для заполнения буфера чтения).

Данные буферизуются в потоках Writable при многократном вызове метода writable.write(chunk). Пока общий размер внутреннего буфера записи меньше порогового значения, заданного параметром highWaterMark, вызовы writable.write() возвращают true. Когда размер внутреннего буфера достигает или превышает highWaterMark, возвращается false.

Одна из ключевых целей API stream, в частности метода stream.pipe(), — ограничить буферизацию данных приемлемым уровнем, чтобы источники и назначения с разной скоростью не перегружали доступную память.

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

Поскольку потоки Duplex и Transform одновременно являются Readable и Writable, каждый из них поддерживает два отдельных внутренних буфера для чтения и записи. Это позволяет каждой стороне работать независимо от другой, обеспечивая надлежащий и эффективный поток данных. Например, экземпляры net.Socket — это потоки Duplex, у которых сторона Readable позволяет получать данные, поступающие из сокета, а сторона Writable позволяет записывать данные в сокет. Поскольку данные могут записываться в сокет быстрее или медленнее, чем поступают, каждая сторона должна работать (и буферизовать данные) независимо от другой.

Механизмы внутренней буферизации являются деталью реализации и могут быть изменены в любое время. Однако в некоторых сложных реализациях внутренние буферы можно получить с помощью writable.writableBuffer или readable.readableBuffer. Использовать эти недокументированные свойства не рекомендуется.

API для потребителей потоков

Почти все приложения Node.js, независимо от их сложности, так или иначе используют потоки. Ниже приведён пример использования потоков в приложении Node.js, реализующем HTTP-сервер:

const http = require('node:http');

const server = http.createServer((req, res) => {
  // `req` is an http.IncomingMessage, which is a readable stream.
  // `res` is an http.ServerResponse, which is a writable stream.

  let body = '';
  // Get the data as utf8 strings.
  // If an encoding is not set, Buffer objects will be received.
  req.setEncoding('utf8');

  // Readable streams emit 'data' events once a listener is added.
  req.on('data', (chunk) => {
    body += chunk;
  });

  // The 'end' event indicates that the entire body has been received.
  req.on('end', () => {
    try {
      const data = JSON.parse(body);
      // Write back something interesting to the user:
      res.write(typeof data);
      res.end();
    } catch (er) {
      // uh oh! bad json!
      res.statusCode = 400;
      return res.end(`error: ${er.message}`);
    }
  });
});

server.listen(1337);

// $ curl localhost:1337 -d "{}"
// object
// $ curl localhost:1337 -d "\"foo\""
// string
// $ curl localhost:1337 -d "not json"
// error: Unexpected token 'o', "not json" is not valid JSON copy

Потоки Writable (например, res в примере) предоставляют такие методы, как write() и end(), которые используются для записи данных в поток.

Потоки Readable используют API EventEmitter, чтобы уведомлять код приложения о доступности данных для чтения из потока. Эти данные можно читать из потока несколькими способами.

Потоки Writable и Readable используют API EventEmitter различными способами для передачи сведений о текущем состоянии потока.

Потоки Duplex и Transform являются одновременно потоками Writable и Readable.

Приложениям, которые записывают данные в поток или считывают их из потока, не требуется напрямую реализовывать интерфейсы потоков, и обычно у них нет причин вызывать require('node:stream').

Разработчикам, желающим реализовать новые типы потоков, следует обратиться к разделу API для разработчиков потоков.

Потоки для записи

Потоки для записи представляют собой абстракцию назначения, в которое записываются данные.

К примерам потоков Writable относятся:

• HTTP-запросы на стороне клиента
• HTTP-ответы на стороне сервера
• потоки записи fs
• потоки zlib
• потоки crypto
• TCP-сокеты
• stdin дочернего процесса
• process.stdout, process.stderr

Некоторые из этих примеров на самом деле являются потоками Duplex, реализующими интерфейс Writable.

Все потоки Writable реализуют интерфейс, определённый классом stream.Writable.

Хотя конкретные экземпляры потоков Writable могут различаться по-разному, все потоки Writable следуют одной и той же основной схеме использования, показанной в примере ниже:

const myStream = getWritableStreamSomehow();
myStream.write('some data');
myStream.write('some more data');
myStream.end('done writing data'); copy

Класс: stream.Writable

Событие: 'close'

Событие 'close' генерируется, когда поток и все его базовые ресурсы (например, дескриптор файла) закрыты. Это событие означает, что больше не будет сгенерировано никаких событий и дальнейшие вычисления выполняться не будут.

Поток Writable всегда генерирует событие 'close', если он создан с параметром emitClose.

Событие: 'drain'

Если вызов stream.write(chunk) возвращает false, событие 'drain' будет сгенерировано, когда можно будет возобновить запись данных в поток.

// Write the data to the supplied writable stream one million times.
// Be attentive to back-pressure.
function writeOneMillionTimes(writer, data, encoding, callback) {
  let i = 1000000;
  write();
  function write() {
    let ok = true;
    do {
      i--;
      if (i === 0) {
        // Last time!
        writer.write(data, encoding, callback);
      } else {
        // See if we should continue, or wait.
        // Don't pass the callback, because we're not done yet.
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // Had to stop early!
      // Write some more once it drains.
      writer.once('drain', write);
    }
  }
} copy

Событие: 'error'

• <Error>

Событие 'error' генерируется, если при записи или передаче данных через поток произошла ошибка. При вызове функции обратного вызова обработчика ей передаётся один аргумент Error.

При генерации события 'error' поток закрывается, если при его создании параметру autoDestroy не было присвоено значение false.

После 'error' не должны генерироваться никакие события, кроме 'close' (включая события 'error').

Событие: 'finish'

Событие 'finish' генерируется после вызова метода stream.end() и сброса всех данных в базовую систему.

const writer = getWritableStreamSomehow();
for (let i = 0; i < 100; i++) {
  writer.write(`hello, #${i}!\n`);
}
writer.on('finish', () => {
  console.log('All writes are now complete.');
});
writer.end('This is the end\n'); copy

Событие: 'pipe'

• src <stream.Readable> исходный поток, передающий данные в этот поток для записи

Событие 'pipe' генерируется при вызове метода stream.pipe() для потока для чтения, добавляющего этот поток для записи в список своих назначений.

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('pipe', (src) => {
  console.log('Something is piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer); copy

Событие: 'unpipe'

• src <stream.Readable> исходный поток, который перестал передавать данные в этот поток для записи с помощью отключения передачи

Событие 'unpipe' генерируется при вызове метода stream.unpipe() для потока Readable, удаляющего этот поток Writable из списка назначений.

Это событие также генерируется, если в этом потоке Writable возникает ошибка при передаче в него данных из потока Readable.

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('unpipe', (src) => {
  console.log('Something has stopped piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer);
reader.unpipe(writer); copy

writable.cork()

Метод writable.cork() принудительно помещает все записываемые данные в буфер в памяти. Буферизованные данные будут сброшены при вызове метода stream.uncork() или stream.end().

Основное назначение writable.cork() — обеспечить обработку ситуации, когда в поток быстро записывается несколько небольших фрагментов данных. Вместо немедленной передачи их в базовое назначение writable.cork() буферизует все фрагменты до вызова writable.uncork(), который передаст их все в writable._writev(), если он реализован. Это предотвращает блокировку в начале очереди, когда данные буферизуются в ожидании обработки первого небольшого фрагмента. Однако использование writable.cork() без реализации writable._writev() может отрицательно сказаться на пропускной способности.

См. также: writable.uncork(), writable._writev().

writable.destroy([error])

• error <Error> Необязательная ошибка, передаваемая вместе с событием 'error'.
• Возвращает: <this>

Уничтожает поток. При необходимости генерирует событие 'error', а также событие 'close' (если только для emitClose не задано значение false). После этого вызова поток для записи завершается, а последующие вызовы write() или end() приведут к ошибке ERR_STREAM_DESTROYED. Это немедленный и необратимый способ уничтожения потока. Предыдущие вызовы write() могли не завершиться и могут вызвать ошибку ERR_STREAM_DESTROYED. Если перед закрытием следует сбросить данные, используйте end() вместо destroy либо дождитесь события 'drain', прежде чем уничтожать поток.

const { Writable } = require('node:stream');

const myStream = new Writable();

const fooErr = new Error('foo error');
myStream.destroy(fooErr);
myStream.on('error', (fooErr) => console.error(fooErr.message)); // foo error copy

const { Writable } = require('node:stream');

const myStream = new Writable();

myStream.destroy();
myStream.on('error', function wontHappen() {}); copy

const { Writable } = require('node:stream');

const myStream = new Writable();
myStream.destroy();

myStream.write('foo', (error) => console.error(error.code));
// ERR_STREAM_DESTROYED copy

После вызова destroy() все последующие вызовы ничего не будут делать, а никакие дальнейшие ошибки, кроме ошибок из _destroy(), не могут быть сгенерированы как 'error'.

Разработчикам не следует переопределять этот метод; вместо этого следует реализовать writable._destroy().

writable.closed

• <boolean>

Имеет значение true после генерации события 'close'.

writable.destroyed

• <boolean>

Имеет значение true после вызова writable.destroy().

const { Writable } = require('node:stream');

const myStream = new Writable();

console.log(myStream.destroyed); // false
myStream.destroy();
console.log(myStream.destroyed); // true copy

writable.end([chunk[, encoding]][, callback])

• chunk <string> | <Buffer> | <TypedArray> | <DataView> | <any> Необязательные данные для записи. Для потоков, работающих не в объектном режиме, chunk должен быть значением типа <string>, <Buffer>, <TypedArray> или <DataView>. Для потоков в объектном режиме chunk может иметь любое значение JavaScript, кроме null.
• encoding <string> Кодировка, если chunk — строка
• callback <Function> Функция обратного вызова, вызываемая по завершении работы потока.
• Возвращает: <this>

Вызов метода writable.end() указывает, что в поток Writable больше не будут записываться данные. Необязательные аргументы chunk и encoding позволяют записать ещё один заключительный фрагмент данных непосредственно перед закрытием потока.

Вызов метода stream.write() после вызова stream.end() приведёт к ошибке.

// Write 'hello, ' and then end with 'world!'.
const fs = require('node:fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed! copy

writable.setDefaultEncoding(encoding)

• encoding <string> Новая кодировка по умолчанию
• Возвращает: <this>

Метод writable.setDefaultEncoding() задаёт кодировку encoding по умолчанию для потока Writable.

writable.uncork()

Метод writable.uncork() сбрасывает все данные, буферизованные с момента вызова stream.cork().

Если для управления буферизацией записи в поток используются writable.cork() и writable.uncork(), откладывайте вызовы writable.uncork() с помощью process.nextTick(). Это позволяет объединить все вызовы writable.write(), выполняемые на протяжении одной фазы цикла событий Node.js.

stream.cork();
stream.write('some ');
stream.write('data ');
process.nextTick(() => stream.uncork()); copy

Если метод writable.cork() вызывается для потока несколько раз, для сброса буферизованных данных необходимо вызвать writable.uncork() столько же раз.

stream.cork();
stream.write('some ');
stream.cork();
stream.write('data ');
process.nextTick(() => {
  stream.uncork();
  // The data will not be flushed until uncork() is called a second time.
  stream.uncork();
}); copy

См. также: writable.cork().

writable.writable

• <boolean>

Имеет значение true, если безопасно вызывать writable.write(), то есть если поток не был уничтожен, не завершился с ошибкой и не завершил работу.

writable.writableAborted

• <boolean>

Возвращает, был ли поток уничтожен или завершился с ошибкой до генерации 'finish'.

writable.writableEnded

• <boolean>

Имеет значение true после вызова writable.end(). Это свойство не указывает, были ли данные сброшены; для этого используйте writable.writableFinished.

writable.writableCorked

• <integer>

Количество вызовов writable.uncork(), необходимых для полного снятия буферизации потока.

writable.errored

• <Error>

Возвращает ошибку, если поток был уничтожен из-за ошибки.

writable.writableFinished

• <boolean>

Принимает значение true непосредственно перед генерацией события 'finish'.

writable.writableHighWaterMark

• <number>

Возвращает значение highWaterMark, переданное при создании этого Writable.

writable.writableLength

• <number>

Это свойство содержит количество байтов (или объектов) в очереди, готовых к записи. Значение предоставляет данные для анализа состояния highWaterMark.

writable.writableNeedDrain

• <boolean>

Имеет значение true, если буфер потока был заполнен и поток сгенерирует событие 'drain'.

writable.writableObjectMode

• <boolean>

Геттер свойства objectMode заданного потока Writable.

writable[Symbol.asyncDispose]()

Вызывает writable.destroy() с AbortError и возвращает промис, который выполняется после завершения работы потока.

writable.write(chunk[, encoding][, callback])

• chunk <string> | <Buffer> | <TypedArray> | <DataView> | <any> Необязательные данные для записи. Для потоков, работающих не в объектном режиме, chunk должен быть значением типа <string>, <Buffer>, <TypedArray> или <DataView>. Для потоков в объектном режиме chunk может иметь любое значение JavaScript, кроме null.
• encoding <string> | <null> Кодировка, если chunk — строка. По умолчанию: 'utf8'
• callback <Function> Функция обратного вызова, вызываемая после сброса этого фрагмента данных.
• Возвращает: <boolean> false, если вызывающий код должен дождаться генерации события 'drain', прежде чем продолжить запись дополнительных данных; в противном случае — true.

Метод writable.write() записывает данные в поток и вызывает переданную функцию callback после полной обработки данных. Если возникает ошибка, callback будет вызвана, а первым аргументом ей будет передана ошибка. callback вызывается асинхронно и до генерации 'error'.

Возвращаемое значение равно true, если после приёма chunk внутренний буфер меньше highWaterMark, заданного при создании потока. Если возвращается false, дальнейшую запись данных в поток следует прекратить до генерации события 'drain'.

Пока поток не готов к сбросу данных, вызовы write() будут буферизовать chunk и возвращать false. После сброса всех буферизованных фрагментов (их передачи операционной системе) будет сгенерировано событие 'drain'. Если write() возвращает false, не записывайте новые фрагменты до генерации события 'drain'. Хотя вызовы write() для потока, не готового к сбросу данных, допустимы, Node.js будет буферизовать все записанные фрагменты до достижения максимального потребления памяти, после чего безусловно прервёт работу. Ещё до аварийного завершения высокое потребление памяти ухудшит производительность сборщика мусора и увеличит RSS (который обычно не возвращается системе, даже когда память больше не нужна). Поскольку TCP-сокеты могут никогда не освободить буфер, если удалённый узел не считывает данные, запись в такой сокет может привести к уязвимости, которой можно воспользоваться удалённо.

Запись данных, когда поток не готов к сбросу, особенно проблематична для потока Transform, поскольку потоки Transform по умолчанию приостановлены до тех пор, пока данные не будут переданы с помощью pipe либо не будет добавлен обработчик события 'data' или 'readable'.

Если данные можно сгенерировать или получить по запросу, рекомендуется инкапсулировать эту логику в потоке Readable и использовать stream.pipe(). Однако, если предпочтительно вызывать write(), можно учитывать противодавление и избегать проблем с памятью с помощью события 'drain':

function write(data, cb) {
  if (!stream.write(data)) {
    stream.once('drain', cb);
  } else {
    process.nextTick(cb);
  }
}

// Wait for cb to be called before doing any other write.
write('hello', () => {
  console.log('Write completed, do more writes now.');
}); copy

Поток Writable в объектном режиме всегда игнорирует аргумент encoding.

Потоки для чтения

Потоки для чтения представляют собой абстракцию источника, из которого считываются данные.

К примерам потоков Readable относятся:

• HTTP-ответы на стороне клиента
• HTTP-запросы на стороне сервера
• потоки чтения fs
• потоки zlib
• потоки crypto
• TCP-сокеты
• stdout и stderr дочернего процесса
• process.stdin

Все потоки Readable реализуют интерфейс, определённый классом stream.Readable.

Два режима чтения

Потоки Readable работают в одном из двух режимов: активном или приостановленном. Эти режимы не зависят от объектного режима. Поток Readable может работать или не работать в объектном режиме независимо от того, находится ли он в активном или приостановленном режиме.

• В активном режиме данные автоматически считываются из базовой системы и передаются приложению как можно быстрее с помощью событий через интерфейс EventEmitter.

• В приостановленном режиме для чтения фрагментов данных из потока необходимо явно вызывать метод stream.read().

Все потоки Readable начинают работу в приостановленном режиме, но могут переключиться в активный режим одним из следующих способов:

• Добавить обработчик события 'data'.
• Вызвать метод stream.resume().
• Вызвать метод stream.pipe(), чтобы передать данные потоку Writable.

Поток Readable можно переключить обратно в приостановленный режим одним из следующих способов:

• Если назначений передачи данных нет, вызвав метод stream.pause().
• Если назначения передачи данных есть, удалив их все. Несколько назначений можно удалить с помощью метода stream.unpipe().

Важно помнить, что Readable не будет генерировать данные, пока не будет предоставлен механизм их потребления или игнорирования. Если механизм потребления отключён или удалён, Readable попытается прекратить генерацию данных.

Для обеспечения обратной совместимости удаление обработчиков события 'data' не приводит к автоматической приостановке потока. Кроме того, если имеются назначения передачи данных, вызов stream.pause() не гарантирует, что поток останется приостановленным после того, как назначения обработают данные и запросят новые.

Если поток Readable переключён в активный режим и нет потребителей, способных обрабатывать данные, эти данные будут потеряны. Например, это может произойти, если вызвать метод readable.resume() без обработчика события 'data' или удалить из потока обработчик события 'data'.

Добавление обработчика события 'readable' автоматически останавливает передачу данных из потока, и данные необходимо потреблять с помощью readable.read(). Если обработчик события 'readable' удалён, поток снова начнёт передавать данные, если имеется обработчик события 'data'.

Три состояния

«Два режима» работы потока Readable — это упрощённая абстракция более сложного управления внутренним состоянием, которое происходит в реализации потока Readable.

В частности, в любой момент времени каждый Readable находится в одном из трёх возможных состояний:

• readable.readableFlowing === null
• readable.readableFlowing === false
• readable.readableFlowing === true

Когда readable.readableFlowing имеет значение null, механизм для потребления данных потока отсутствует. Поэтому поток не будет генерировать данные. В этом состоянии добавление обработчика события 'data', вызов метода readable.pipe() или вызов метода readable.resume() переключит readable.readableFlowing в состояние true, и Readable начнёт активно генерировать события по мере создания данных.

Вызов readable.pause(), readable.unpipe() или получение обратного давления приведёт к тому, что readable.readableFlowing будет установлено в состояние false, временно приостанавливая передачу событий, но не генерацию данных. В этом состоянии добавление обработчика события 'data' не переключит readable.readableFlowing в состояние true.

const { PassThrough, Writable } = require('node:stream');
const pass = new PassThrough();
const writable = new Writable();

pass.pipe(writable);
pass.unpipe(writable);
// readableFlowing is now false.

pass.on('data', (chunk) => { console.log(chunk.toString()); });
// readableFlowing is still false.
pass.write('ok');  // Will not emit 'data'.
pass.resume();     // Must be called to make stream emit 'data'.
// readableFlowing is now true. copy

Пока readable.readableFlowing имеет значение false, данные могут накапливаться во внутреннем буфере потока.

Выберите один стиль API

API потока Readable развивался в нескольких версиях Node.js и предоставляет несколько методов потребления данных потока. В целом разработчикам следует выбрать один способ потребления данных и никогда не использовать несколько способов для потребления данных из одного потока. В частности, сочетание on('data'), on('readable'), pipe() или асинхронных итераторов может привести к непредсказуемому поведению.

Класс: stream.Readable

Событие: 'close'

Событие 'close' генерируется, когда поток и все его базовые ресурсы (например, файловый дескриптор) закрыты. Это событие указывает, что больше не будет генерироваться событий и не будет выполняться дальнейших вычислений.

Поток Readable всегда генерирует событие 'close', если он создан с параметром emitClose.

Событие: 'data'

• chunk <Buffer> | <string> | <any> Фрагмент данных. Для потоков, не работающих в режиме объектов, фрагмент будет строкой или Buffer. Для потоков, работающих в режиме объектов, фрагментом может быть любое значение JavaScript, кроме null.

Событие 'data' генерируется всякий раз, когда поток передаёт потребителю владение фрагментом данных. Это может происходить при переключении потока в режим передачи данных вызовом readable.pipe(), readable.resume() или добавлением функции-обработчика для события 'data'. Событие 'data' также генерируется при вызове метода readable.read(), если доступен фрагмент данных для возврата.

Добавление обработчика события 'data' к потоку, который явно не был приостановлен, переключит поток в режим передачи данных. Данные будут передаваться, как только станут доступны.

Функции-обработчику будет передан фрагмент данных в виде строки, если для потока с помощью метода readable.setEncoding() задана кодировка по умолчанию; в противном случае данные будут переданы в виде Buffer.

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
}); copy

Событие: 'end'

Событие 'end' генерируется, когда в потоке больше нет данных для потребления.

Событие 'end' не будет сгенерировано, пока данные не будут полностью потреблены. Этого можно добиться, переключив поток в режим передачи данных или многократно вызывая stream.read(), пока все данные не будут потреблены.

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
});
readable.on('end', () => {
  console.log('There will be no more data.');
}); copy

Событие: 'error'

• <Error>

Событие 'error' может быть сгенерировано реализацией Readable в любой момент. Обычно это происходит, если базовый поток не может генерировать данные из-за внутреннего сбоя или если реализация потока пытается передать недопустимый фрагмент данных.

Функции-обработчику будет передан один объект Error.

Событие: 'pause'

Событие 'pause' генерируется при вызове stream.pause(), если readableFlowing не имеет значения false.

Событие: 'readable'

Событие 'readable' генерируется, когда в потоке доступны данные для чтения — в объёме до заданного порогового значения буфера (state.highWaterMark). Фактически оно указывает, что в буфере потока появилась новая информация. Если данные доступны в этом буфере, для их получения можно вызвать stream.read(). Кроме того, событие 'readable' может быть сгенерировано и при достижении конца потока.

const readable = getReadableStreamSomehow();
readable.on('readable', function() {
  // There is some data to read now.
  let data;

  while ((data = this.read()) !== null) {
    console.log(data);
  }
}); copy

Если достигнут конец потока, вызов stream.read() вернёт null и вызовет событие 'end'. То же самое произойдёт, если данных для чтения не было вовсе. Например, в следующем примере foo.txt — пустой файл:

const fs = require('node:fs');
const rr = fs.createReadStream('foo.txt');
rr.on('readable', () => {
  console.log(`readable: ${rr.read()}`);
});
rr.on('end', () => {
  console.log('end');
}); copy

Результат выполнения этого скрипта:

$ node test.js
readable: null
end copy

В некоторых случаях добавление обработчика события 'readable' приводит к чтению некоторого объёма данных во внутренний буфер.

В целом механизмы событий readable.pipe() и 'data' проще для понимания, чем событие 'readable'. Однако обработка 'readable' может повысить пропускную способность.

Если одновременно используются 'readable' и 'data', управление потоком будет осуществляться с приоритетом 'readable', то есть 'data' будет генерироваться только при вызове stream.read(). Свойство readableFlowing будет иметь значение false. Если при удалении 'readable' останутся обработчики 'data', поток начнёт передавать данные, то есть события 'data' будут генерироваться без вызова .resume().

Событие: 'resume'

Событие 'resume' генерируется при вызове stream.resume(), если readableFlowing не имеет значения true.

readable.destroy([error])

• error <Error> Ошибка, передаваемая в качестве данных события 'error'
• Возвращает: <this>

Уничтожает поток. При необходимости генерирует событие 'error' и событие 'close' (если только emitClose не установлено в false). После этого вызова поток для чтения освободит все внутренние ресурсы, а последующие вызовы push() будут игнорироваться.

После вызова destroy() все последующие вызовы ничего не будут делать, и никакие дальнейшие ошибки, кроме _destroy(), не будут генерироваться как 'error'.

Реализующим поток разработчикам не следует переопределять этот метод; вместо этого следует реализовать readable._destroy().

readable.closed

• <boolean>

Имеет значение true после генерации 'close'.

readable.destroyed

• <boolean>

Имеет значение true после вызова readable.destroy().

readable.isPaused()

• Возвращает: <boolean>

Метод readable.isPaused() возвращает текущее состояние работы Readable. Он используется главным образом механизмом, лежащим в основе метода readable.pipe(). В большинстве обычных случаев нет необходимости вызывать этот метод напрямую.

const readable = new stream.Readable();

readable.isPaused(); // === false
readable.pause();
readable.isPaused(); // === true
readable.resume();
readable.isPaused(); // === false copy

readable.pause()

• Возвращает: <this>

Метод readable.pause() останавливает генерацию событий 'data' потоком в режиме передачи данных и выводит его из этого режима. Все появляющиеся данные остаются во внутреннем буфере.

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
  readable.pause();
  console.log('There will be no additional data for 1 second.');
  setTimeout(() => {
    console.log('Now data will start flowing again.');
    readable.resume();
  }, 1000);
}); copy

Метод readable.pause() не действует, если имеется обработчик события 'readable'.

readable.pipe(destination[, options])

• destination <stream.Writable> Получатель данных
• options <Object> Параметры передачи
  • end <boolean> Завершить запись, когда чтение будет завершено. По умолчанию: true.
• Возвращает: <stream.Writable> Поток-получатель, позволяющий создавать цепочки потоков, если это поток Duplex или Transform

Метод readable.pipe() подключает поток Writable к readable, автоматически переключая его в режим передачи данных и отправляя все его данные в подключённый поток Writable. Передача данных будет автоматически регулироваться, чтобы более быстрый поток Readable не перегружал поток-получатель Writable.

Следующий пример передаёт все данные из readable в файл с именем file.txt:

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt'.
readable.pipe(writable); copy

К одному потоку Readable можно подключить несколько потоков Writable.

Метод readable.pipe() возвращает ссылку на поток-получатель, что позволяет создавать цепочки передаваемых потоков:

const fs = require('node:fs');
const zlib = require('node:zlib');
const r = fs.createReadStream('file.txt');
const z = zlib.createGzip();
const w = fs.createWriteStream('file.txt.gz');
r.pipe(z).pipe(w); copy

По умолчанию для потока-получателя Writable вызывается stream.end(), когда исходный поток Readable генерирует 'end', в результате чего поток-получатель перестаёт принимать данные для записи. Чтобы отключить такое поведение по умолчанию, параметр end можно передать со значением false, и поток-получатель останется открытым:

reader.pipe(writer, { end: false });
reader.on('end', () => {
  writer.end('Goodbye\n');
}); copy

Важно учитывать, что если во время обработки поток Readable генерирует ошибку, поток-получатель Writable не закрывается автоматически. В случае ошибки каждый поток необходимо закрыть вручную, чтобы предотвратить утечки памяти.

Потоки Writable process.stderr и process.stdout никогда не закрываются до завершения процесса Node.js, независимо от заданных параметров.

readable.read([size])

• size <number> Необязательный аргумент, задающий объём данных для чтения.
• Возвращает: <string> | <Buffer> | <null> | <any>

Метод readable.read() читает данные из внутреннего буфера и возвращает их. Если данных для чтения нет, возвращается null. По умолчанию данные возвращаются как объект Buffer, если только кодировка не задана методом readable.setEncoding() или поток не работает в режиме объектов.

Необязательный аргумент size задаёт точное количество байтов для чтения. Если доступно менее size байтов, возвращается null, если только поток не завершился; в этом случае возвращаются все данные, оставшиеся во внутреннем буфере.

Если аргумент size не указан, возвращаются все данные из внутреннего буфера.

Аргумент size должен быть меньше или равен 1 ГиБ.

Метод readable.read() следует вызывать только для потоков Readable, работающих в режиме паузы. В режиме передачи данных readable.read() вызывается автоматически, пока внутренний буфер не будет полностью опустошён.

const readable = getReadableStreamSomehow();

// 'readable' may be triggered multiple times as data is buffered in
readable.on('readable', () => {
  let chunk;
  console.log('Stream is readable (new data received in buffer)');
  // Use a loop to make sure we read all currently available data
  while (null !== (chunk = readable.read())) {
    console.log(`Read ${chunk.length} bytes of data...`);
  }
});

// 'end' will be triggered once when there is no more data available
readable.on('end', () => {
  console.log('Reached end of stream.');
}); copy

Каждый вызов readable.read() возвращает фрагмент данных или null, указывающий, что в данный момент больше нет данных для чтения. Эти фрагменты не объединяются автоматически. Поскольку один вызов read() не возвращает все данные, может потребоваться цикл while, который будет непрерывно читать фрагменты, пока не будут получены все данные. При чтении большого файла .read() может временно вернуть null, указывая, что все данные из буфера были прочитаны, но в буфер могут поступить новые данные. В таких случаях новое событие 'readable' генерируется, когда в буфере появятся дополнительные данные, а событие 'end' обозначает конец передачи данных.

Поэтому для чтения всего содержимого файла из readable необходимо собирать фрагменты данных в нескольких событиях 'readable':

const chunks = [];

readable.on('readable', () => {
  let chunk;
  while (null !== (chunk = readable.read())) {
    chunks.push(chunk);
  }
});

readable.on('end', () => {
  const content = chunks.join('');
}); copy

Поток Readable в режиме объектов всегда возвращает один элемент при вызове readable.read(size), независимо от значения аргумента size.

Если метод readable.read() возвращает фрагмент данных, также генерируется событие 'data'.

Вызов stream.read([size]) после генерации события 'end' вернёт null. Ошибка времени выполнения не возникнет.

readable.readable

• <boolean>

Имеет значение true, если безопасно вызвать readable.read(), то есть поток не был уничтожен и не генерировал 'error' или 'end'.

readable.readableAborted

• <boolean>

Возвращает, был ли поток уничтожен или в нём возникла ошибка до генерации 'end'.

readable.readableDidRead

• <boolean>

Возвращает, было ли сгенерировано 'data'.

readable.readableEncoding

• <null> | <string>

Геттер свойства encoding заданного потока Readable. Свойство encoding можно задать с помощью метода readable.setEncoding().

readable.readableEnded

• <boolean>

Приобретает значение true, когда генерируется событие 'end'.

readable.errored

• <Error>

Возвращает ошибку, если поток был уничтожен из-за ошибки.

readable.readableFlowing

• <boolean>

Это свойство отражает текущее состояние потока Readable, описанное в разделе Три состояния.

readable.readableHighWaterMark

• <number>

Возвращает значение highWaterMark, переданное при создании этого Readable.

readable.readableLength

• <number>

Это свойство содержит количество байтов (или объектов) в очереди, готовых к чтению. Значение предоставляет данные о состоянии highWaterMark.

readable.readableObjectMode

• <boolean>

Геттер свойства objectMode заданного потока Readable.

readable.resume()

• Возвращает: <this>

Метод readable.resume() возобновляет генерацию событий 'data' явно приостановленным потоком Readable и переключает его в режим передачи данных.

Метод readable.resume() можно использовать, чтобы полностью потребить данные из потока, не обрабатывая их:

getReadableStreamSomehow()
  .resume()
  .on('end', () => {
    console.log('Reached the end, but did not read anything.');
  }); copy

Метод readable.resume() не действует, если имеется обработчик события 'readable'.

readable.setEncoding(encoding)

• encoding <string> Используемая кодировка.
• Возвращает: <this>

Метод readable.setEncoding() задаёт кодировку символов для данных, считываемых из потока Readable.

По умолчанию кодировка не задана, и данные потока возвращаются в виде объектов Buffer. Если задать кодировку, данные потока будут возвращаться в виде строк в указанной кодировке, а не объектов Buffer. Например, вызов readable.setEncoding('utf8') приведёт к интерпретации выходных данных как данных UTF-8 и их передаче в виде строк. Вызов readable.setEncoding('hex') приведёт к кодированию данных в виде шестнадцатеричной строки.

Поток Readable корректно обрабатывает многобайтовые символы, поступающие через поток, которые в противном случае могли бы декодироваться неправильно при простом чтении из потока в виде объектов Buffer.

const readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', (chunk) => {
  assert.equal(typeof chunk, 'string');
  console.log('Got %d characters of string data:', chunk.length);
}); copy

readable.unpipe([destination])

• destination <stream.Writable> Необязательный конкретный поток для отсоединения
• Возвращает: <this>

Метод readable.unpipe() отсоединяет поток Writable, ранее подключённый с помощью метода stream.pipe().

Если destination не указан, отсоединяются все каналы передачи.

Если destination указан, но для него не настроен канал передачи, метод ничего не делает.

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt',
// but only for the first second.
readable.pipe(writable);
setTimeout(() => {
  console.log('Stop writing to file.txt.');
  readable.unpipe(writable);
  console.log('Manually close the file stream.');
  writable.end();
}, 1000); copy

readable.unshift(chunk[, encoding])

• chunk <Buffer> | <TypedArray> | <DataView> | <string> | <null> | <any> Фрагмент данных для добавления в начало очереди чтения. Для потоков, не работающих в режиме объектов, chunk должен быть значением типа <string>, <Buffer>, <TypedArray>, <DataView> или null. Для потоков в режиме объектов chunk может быть любым значением JavaScript.
• encoding <string> Кодировка строковых фрагментов. Должна быть допустимой кодировкой Buffer, например 'utf8' или 'ascii'.

Передача chunk в качестве null сигнализирует о конце потока (EOF) и действует так же, как readable.push(null), после чего больше нельзя записывать данные. Сигнал EOF помещается в конец буфера, а все буферизованные данные будут переданы.

Метод readable.unshift() помещает фрагмент данных обратно во внутренний буфер. Это полезно в некоторых ситуациях, когда поток потребляется кодом, которому нужно «отменить потребление» некоторого объёма данных, оптимистично извлечённых из источника, чтобы передать эти данные другой стороне.

Метод stream.unshift(chunk) нельзя вызывать после генерации события 'end', иначе будет выброшена ошибка времени выполнения.

Разработчикам, использующим stream.unshift(), часто следует рассмотреть возможность перехода на поток Transform. Дополнительные сведения см. в разделе API для разработчиков реализаций потоков.

// Pull off a header delimited by \n\n.
// Use unshift() if we get too much.
// Call the callback with (error, header, stream).
const { StringDecoder } = require('node:string_decoder');
function parseHeader(stream, callback) {
  stream.on('error', callback);
  stream.on('readable', onReadable);
  const decoder = new StringDecoder('utf8');
  let header = '';
  function onReadable() {
    let chunk;
    while (null !== (chunk = stream.read())) {
      const str = decoder.write(chunk);
      if (str.includes('\n\n')) {
        // Found the header boundary.
        const split = str.split(/\n\n/);
        header += split.shift();
        const remaining = split.join('\n\n');
        const buf = Buffer.from(remaining, 'utf8');
        stream.removeListener('error', callback);
        // Remove the 'readable' listener before unshifting.
        stream.removeListener('readable', onReadable);
        if (buf.length)
          stream.unshift(buf);
        // Now the body of the message can be read from the stream.
        callback(null, header, stream);
        return;
      }
      // Still reading the header.
      header += str;
    }
  }
} copy

В отличие от stream.push(chunk), stream.unshift(chunk) не завершает процесс чтения, сбрасывая внутреннее состояние чтения потока. Это может привести к неожиданным результатам, если readable.unshift() вызывается во время чтения (то есть внутри реализации stream._read() пользовательского потока). Немедленный вызов stream.push('') после readable.unshift() корректно сбросит состояние чтения, однако лучше просто не вызывать readable.unshift() во время чтения.

readable.wrap(stream)

• stream <Stream> Читаемый поток «старого стиля»
• Возвращает: <this>

До Node.js 0.10 потоки не реализовывали весь API модуля node:stream в том виде, в котором он определён сейчас. (Дополнительные сведения см. в разделе Совместимость.)

При использовании старой библиотеки Node.js, которая генерирует события 'data' и имеет метод stream.pause(), который служит лишь рекомендацией, метод readable.wrap() можно использовать для создания потока Readable, использующего старый поток в качестве источника данных.

Метод readable.wrap() редко бывает необходим, но он предоставлен для удобства взаимодействия со старыми приложениями и библиотеками Node.js.

const { OldReader } = require('./old-api-module.js');
const { Readable } = require('node:stream');
const oreader = new OldReader();
const myReader = new Readable().wrap(oreader);

myReader.on('readable', () => {
  myReader.read(); // etc.
}); copy

readable[Symbol.asyncIterator]()

• Возвращает: <AsyncIterator> для полного чтения потока.

const fs = require('node:fs');

async function print(readable) {
  readable.setEncoding('utf8');
  let data = '';
  for await (const chunk of readable) {
    data += chunk;
  }
  console.log(data);
}

print(fs.createReadStream('file')).catch(console.error); copy

Если цикл завершается оператором break, return или throw, поток будет уничтожен. Другими словами, итерация по потоку приводит к его полному чтению. Поток будет считываться блоками размером, равным параметру highWaterMark. В приведённом выше примере кода данные будут переданы одним блоком, если размер файла меньше 64 КиБ, поскольку для fs.createReadStream() не указан параметр highWaterMark.

readable[Symbol.asyncDispose]()

Вызывает readable.destroy() с объектом AbortError и возвращает промис, который выполняется после завершения потока.

readable.compose(stream[, options])

• stream <Stream> | <Iterable> | <AsyncIterable> | <Function>
• options <Object>
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Duplex> поток, объединённый с потоком stream.

import { Readable } from 'node:stream';

async function* splitToWords(source) {
  for await (const chunk of source) {
    const words = String(chunk).split(' ');

    for (const word of words) {
      yield word;
    }
  }
}

const wordsStream = Readable.from(['this is', 'compose as operator']).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // prints ['this', 'is', 'compose', 'as', 'operator'] copy

Дополнительные сведения см. в разделе stream.compose.

readable.iterator([options])

• options <Object>
  • destroyOnReturn <boolean> Если задано значение false, вызов return для асинхронного итератора или выход из итерации for await...of с помощью break, return или throw не уничтожит поток. По умолчанию: true.
• Возвращает: <AsyncIterator> для чтения потока.

Итератор, созданный этим методом, позволяет пользователям отменить уничтожение потока, если цикл for await...of завершён с помощью return, break или throw, а также задать уничтожение потока итератором, если во время итерации в потоке возникла ошибка.

const { Readable } = require('node:stream');

async function printIterator(readable) {
  for await (const chunk of readable.iterator({ destroyOnReturn: false })) {
    console.log(chunk); // 1
    break;
  }

  console.log(readable.destroyed); // false

  for await (const chunk of readable.iterator({ destroyOnReturn: false })) {
    console.log(chunk); // Will print 2 and then 3
  }

  console.log(readable.destroyed); // True, stream was totally consumed
}

async function printSymbolAsyncIterator(readable) {
  for await (const chunk of readable) {
    console.log(chunk); // 1
    break;
  }

  console.log(readable.destroyed); // true
}

async function showBoth() {
  await printIterator(Readable.from([1, 2, 3]));
  await printSymbolAsyncIterator(Readable.from([1, 2, 3]));
}

showBoth(); copy

readable.map(fn[, options])

• fn <Function> | <AsyncFunction> функция для обработки каждого блока потока.
  • data <any> блок данных из потока.
  • options <Object>
    • signal <AbortSignal> сигнал прерывается при уничтожении потока, что позволяет досрочно прервать вызов fn.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для обработки потока. По умолчанию: 1.
  • highWaterMark <number> количество элементов для буферизации в ожидании обработки пользователем преобразованных элементов. По умолчанию: concurrency * 2 - 1.
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Readable> поток, преобразованный функцией fn.

Этот метод позволяет преобразовывать поток. Функция fn вызывается для каждого блока потока. Если функция fn возвращает промис, этот промис будет обработан с помощью await перед передачей результата в выходной поток.

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous mapper.
for await (const chunk of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
  console.log(chunk); // 2, 4, 6, 8
}
// With an asynchronous mapper, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map((domain) => resolver.resolve4(domain), { concurrency: 2 });
for await (const result of dnsResults) {
  console.log(result); // Logs the DNS result of resolver.resolve4.
} copy

readable.filter(fn[, options])

• fn <Function> | <AsyncFunction> функция для фильтрации блоков потока.
  • data <any> блок данных из потока.
  • options <Object>
    • signal <AbortSignal> сигнал прерывается при уничтожении потока, что позволяет досрочно прервать вызов fn.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для обработки потока. По умолчанию: 1.
  • highWaterMark <number> количество элементов для буферизации в ожидании обработки пользователем отфильтрованных элементов. По умолчанию: concurrency * 2 - 1.
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Readable> поток, отфильтрованный предикатом fn.

Этот метод позволяет фильтровать поток. Для каждого блока потока вызывается функция fn, и если она возвращает истинное значение, блок передаётся в выходной поток. Если функция fn возвращает промис, этот промис будет обработан с помощью await.

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous predicate.
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// With an asynchronous predicate, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).filter(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address.ttl > 60;
}, { concurrency: 2 });
for await (const result of dnsResults) {
  // Logs domains with more than 60 seconds on the resolved dns record.
  console.log(result);
} copy

readable.forEach(fn[, options])

• fn <Function> | <AsyncFunction> функция, вызываемая для каждого блока потока.
  • data <any> блок данных из потока.
  • options <Object>
    • signal <AbortSignal> сигнал прерывается при уничтожении потока, что позволяет досрочно прервать вызов fn.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для обработки потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Promise> промис, который выполняется после завершения потока.

Этот метод позволяет перебирать поток. Для каждого блока потока вызывается функция fn. Если функция fn возвращает промис, этот промис будет обработан с помощью await.

Этот метод отличается от циклов for await...of тем, что позволяет при необходимости обрабатывать блоки параллельно. Кроме того, итерацию forEach можно остановить только передав параметр signal и прервав соответствующий AbortController, тогда как for await...of можно остановить с помощью break или return. В обоих случаях поток будет уничтожен.

Этот метод отличается от прослушивания события 'data' тем, что он использует событие readable в базовом механизме и позволяет ограничить количество одновременных вызовов fn.

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous predicate.
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// With an asynchronous predicate, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address;
}, { concurrency: 2 });
await dnsResults.forEach((result) => {
  // Logs result, similar to `for await (const result of dnsResults)`
  console.log(result);
});
console.log('done'); // Stream has finished copy

readable.toArray([options])

• options <Object>
  • signal <AbortSignal> позволяет отменить операцию toArray при прерывании сигнала.
• Возвращает: <Promise> промис, содержащий массив с содержимым потока.

Этот метод позволяет легко получить содержимое потока.

Поскольку этот метод считывает весь поток в память, он сводит на нет преимущества потоков. Он предназначен для обеспечения совместимости и удобства, а не для использования в качестве основного способа чтения потоков.

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

await Readable.from([1, 2, 3, 4]).toArray(); // [1, 2, 3, 4]

// Make dns queries concurrently using .map and collect
// the results into an array using toArray
const dnsResults = await Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address;
}, { concurrency: 2 }).toArray(); copy

readable.some(fn[, options])

• fn <Function> | <AsyncFunction> функция, вызываемая для каждого блока потока.
  • data <any> блок данных из потока.
  • options <Object>
    • signal <AbortSignal> сигнал прерывается при уничтожении потока, что позволяет досрочно прервать вызов fn.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для обработки потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Promise> промис, результатом которого будет true, если fn вернуло истинное значение хотя бы для одного из блоков.

Этот метод похож на Array.prototype.some и вызывает fn для каждого блока потока, пока ожидаемое возвращаемое значение не станет true (или любым истинным значением). Когда ожидаемое возвращаемое значение вызова fn для блока становится истинным, поток уничтожается, а промис выполняется со значением true. Если ни один вызов fn для блоков не возвращает истинное значение, промис выполняется со значением false.

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).some((x) => x > 2); // true
await Readable.from([1, 2, 3, 4]).some((x) => x < 0); // false

// With an asynchronous predicate, making at most 2 file checks at a time.
const anyBigFile = await Readable.from([
  'file1',
  'file2',
  'file3',
]).some(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
console.log(anyBigFile); // `true` if any file in the list is bigger than 1MB
console.log('done'); // Stream has finished copy

readable.find(fn[, options])

• fn <Function> | <AsyncFunction> функция, вызываемая для каждого блока потока.
  • data <any> блок данных из потока.
  • options <Object>
    • signal <AbortSignal> сигнал прерывается при уничтожении потока, что позволяет досрочно прервать вызов fn.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для обработки потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Promise> промис, результатом которого будет первый блок, для которого fn вычислило истинное значение, или undefined, если элемент не найден.

Этот метод похож на Array.prototype.find и вызывает fn для каждого блока потока, чтобы найти блок, для которого fn имеет истинное значение. Когда ожидаемое возвращаемое значение вызова fn становится истинным, поток уничтожается, а промис выполняется со значением, для которого fn вернуло истинное значение. Если все вызовы fn для блоков возвращают ложное значение, промис выполняется со значением undefined.

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).find((x) => x > 2); // 3
await Readable.from([1, 2, 3, 4]).find((x) => x > 0); // 1
await Readable.from([1, 2, 3, 4]).find((x) => x > 10); // undefined

// With an asynchronous predicate, making at most 2 file checks at a time.
const foundBigFile = await Readable.from([
  'file1',
  'file2',
  'file3',
]).find(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
console.log(foundBigFile); // File name of large file, if any file in the list is bigger than 1MB
console.log('done'); // Stream has finished copy

readable.every(fn[, options])

• fn <Function> | <AsyncFunction> функция, вызываемая для каждого блока потока.
  • data <any> блок данных из потока.
  • options <Object>
    • signal <AbortSignal> сигнал прерывается при уничтожении потока, что позволяет досрочно прервать вызов fn.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для обработки потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Promise> промис, результатом которого будет true, если fn вернуло истинное значение для всех блоков.

Этот метод похож на Array.prototype.every и вызывает fn для каждого блока потока, чтобы проверить, являются ли все ожидаемые возвращаемые значения истинными для fn. Когда ожидаемое возвращаемое значение вызова fn для блока становится ложным, поток уничтожается, а промис выполняется со значением false. Если все вызовы fn для блоков возвращают истинное значение, промис выполняется со значением true.

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).every((x) => x > 2); // false
await Readable.from([1, 2, 3, 4]).every((x) => x > 0); // true

// With an asynchronous predicate, making at most 2 file checks at a time.
const allBigFiles = await Readable.from([
  'file1',
  'file2',
  'file3',
]).every(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
// `true` if all files in the list are bigger than 1MiB
console.log(allBigFiles);
console.log('done'); // Stream has finished copy

readable.flatMap(fn[, options])

• fn <Function> | <AsyncGeneratorFunction> | <AsyncFunction> функция для обработки каждого блока потока.
  • data <any> блок данных из потока.
  • options <Object>
    • signal <AbortSignal> сигнал прерывается при уничтожении потока, что позволяет досрочно прервать вызов fn.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для обработки потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Readable> поток, преобразованный функцией fn с разворачиванием результата.

Этот метод возвращает новый поток, применяя заданную функцию обратного вызова к каждому блоку потока, а затем разворачивая результат.

Функция fn может возвращать поток, другой итерируемый или асинхронно итерируемый объект; результирующие потоки будут объединены (развёрнуты) в возвращаемый поток.

import { Readable } from 'node:stream';
import { createReadStream } from 'node:fs';

// With a synchronous mapper.
for await (const chunk of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
  console.log(chunk); // 1, 1, 2, 2, 3, 3, 4, 4
}
// With an asynchronous mapper, combine the contents of 4 files
const concatResult = Readable.from([
  './1.mjs',
  './2.mjs',
  './3.mjs',
  './4.mjs',
]).flatMap((fileName) => createReadStream(fileName));
for await (const result of concatResult) {
  // This will contain the contents (all chunks) of all 4 files
  console.log(result);
} copy

readable.drop(limit[, options])

• limit <number> количество блоков, которые нужно отбросить из читаемого потока.
• options <Object>
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Readable> поток, из которого отброшено limit блоков.

Этот метод возвращает новый поток, из которого отброшены первые limit блоков.

import { Readable } from 'node:stream';

await Readable.from([1, 2, 3, 4]).drop(2).toArray(); // [3, 4] copy

readable.take(limit[, options])

• limit <number> количество блоков, которые нужно получить из читаемого потока.
• options <Object>
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Readable> поток, содержащий limit блоков.

Этот метод возвращает новый поток с первыми limit блоками.

import { Readable } from 'node:stream';

await Readable.from([1, 2, 3, 4]).take(2).toArray(); // [1, 2] copy

readable.asIndexedPairs([options])

• options <Object>
  • signal <AbortSignal> позволяет уничтожить поток при прерывании сигнала.
• Возвращает: <Readable> поток индексированных пар.

Этот метод возвращает новый поток, в котором блоки базового потока объединены со счётчиком в форме [index, chunk]. Первое значение индекса равно 0 и увеличивается на 1 для каждого созданного блока.

import { Readable } from 'node:stream';

const pairs = await Readable.from(['a', 'b', 'c']).asIndexedPairs().toArray();
console.log(pairs); // [[0, 'a'], [1, 'b'], [2, 'c']] copy

readable.reduce(fn[, initial[, options]])

• fn <Function> | <AsyncFunction> функция-редуктор, вызываемая для каждого фрагмента потока.
  • previous <any> значение, полученное при последнем вызове fn, или значение initial, если оно указано, или первый фрагмент потока в противном случае.
  • data <any> фрагмент данных из потока.
  • options <Object>
    • signal <AbortSignal> сигнал отмены, срабатывающий при уничтожении потока и позволяющий досрочно прервать вызов fn.
• initial <any> начальное значение, используемое при свёртке.
• options <Object>
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал отменён.
• Возвращает: <Promise> промис с конечным значением свёртки.

Этот метод вызывает fn для каждого фрагмента потока по порядку, передавая ему результат вычисления для предыдущего элемента. Он возвращает промис с конечным значением свёртки.

Если значение initial не указано, в качестве начального значения используется первый фрагмент потока. Если поток пуст, промис отклоняется с ошибкой TypeError, у которой свойство кода имеет значение ERR_INVALID_ARGS.

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .reduce(async (totalSize, file) => {
    const { size } = await stat(join(directoryPath, file));
    return totalSize + size;
  }, 0);

console.log(folderSize); copy

Функция-редуктор обрабатывает элементы потока по одному, поэтому параметр concurrency и параллелизм отсутствуют. Чтобы выполнять reduce параллельно, можно извлечь асинхронную функцию в метод readable.map.

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .map((file) => stat(join(directoryPath, file)), { concurrency: 2 })
  .reduce((totalSize, { size }) => totalSize + size, 0);

console.log(folderSize); copy

Двунаправленные потоки и потоки преобразования

Класс: stream.Duplex

Двунаправленные потоки — это потоки, реализующие интерфейсы Readable и Writable.

Примеры потоков Duplex:

• TCP-сокеты
• потоки zlib
• потоки crypto

duplex.allowHalfOpen

• <boolean>

Если false, поток автоматически завершит записываемую сторону, когда завершится читаемая сторона. Изначально задаётся параметром конструктора allowHalfOpen, значение по умолчанию — true.

Это можно изменить вручную, чтобы поменять поведение с полуоткрытым состоянием существующего экземпляра потока Duplex, но изменить это значение необходимо до отправки события 'end'.

Класс: stream.Transform

Потоки преобразования — это потоки Duplex, в которых выходные данные так или иначе зависят от входных. Как и все потоки Duplex, потоки Transform реализуют интерфейсы Readable и Writable.

Примеры потоков Transform:

• потоки zlib
• потоки crypto

transform.destroy([error])

• error <Error>
• Возвращает: <this>

Уничтожает поток и при необходимости отправляет событие 'error'. После этого вызова поток преобразования освободит все внутренние ресурсы. Реализующим классам не следует переопределять этот метод; вместо этого следует реализовать readable._destroy(). Реализация _destroy() по умолчанию для Transform также отправляет событие 'close', если для emitClose не установлено значение false.

После вызова destroy() все последующие вызовы ничего не делают, и ошибки больше не отправляются как 'error', за исключением ошибок, возникающих в _destroy().

stream.duplexPair([options])

• options <Object> Значение, передаваемое обоим конструкторам Duplex для задания таких параметров, как буферизация.
• Возвращает: <Array> из двух экземпляров Duplex.

Вспомогательная функция duplexPair возвращает массив из двух элементов, каждый из которых является потоком Duplex, соединённым с другой стороной:

const [ sideA, sideB ] = duplexPair(); copy

Всё, что записывается в один поток, становится доступным для чтения из другого. Это поведение аналогично сетевому соединению: данные, записанные клиентом, становятся доступными для чтения сервером, и наоборот.

Двунаправленные потоки симметричны: их можно использовать взаимозаменяемо, без каких-либо различий в поведении.

stream.finished(stream[, options], callback)

• stream <Stream> | <ReadableStream> | <WritableStream> Читаемый и/или записываемый поток или веб-поток.
• options <Object>
  • error <boolean> Если установлено значение false, вызов emit('error', err) не считается завершённым. По умолчанию: true.
  • readable <boolean> Если установлено значение false, callback будет вызван при завершении потока, даже если поток всё ещё может быть доступен для чтения. По умолчанию: true.
  • writable <boolean> Если установлено значение false, callback будет вызван при завершении потока, даже если в поток всё ещё можно записывать данные. По умолчанию: true.
  • signal <AbortSignal> позволяет прервать ожидание завершения потока. Если сигнал отменён, сам поток не будет отменён. Callback будет вызван с ошибкой AbortError. Все зарегистрированные этой функцией обработчики также будут удалены.
  • cleanup <boolean> удалить все зарегистрированные обработчики событий потока. По умолчанию: false.
• callback <Function> Функция обратного вызова, принимающая необязательный аргумент ошибки.
• Возвращает: <Function> Функция очистки, удаляющая все зарегистрированные обработчики.

Функция для получения уведомления о том, что поток больше не доступен для чтения или записи, в нём произошла ошибка либо поток был преждевременно закрыт.

const { finished } = require('node:stream');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

finished(rs, (err) => {
  if (err) {
    console.error('Stream failed.', err);
  } else {
    console.log('Stream is done reading.');
  }
});

rs.resume(); // Drain the stream. copy

Особенно полезна при обработке ошибок, когда поток преждевременно уничтожается (например, при отмене HTTP-запроса) и не отправляет события 'end' или 'finish'.

API finished предоставляет вариант на основе промиса.

stream.finished() оставляет незакреплённые обработчики событий (в частности, 'error', 'end', 'finish' и 'close') после вызова callback. Это сделано для того, чтобы неожиданные события 'error' (вызванные некорректной реализацией потока) не приводили к неожиданным сбоям. Если такое поведение нежелательно, функцию очистки, возвращённую из callback, необходимо вызвать внутри него:

const cleanup = finished(rs, (err) => {
  cleanup();
  // ...
}); copy

stream.pipeline(source[, ...transforms], destination, callback)

stream.pipeline(streams, callback)

• streams <Stream[]> | <Iterable[]> | <AsyncIterable[]> | <Function[]> | <ReadableStream[]> | <WritableStream[]> | <TransformStream[]>
• source <Stream> | <Iterable> | <AsyncIterable> | <Function> | <ReadableStream>
  • Возвращает: <Iterable> | <AsyncIterable>
• ...transforms <Stream> | <Function> | <TransformStream>
  • source <AsyncIterable>
  • Возвращает: <AsyncIterable>
• destination <Stream> | <Function> | <WritableStream>
  • source <AsyncIterable>
  • Возвращает: <AsyncIterable> | <Promise>
• callback <Function> Вызывается после полного завершения конвейера.
  • err <Error>
  • val Разрешённое значение Promise, возвращённое destination.
• Возвращает: <Stream>

Метод модуля для организации передачи данных между потоками и генераторами с пересылкой ошибок, корректной очисткой и вызовом callback после завершения конвейера.

const { pipeline } = require('node:stream');
const fs = require('node:fs');
const zlib = require('node:zlib');

// Use the pipeline API to easily pipe a series of streams
// together and get notified when the pipeline is fully done.

// A pipeline to gzip a potentially huge tar file efficiently:

pipeline(
  fs.createReadStream('archive.tar'),
  zlib.createGzip(),
  fs.createWriteStream('archive.tar.gz'),
  (err) => {
    if (err) {
      console.error('Pipeline failed.', err);
    } else {
      console.log('Pipeline succeeded.');
    }
  },
); copy

API pipeline предоставляет вариант на основе промиса.

stream.pipeline() вызывает stream.destroy(err) для всех потоков, кроме:

• Readable потоков, отправивших событие 'end' или 'close'.
• Writable потоков, отправивших событие 'finish' или 'close'.

stream.pipeline() оставляет незакреплённые обработчики событий в потоках после вызова callback. При повторном использовании потоков после сбоя это может привести к утечке обработчиков событий и подавлению ошибок. Если последний поток доступен для чтения, незакреплённые обработчики событий будут удалены, чтобы впоследствии можно было читать данные из последнего потока.

stream.pipeline() закрывает все потоки при возникновении ошибки. Использование IncomingRequest вместе с pipeline может привести к неожиданному поведению: сокет будет уничтожен без отправки ожидаемого ответа. См. пример ниже:

const fs = require('node:fs');
const http = require('node:http');
const { pipeline } = require('node:stream');

const server = http.createServer((req, res) => {
  const fileStream = fs.createReadStream('./fileNotExist.txt');
  pipeline(fileStream, res, (err) => {
    if (err) {
      console.log(err); // No such file
      // this message can't be sent once `pipeline` already destroyed the socket
      return res.end('error!!!');
    }
  });
}); copy

stream.compose(...streams)

• streams <Stream[]> | <Iterable[]> | <AsyncIterable[]> | <Function[]> | <ReadableStream[]> | <WritableStream[]> | <TransformStream[]> | <Duplex[]> | <Function>
• Возвращает: <stream.Duplex>

Объединяет два или более потоков в поток Duplex, который записывает данные в первый поток и считывает их из последнего. Каждый переданный поток соединяется со следующим с помощью stream.pipeline. При возникновении ошибки в любом потоке уничтожаются все потоки, в том числе внешний поток Duplex.

Поскольку stream.compose возвращает новый поток, который, в свою очередь, можно (и следует) передавать в другие потоки, это позволяет создавать композиции. В отличие от этого, при передаче потоков в stream.pipeline первый поток обычно доступен для чтения, а последний — для записи, образуя замкнутую цепь.

Если передан Function, он должен быть фабричным методом, принимающим source Iterable.

import { compose, Transform } from 'node:stream';

const removeSpaces = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, String(chunk).replace(' ', ''));
  },
});

async function* toUpper(source) {
  for await (const chunk of source) {
    yield String(chunk).toUpperCase();
  }
}

let res = '';
for await (const buf of compose(removeSpaces, toUpper).end('hello world')) {
  res += buf;
}

console.log(res); // prints 'HELLOWORLD' copy

stream.compose можно использовать для преобразования асинхронных итерируемых объектов, генераторов и функций в потоки.

• AsyncIterable преобразует объект в читаемый Duplex. Не может выдавать null.
• AsyncGeneratorFunction преобразует объект в читаемый и записываемый поток преобразования Duplex. В качестве первого параметра должен принимать исходный AsyncIterable. Не может выдавать null.
• AsyncFunction преобразует объект в записываемый Duplex. Должен возвращать null или undefined.

import { compose } from 'node:stream';
import { finished } from 'node:stream/promises';

// Convert AsyncIterable into readable Duplex.
const s1 = compose(async function*() {
  yield 'Hello';
  yield 'World';
}());

// Convert AsyncGenerator into transform Duplex.
const s2 = compose(async function*(source) {
  for await (const chunk of source) {
    yield String(chunk).toUpperCase();
  }
});

let res = '';

// Convert AsyncFunction into writable Duplex.
const s3 = compose(async function(source) {
  for await (const chunk of source) {
    res += chunk;
  }
});

await finished(compose(s1, s2, s3));

console.log(res); // prints 'HELLOWORLD' copy

См. readable.compose(stream), чтобы узнать о stream.compose в качестве оператора.

stream.Readable.from(iterable[, options])

• iterable <Iterable> Объект, реализующий протокол итерации Symbol.asyncIterator или Symbol.iterator. Если передано значение null, отправляется событие 'error'.
• options <Object> Параметры, передаваемые в new stream.Readable([options]). По умолчанию Readable.from() устанавливает options.objectMode в true, если явно не отключить это поведение, установив options.objectMode в false.
• Возвращает: <stream.Readable>

Вспомогательный метод для создания читаемых потоков на основе итераторов.

const { Readable } = require('node:stream');

async function * generate() {
  yield 'hello';
  yield 'streams';
}

const readable = Readable.from(generate());

readable.on('data', (chunk) => {
  console.log(chunk);
}); copy

Вызов Readable.from(string) или Readable.from(buffer) не приводит к итерации строк или буферов в соответствии с семантикой других потоков; это сделано для повышения производительности.

Если в качестве аргумента передан объект Iterable, содержащий промисы, это может привести к необработанному отклонению промиса.

const { Readable } = require('node:stream');

Readable.from([
  new Promise((resolve) => setTimeout(resolve('1'), 1500)),
  new Promise((_, reject) => setTimeout(reject(new Error('2')), 1000)), // Unhandled rejection
]); copy

stream.Readable.fromWeb(readableStream[, options])

• readableStream <ReadableStream>
• options <Object>
  • encoding <string>
  • highWaterMark <number>
  • objectMode <boolean>
  • signal <AbortSignal>
• Возвращает: <stream.Readable>

stream.Readable.isDisturbed(stream)

• stream <stream.Readable> | <ReadableStream>
• Возвращает: boolean

Возвращает значение, указывающее, выполнялось ли чтение из потока или была ли отменена операция.

stream.isErrored(stream)

• stream <Readable> | <Writable> | <Duplex> | <WritableStream> | <ReadableStream>
• Возвращает: <boolean>

Возвращает значение, указывающее, возникала ли в потоке ошибка.

stream.isReadable(stream)

• stream <Readable> | <Duplex> | <ReadableStream>
• Возвращает: <boolean>

Возвращает значение, указывающее, доступен ли поток для чтения.

stream.Readable.toWeb(streamReadable[, options])

• streamReadable <stream.Readable>
• options <Object>
  • strategy <Object>
    • highWaterMark <number> Максимальный размер внутренней очереди (созданного ReadableStream), при достижении которого при чтении из указанного stream.Readable применяется обратное давление. Если значение не задано, оно берётся из указанного stream.Readable.
    • size <Function> Функция, определяющая размер указанного фрагмента данных. Если значение не задано, размер всех фрагментов будет равен 1.
      • chunk <any>
      • Возвращает: <number>
• Возвращает: <ReadableStream>

stream.Writable.fromWeb(writableStream[, options])

• writableStream <WritableStream>
• options <Object>
  • decodeStrings <boolean>
  • highWaterMark <number>
  • objectMode <boolean>
  • signal <AbortSignal>
• Возвращает: <stream.Writable>

stream.Writable.toWeb(streamWritable)

• streamWritable <stream.Writable>
• Возвращает: <WritableStream>

stream.Duplex.from(src)

• src <Stream> | <Blob> | <ArrayBuffer> | <string> | <Iterable> | <AsyncIterable> | <AsyncGeneratorFunction> | <AsyncFunction> | <Promise> | <Object> | <ReadableStream> | <WritableStream>

Вспомогательный метод для создания двунаправленных потоков.

• Stream преобразует поток для записи в поток для записи Duplex, а поток для чтения — в Duplex.
• Blob преобразует в поток для чтения Duplex.
• string преобразует в поток для чтения Duplex.
• ArrayBuffer преобразует в поток для чтения Duplex.
• AsyncIterable преобразует в поток для чтения Duplex. Не может выдавать null.
• AsyncGeneratorFunction преобразует в преобразующий поток для чтения и записи Duplex. В качестве первого параметра должен принимать исходный AsyncIterable. Не может выдавать null.
• AsyncFunction преобразует в поток для записи Duplex. Должен возвращать либо null, либо undefined
• Object ({ writable, readable }) преобразует readable и writable в Stream, а затем объединяет их в Duplex, где Duplex будет записывать в writable и читать из readable.
• Promise преобразует в поток для чтения Duplex. Значение null игнорируется.
• ReadableStream преобразует в поток для чтения Duplex.
• WritableStream преобразует в поток для записи Duplex.
• Возвращает: <stream.Duplex>

Если в качестве аргумента передан объект Iterable, содержащий промисы, это может привести к необработанному отклонению промиса.

const { Duplex } = require('node:stream');

Duplex.from([
  new Promise((resolve) => setTimeout(resolve('1'), 1500)),
  new Promise((_, reject) => setTimeout(reject(new Error('2')), 1000)), // Unhandled rejection
]); copy

stream.Duplex.fromWeb(pair[, options])

• pair <Object>
  • readable <ReadableStream>
  • writable <WritableStream>
• options <Object>
  • allowHalfOpen <boolean>
  • decodeStrings <boolean>
  • encoding <string>
  • highWaterMark <number>
  • objectMode <boolean>
  • signal <AbortSignal>
• Возвращает: <stream.Duplex>

import { Duplex } from 'node:stream';
import {
  ReadableStream,
  WritableStream,
} from 'node:stream/web';

const readable = new ReadableStream({
  start(controller) {
    controller.enqueue('world');
  },
});

const writable = new WritableStream({
  write(chunk) {
    console.log('writable', chunk);
  },
});

const pair = {
  readable,
  writable,
};
const duplex = Duplex.fromWeb(pair, { encoding: 'utf8', objectMode: true });

duplex.write('hello');

for await (const chunk of duplex) {
  console.log('readable', chunk);
}

const { Duplex } = require('node:stream');
const {
  ReadableStream,
  WritableStream,
} = require('node:stream/web');

const readable = new ReadableStream({
  start(controller) {
    controller.enqueue('world');
  },
});

const writable = new WritableStream({
  write(chunk) {
    console.log('writable', chunk);
  },
});

const pair = {
  readable,
  writable,
};
const duplex = Duplex.fromWeb(pair, { encoding: 'utf8', objectMode: true });

duplex.write('hello');
duplex.once('readable', () => console.log('readable', duplex.read()));

stream.Duplex.toWeb(streamDuplex)

• streamDuplex <stream.Duplex>
• Возвращает: <Object>
  • readable <ReadableStream>
  • writable <WritableStream>

import { Duplex } from 'node:stream';

const duplex = Duplex({
  objectMode: true,
  read() {
    this.push('world');
    this.push(null);
  },
  write(chunk, encoding, callback) {
    console.log('writable', chunk);
    callback();
  },
});

const { readable, writable } = Duplex.toWeb(duplex);
writable.getWriter().write('hello');

const { value } = await readable.getReader().read();
console.log('readable', value);

const { Duplex } = require('node:stream');

const duplex = Duplex({
  objectMode: true,
  read() {
    this.push('world');
    this.push(null);
  },
  write(chunk, encoding, callback) {
    console.log('writable', chunk);
    callback();
  },
});

const { readable, writable } = Duplex.toWeb(duplex);
writable.getWriter().write('hello');

readable.getReader().read().then((result) => {
  console.log('readable', result.value);
});

stream.addAbortSignal(signal, stream)

• signal <AbortSignal> Сигнал, представляющий возможность отмены
• stream <Stream> | <ReadableStream> | <WritableStream> Поток, к которому нужно привязать сигнал.

Привязывает AbortSignal к потоку для чтения или записи. Это позволяет коду управлять уничтожением потока с помощью AbortController.

Вызов abort для AbortController, соответствующего переданному AbortSignal, будет вести себя так же, как вызов .destroy(new AbortError()) для потока и controller.error(new AbortError()) для веб-потоков.

const fs = require('node:fs');

const controller = new AbortController();
const read = addAbortSignal(
  controller.signal,
  fs.createReadStream(('object.json')),
);
// Later, abort the operation closing the stream
controller.abort(); copy

Или используя AbortSignal с потоком для чтения в качестве асинхронного итерируемого объекта:

const controller = new AbortController();
setTimeout(() => controller.abort(), 10_000); // set a timeout
const stream = addAbortSignal(
  controller.signal,
  fs.createReadStream(('object.json')),
);
(async () => {
  try {
    for await (const chunk of stream) {
      await process(chunk);
    }
  } catch (e) {
    if (e.name === 'AbortError') {
      // The operation was cancelled
    } else {
      throw e;
    }
  }
})(); copy

Или используя AbortSignal с ReadableStream:

const controller = new AbortController();
const rs = new ReadableStream({
  start(controller) {
    controller.enqueue('hello');
    controller.enqueue('world');
    controller.close();
  },
});

addAbortSignal(controller.signal, rs);

finished(rs, (err) => {
  if (err) {
    if (err.name === 'AbortError') {
      // The operation was cancelled
    }
  }
});

const reader = rs.getReader();

reader.read().then(({ value, done }) => {
  console.log(value); // hello
  console.log(done); // false
  controller.abort();
}); copy

stream.getDefaultHighWaterMark(objectMode)

• objectMode <boolean>
• Возвращает: <integer>

Возвращает используемое по умолчанию значение highWaterMark для потоков. По умолчанию это 65536 (64 КиБ) или 16 для objectMode.

stream.setDefaultHighWaterMark(objectMode, value)

• objectMode <boolean>
• value <integer> Значение highWaterMark

Устанавливает используемое по умолчанию значение highWaterMark для потоков.

API для разработчиков потоков

API модуля node:stream разработан так, чтобы упростить реализацию потоков с использованием модели прототипного наследования JavaScript.

Сначала разработчик потока объявляет новый класс JavaScript, расширяющий один из четырёх базовых классов потоков (stream.Writable, stream.Readable, stream.Duplex или stream.Transform), и вызывает соответствующий конструктор родительского класса:

const { Writable } = require('node:stream');

class MyWritable extends Writable {
  constructor({ highWaterMark, ...options }) {
    super({ highWaterMark });
    // ...
  }
} copy

При расширении потоков учитывайте, какие параметры пользователь может и должен передавать, прежде чем передавать их базовому конструктору. Например, если реализация делает предположения относительно параметров autoDestroy и emitClose, не позволяйте пользователю переопределять их. Явно указывайте, какие параметры передаются дальше, вместо того чтобы неявно передавать все параметры.

Затем новый класс потока должен реализовать один или несколько специальных методов, в зависимости от создаваемого типа потока, как показано в таблице ниже:

Код реализации потока никогда не должен вызывать «публичные» методы потока, предназначенные для использования потребителями (как описано в разделе API для потребителей потоков). Это может привести к нежелательным побочным эффектам в коде приложения, использующего поток.

Не переопределяйте публичные методы, такие как write(), end(), cork(), uncork(), read() и destroy(), и не генерируйте внутренние события, такие как 'error', 'data', 'end', 'finish' и 'close', через .emit(). Это может нарушить текущие и будущие инварианты потоков, привести к проблемам с поведением и/или совместимостью с другими потоками и утилитами для работы с потоками, а также не оправдать ожиданий пользователей.

Упрощённое создание

Во многих простых случаях поток можно создать без наследования. Для этого можно напрямую создать экземпляр объекта stream.Writable, stream.Readable, stream.Duplex или stream.Transform и передать соответствующие методы в качестве параметров конструктора.

const { Writable } = require('node:stream');

const myWritable = new Writable({
  construct(callback) {
    // Initialize state and load resources...
  },
  write(chunk, encoding, callback) {
    // ...
  },
  destroy() {
    // Free resources...
  },
}); copy

Реализация записываемого потока

Для реализации потока Writable расширяется класс stream.Writable.

Пользовательские потоки Writable должны вызывать конструктор new stream.Writable([options]) и реализовывать метод writable._write() и/или writable._writev().

new stream.Writable([options])

• options <Object>
  • highWaterMark <number> Размер буфера, при котором вызов stream.write() начинает возвращать false. По умолчанию: 16384 (16 КиБ) или 16 для потоков objectMode.
  • decodeStrings <boolean> Следует ли кодировать значения string, переданные в stream.write(), в значения Buffer (с кодировкой, указанной в вызове stream.write()) перед их передачей в stream._write(). Данные других типов не преобразуются (то есть значения Buffer не декодируются в значения string). Значение false предотвращает преобразование значений string. По умолчанию: true.
  • defaultEncoding <string> Кодировка по умолчанию, используемая, если в качестве аргумента stream.write() не указана кодировка. По умолчанию: 'utf8'.
  • objectMode <boolean> Допустим ли вызов stream.write(anyObj). Если параметр установлен, можно записывать значения JavaScript, отличные от строк, <Buffer>, <TypedArray> или <DataView>, если это поддерживается реализацией потока. По умолчанию: false.
  • emitClose <boolean> Должен ли поток генерировать событие 'close' после уничтожения. По умолчанию: true.
  • write <Function> Реализация метода stream._write().
  • writev <Function> Реализация метода stream._writev().
  • destroy <Function> Реализация метода stream._destroy().
  • final <Function> Реализация метода stream._final().
  • construct <Function> Реализация метода stream._construct().
  • autoDestroy <boolean> Должен ли поток автоматически вызывать .destroy() для себя после завершения. По умолчанию: true.
  • signal <AbortSignal> Сигнал, указывающий на возможную отмену.

const { Writable } = require('node:stream');

class MyWritable extends Writable {
  constructor(options) {
    // Calls the stream.Writable() constructor.
    super(options);
    // ...
  }
} copy

Или при использовании конструкторов в стиле до ES6:

const { Writable } = require('node:stream');
const util = require('node:util');

function MyWritable(options) {
  if (!(this instanceof MyWritable))
    return new MyWritable(options);
  Writable.call(this, options);
}
util.inherits(MyWritable, Writable); copy

Или при использовании упрощённого подхода к созданию конструктора:

const { Writable } = require('node:stream');

const myWritable = new Writable({
  write(chunk, encoding, callback) {
    // ...
  },
  writev(chunks, callback) {
    // ...
  },
}); copy

Вызов abort для AbortController, соответствующего переданному AbortSignal, будет работать так же, как вызов .destroy(new AbortError()) для записываемого потока.

const { Writable } = require('node:stream');

const controller = new AbortController();
const myWritable = new Writable({
  write(chunk, encoding, callback) {
    // ...
  },
  writev(chunks, callback) {
    // ...
  },
  signal: controller.signal,
});
// Later, abort the operation closing the stream
controller.abort(); copy

writable._construct(callback)

• callback <Function> Вызовите эту функцию (при необходимости передав аргумент ошибки), когда инициализация потока завершится.

Метод _construct() НЕЛЬЗЯ вызывать напрямую. Его могут реализовать дочерние классы; в таком случае он будет вызываться только внутренними методами класса Writable.

Эта необязательная функция будет вызвана в следующем тике после возврата конструктора потока, откладывая вызовы _write(), _final() и _destroy() до вызова callback. Это удобно для инициализации состояния или асинхронной инициализации ресурсов до начала использования потока.

const { Writable } = require('node:stream');
const fs = require('node:fs');

class WriteStream extends Writable {
  constructor(filename) {
    super();
    this.filename = filename;
    this.fd = null;
  }
  _construct(callback) {
    fs.open(this.filename, 'w', (err, fd) => {
      if (err) {
        callback(err);
      } else {
        this.fd = fd;
        callback();
      }
    });
  }
  _write(chunk, encoding, callback) {
    fs.write(this.fd, chunk, callback);
  }
  _destroy(err, callback) {
    if (this.fd) {
      fs.close(this.fd, (er) => callback(er || err));
    } else {
      callback(err);
    }
  }
} copy

writable._write(chunk, encoding, callback)

• chunk <Buffer> | <string> | <any> Записываемый Buffer, преобразованный из string, переданного в stream.write(). Если параметр decodeStrings потока имеет значение false или поток работает в режиме объектов, фрагмент не преобразуется и будет иметь тот же вид, что и значение, переданное в stream.write().
• encoding <string> Если фрагмент является строкой, encoding задаёт кодировку символов этой строки. Если фрагмент является Buffer или поток работает в режиме объектов, encoding можно игнорировать.
• callback <Function> Вызовите эту функцию (при необходимости передав аргумент ошибки), когда обработка переданного фрагмента завершится.

Все реализации потоков Writable должны предоставлять метод writable._write() и/или writable._writev() для передачи данных базовому ресурсу.

Потоки Transform реализуют метод writable._write() самостоятельно.

Эту функцию НЕЛЬЗЯ вызывать непосредственно из кода приложения. Её должны реализовывать дочерние классы, а вызываться она должна только внутренними методами класса Writable.

Функцию callback необходимо вызвать синхронно внутри writable._write() или асинхронно (то есть в другом тике), чтобы сообщить об успешном завершении записи или об ошибке. Первым аргументом, передаваемым в callback, должен быть объект Error, если вызов завершился ошибкой, или null, если запись прошла успешно.

Все вызовы writable.write(), произошедшие между вызовом writable._write() и вызовом callback, приведут к буферизации записанных данных. При вызове callback поток может сгенерировать событие 'drain'. Если реализация потока способна обрабатывать несколько фрагментов данных одновременно, следует реализовать метод writable._writev().

Если свойству decodeStrings явно присвоено значение false в параметрах конструктора, то chunk останется тем же объектом, который передан в .write(), и может быть строкой, а не Buffer. Это необходимо для поддержки реализаций с оптимизированной обработкой строковых данных в определённых кодировках. В таком случае аргумент encoding указывает кодировку символов строки. В противном случае аргумент encoding можно безопасно игнорировать.

Перед именем метода writable._write() стоит символ подчёркивания, поскольку он является внутренним для определяющего его класса и никогда не должен вызываться напрямую пользовательскими программами.

writable._writev(chunks, callback)

• chunks <Object[]> Записываемые данные. Значение представляет собой массив объектов <Object>, каждый из которых соответствует отдельному фрагменту данных для записи. Эти объекты имеют следующие свойства:
  • chunk <Buffer> | <string> Экземпляр буфера или строка с записываемыми данными. Значение chunk будет строкой, если Writable создан с параметром decodeStrings, имеющим значение false, и в write() передана строка.
  • encoding <string> Кодировка символов значения chunk. Если chunk — это Buffer, значение encoding будет 'buffer'.
• callback <Function> Функция обратного вызова (при необходимости с аргументом ошибки), которая вызывается после завершения обработки переданных фрагментов.

Эту функцию НЕЛЬЗЯ вызывать непосредственно из кода приложения. Её должны реализовывать дочерние классы, а вызываться она должна только внутренними методами класса Writable.

Метод writable._writev() можно реализовать дополнительно к writable._write() или вместо него в реализациях потоков, способных обрабатывать несколько фрагментов данных одновременно. Если метод реализован и имеются данные, буферизованные в результате предыдущих операций записи, вместо _write() будет вызван _writev().

Перед именем метода writable._writev() стоит символ подчёркивания, поскольку он является внутренним для определяющего его класса и никогда не должен вызываться напрямую пользовательскими программами.

writable._destroy(err, callback)

• err <Error> Возможная ошибка.
• callback <Function> Функция обратного вызова, принимающая необязательный аргумент ошибки.

Метод _destroy() вызывается методом writable.destroy(). Его можно переопределить в дочерних классах, но его нельзя вызывать напрямую.

writable._final(callback)

• callback <Function> Вызовите эту функцию (при необходимости передав аргумент ошибки), когда запись всех оставшихся данных завершится.

Метод _final() нельзя вызывать напрямую. Его могут реализовать дочерние классы; в таком случае он будет вызываться только внутренними методами класса Writable.

Эта необязательная функция вызывается перед закрытием потока и откладывает событие 'finish' до вызова callback. Это удобно для закрытия ресурсов или записи буферизованных данных перед завершением потока.

Ошибки при записи

Об ошибках, возникающих при обработке методов writable._write(), writable._writev() и writable._final(), необходимо сообщать, вызывая функцию обратного вызова и передавая ошибку первым аргументом. Выбрасывание Error из этих методов или ручная генерация события 'error' приводит к неопределённому поведению.

Если поток Readable передаёт данные в поток Writable, а Writable генерирует ошибку, поток Readable перестанет получать данные из источника.

const { Writable } = require('node:stream');

const myWritable = new Writable({
  write(chunk, encoding, callback) {
    if (chunk.toString().indexOf('a') >= 0) {
      callback(new Error('chunk is invalid'));
    } else {
      callback();
    }
  },
}); copy

Пример записываемого потока

Ниже показана довольно упрощённая (и в некоторой степени бесполезная) реализация пользовательского потока Writable. Хотя конкретный экземпляр потока Writable не представляет особой практической ценности, этот пример демонстрирует все необходимые элементы пользовательского экземпляра потока Writable:

const { Writable } = require('node:stream');

class MyWritable extends Writable {
  _write(chunk, encoding, callback) {
    if (chunk.toString().indexOf('a') >= 0) {
      callback(new Error('chunk is invalid'));
    } else {
      callback();
    }
  }
} copy

Декодирование буферов в записываемом потоке

Декодирование буферов — распространённая задача, например при использовании преобразователей, на вход которых подаётся строка. При использовании многобайтовых кодировок символов, таких как UTF-8, это непростая задача. В следующем примере показано, как декодировать многобайтовые строки с помощью StringDecoder и Writable.

const { Writable } = require('node:stream');
const { StringDecoder } = require('node:string_decoder');

class StringWritable extends Writable {
  constructor(options) {
    super(options);
    this._decoder = new StringDecoder(options && options.defaultEncoding);
    this.data = '';
  }
  _write(chunk, encoding, callback) {
    if (encoding === 'buffer') {
      chunk = this._decoder.write(chunk);
    }
    this.data += chunk;
    callback();
  }
  _final(callback) {
    this.data += this._decoder.end();
    callback();
  }
}

const euro = [[0xE2, 0x82], [0xAC]].map(Buffer.from);
const w = new StringWritable();

w.write('currency: ');
w.write(euro[0]);
w.end(euro[1]);

console.log(w.data); // currency: € copy

Реализация читаемого потока

Для реализации потока Readable расширяется класс stream.Readable.

Пользовательские потоки Readable должны вызывать конструктор new stream.Readable([options]) и реализовывать метод readable._read().

new stream.Readable([options])

• options <Object>
  • highWaterMark <number> Максимальное число байтов, хранящихся во внутреннем буфере до прекращения чтения базового ресурса. По умолчанию: 16384 (16 КиБ) или 16 для потоков objectMode.
  • encoding <string> Если параметр указан, буферы декодируются в строки с использованием указанной кодировки. По умолчанию: null.
  • objectMode <boolean> Должен ли поток работать как поток объектов. Это означает, что stream.read(n) возвращает одно значение, а не Buffer размером n. По умолчанию: false.
  • emitClose <boolean> Должен ли поток генерировать событие 'close' после уничтожения. По умолчанию: true.
  • read <Function> Реализация метода stream._read().
  • destroy <Function> Реализация метода stream._destroy().
  • construct <Function> Реализация метода stream._construct().
  • autoDestroy <boolean> Должен ли поток автоматически вызывать .destroy() для себя после завершения. По умолчанию: true.
  • signal <AbortSignal> Сигнал, указывающий на возможную отмену.

const { Readable } = require('node:stream');

class MyReadable extends Readable {
  constructor(options) {
    // Calls the stream.Readable(options) constructor.
    super(options);
    // ...
  }
} copy

Или при использовании конструкторов в стиле до ES6:

const { Readable } = require('node:stream');
const util = require('node:util');

function MyReadable(options) {
  if (!(this instanceof MyReadable))
    return new MyReadable(options);
  Readable.call(this, options);
}
util.inherits(MyReadable, Readable); copy

Или при использовании упрощённого подхода к созданию конструктора:

const { Readable } = require('node:stream');

const myReadable = new Readable({
  read(size) {
    // ...
  },
}); copy

Вызов abort для AbortController, соответствующего переданному AbortSignal, будет работать так же, как вызов .destroy(new AbortError()) для созданного читаемого потока.

const { Readable } = require('node:stream');
const controller = new AbortController();
const read = new Readable({
  read(size) {
    // ...
  },
  signal: controller.signal,
});
// Later, abort the operation closing the stream
controller.abort(); copy

readable._construct(callback)

• callback <Function> Вызовите эту функцию (при необходимости передав аргумент ошибки), когда инициализация потока завершится.

Метод _construct() НЕЛЬЗЯ вызывать напрямую. Его могут реализовать дочерние классы; в таком случае он будет вызываться только внутренними методами класса Readable.

Эта необязательная функция будет запланирована конструктором потока на следующий тик, откладывая вызовы _read() и _destroy() до вызова callback. Это удобно для инициализации состояния или асинхронной инициализации ресурсов до начала использования потока.

const { Readable } = require('node:stream');
const fs = require('node:fs');

class ReadStream extends Readable {
  constructor(filename) {
    super();
    this.filename = filename;
    this.fd = null;
  }
  _construct(callback) {
    fs.open(this.filename, (err, fd) => {
      if (err) {
        callback(err);
      } else {
        this.fd = fd;
        callback();
      }
    });
  }
  _read(n) {
    const buf = Buffer.alloc(n);
    fs.read(this.fd, buf, 0, n, null, (err, bytesRead) => {
      if (err) {
        this.destroy(err);
      } else {
        this.push(bytesRead > 0 ? buf.slice(0, bytesRead) : null);
      }
    });
  }
  _destroy(err, callback) {
    if (this.fd) {
      fs.close(this.fd, (er) => callback(er || err));
    } else {
      callback(err);
    }
  }
} copy

readable._read(size)

• size <number> Число байтов для асинхронного чтения

Эту функцию НЕЛЬЗЯ вызывать непосредственно из кода приложения. Её должны реализовывать дочерние классы, а вызываться она должна только внутренними методами класса Readable.

Все реализации потоков Readable должны реализовывать метод readable._read() для получения данных из базового ресурса.

При вызове readable._read(), если ресурс предоставляет данные, реализация должна начать помещать их в очередь чтения с помощью метода this.push(dataChunk). _read() будет вызван повторно после каждого вызова this.push(dataChunk), когда поток будет готов принять новые данные. _read() может продолжать читать данные из ресурса и помещать их в очередь, пока readable.push() не вернёт false. Возобновлять помещение дополнительных данных в очередь следует только после повторного вызова _read(), когда чтение было приостановлено.

После вызова метода readable._read() он не будет вызван снова, пока через метод readable.push() не будут помещены новые данные. Пустые данные, например пустые буферы и строки, не приводят к вызову readable._read().

Аргумент size носит рекомендательный характер. Реализации, в которых операция чтения возвращает данные за один вызов, могут использовать аргумент size, чтобы определить, сколько данных получить. Другие реализации могут игнорировать этот аргумент и предоставлять данные по мере их поступления. Нет необходимости ждать, пока станут доступны size байт, прежде чем вызывать stream.push(chunk).

Перед именем метода readable._read() стоит символ подчёркивания, поскольку он является внутренним для определяющего его класса и никогда не должен вызываться напрямую пользовательскими программами.

readable._destroy(err, callback)

• err <Error> Возможная ошибка.
• callback <Function> Функция обратного вызова, принимающая необязательный аргумент ошибки.

Метод _destroy() вызывается методом readable.destroy(). Его можно переопределить в дочерних классах, но его нельзя вызывать напрямую.

readable.push(chunk[, encoding])

• chunk <Buffer> | <TypedArray> | <DataView> | <string> | <null> | <any> Фрагмент данных, добавляемый в очередь чтения. Для потоков, не работающих в режиме объектов, chunk должен иметь тип <string>, <Buffer>, <TypedArray> или <DataView>. Для потоков в режиме объектов chunk может иметь любое значение JavaScript.
• encoding <string> Кодировка строковых фрагментов. Должна быть допустимой кодировкой Buffer, например 'utf8' или 'ascii'.
• Возвращает: <boolean> true, если можно продолжать добавлять фрагменты данных; в противном случае — false.

Если chunk — это <Buffer>, <TypedArray>, <DataView> или <string>, chunk данных будет добавлено во внутреннюю очередь, откуда их смогут получить пользователи потока. Передача chunk в качестве null обозначает конец потока (EOF), после которого записывать данные больше нельзя.

Когда Readable работает в режиме паузы, данные, добавленные с помощью readable.push(), можно прочитать, вызвав метод readable.read() при возникновении события 'readable'.

Когда Readable работает в потоковом режиме, данные, добавленные с помощью readable.push(), будут переданы посредством генерации события 'data'.

Метод readable.push() разработан с максимальной гибкостью. Например, при обёртывании низкоуровневого источника, предоставляющего механизм паузы/возобновления и обратный вызов для данных, этот источник можно обернуть в пользовательский экземпляр Readable:

// `_source` is an object with readStop() and readStart() methods,
// and an `ondata` member that gets called when it has data, and
// an `onend` member that gets called when the data is over.

class SourceWrapper extends Readable {
  constructor(options) {
    super(options);

    this._source = getLowLevelSourceObject();

    // Every time there's data, push it into the internal buffer.
    this._source.ondata = (chunk) => {
      // If push() returns false, then stop reading from source.
      if (!this.push(chunk))
        this._source.readStop();
    };

    // When the source ends, push the EOF-signaling `null` chunk.
    this._source.onend = () => {
      this.push(null);
    };
  }
  // _read() will be called when the stream wants to pull more data in.
  // The advisory size argument is ignored in this case.
  _read(size) {
    this._source.readStart();
  }
} copy

Метод readable.push() используется для добавления содержимого во внутренний буфер. Его работу может запускать метод readable._read().

Для потоков, не работающих в режиме объектов, если параметр chunk метода readable.push() имеет значение undefined, он будет обработан как пустая строка или буфер. Подробнее см. в разделе readable.push('').

Ошибки при чтении

Об ошибках, возникающих при обработке метода readable._read(), необходимо сообщать с помощью метода readable.destroy(err). Генерация Error внутри readable._read() или ручная генерация события 'error' приводит к неопределённому поведению.

const { Readable } = require('node:stream');

const myReadable = new Readable({
  read(size) {
    const err = checkSomeErrorCondition();
    if (err) {
      this.destroy(err);
    } else {
      // Do some work.
    }
  },
}); copy

Пример потока-счётчика

Ниже приведён простой пример потока Readable, который по возрастанию выдаёт числа от 1 до 1 000 000, а затем завершает работу.

const { Readable } = require('node:stream');

class Counter extends Readable {
  constructor(opt) {
    super(opt);
    this._max = 1000000;
    this._index = 1;
  }

  _read() {
    const i = this._index++;
    if (i > this._max)
      this.push(null);
    else {
      const str = String(i);
      const buf = Buffer.from(str, 'ascii');
      this.push(buf);
    }
  }
} copy

Реализация дуплексного потока

Поток Duplex реализует интерфейсы Readable и Writable, например TCP-соединение через сокет.

Поскольку JavaScript не поддерживает множественное наследование, для реализации потока Duplex расширяется класс stream.Duplex (вместо расширения классов stream.Readable и stream.Writable).

Класс stream.Duplex прототипно наследуется от stream.Readable и паразитически — от stream.Writable, однако instanceof корректно работает с обоими базовыми классами благодаря переопределению Symbol.hasInstance в stream.Writable.

Пользовательские потоки Duplex должны вызывать конструктор new stream.Duplex([options]) и реализовывать оба метода: readable._read() и writable._write().

new stream.Duplex(options)

• options <Object> Передаётся конструкторам Writable и Readable. Также содержит следующие поля:
  • allowHalfOpen <boolean> Если задано значение false, записываемая сторона потока автоматически завершит работу после завершения читаемой стороны. По умолчанию: true.
  • readable <boolean> Определяет, должна ли Duplex быть доступна для чтения. По умолчанию: true.
  • writable <boolean> Определяет, должна ли Duplex быть доступна для записи. По умолчанию: true.
  • readableObjectMode <boolean> Задаёт objectMode для читаемой стороны потока. Не действует, если objectMode имеет значение true. По умолчанию: false.
  • writableObjectMode <boolean> Задаёт objectMode для записываемой стороны потока. Не действует, если objectMode имеет значение true. По умолчанию: false.
  • readableHighWaterMark <number> Задаёт highWaterMark для читаемой стороны потока. Не действует, если указан highWaterMark.
  • writableHighWaterMark <number> Задаёт highWaterMark для записываемой стороны потока. Не действует, если указан highWaterMark.

const { Duplex } = require('node:stream');

class MyDuplex extends Duplex {
  constructor(options) {
    super(options);
    // ...
  }
} copy

Или при использовании конструкторов в стиле до ES6:

const { Duplex } = require('node:stream');
const util = require('node:util');

function MyDuplex(options) {
  if (!(this instanceof MyDuplex))
    return new MyDuplex(options);
  Duplex.call(this, options);
}
util.inherits(MyDuplex, Duplex); copy

Или при использовании упрощённого способа создания конструктора:

const { Duplex } = require('node:stream');

const myDuplex = new Duplex({
  read(size) {
    // ...
  },
  write(chunk, encoding, callback) {
    // ...
  },
}); copy

При использовании конвейера:

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

pipeline(
  fs.createReadStream('object.json')
    .setEncoding('utf8'),
  new Transform({
    decodeStrings: false, // Accept string input rather than Buffers
    construct(callback) {
      this.data = '';
      callback();
    },
    transform(chunk, encoding, callback) {
      this.data += chunk;
      callback();
    },
    flush(callback) {
      try {
        // Make sure is valid json.
        JSON.parse(this.data);
        this.push(this.data);
        callback();
      } catch (err) {
        callback(err);
      }
    },
  }),
  fs.createWriteStream('valid-object.json'),
  (err) => {
    if (err) {
      console.error('failed', err);
    } else {
      console.log('completed');
    }
  },
); copy

Пример дуплексного потока

Ниже приведён простой пример потока Duplex, который оборачивает гипотетический низкоуровневый объект-источник, в который можно записывать данные и из которого их можно читать, хотя его API несовместим с потоками Node.js. В примере показан простой поток Duplex, который буферизует входящие данные, записанные через интерфейс Writable, а затем возвращает их через интерфейс Readable.

const { Duplex } = require('node:stream');
const kSource = Symbol('source');

class MyDuplex extends Duplex {
  constructor(source, options) {
    super(options);
    this[kSource] = source;
  }

  _write(chunk, encoding, callback) {
    // The underlying source only deals with strings.
    if (Buffer.isBuffer(chunk))
      chunk = chunk.toString();
    this[kSource].writeSomeData(chunk);
    callback();
  }

  _read(size) {
    this[kSource].fetchSomeData(size, (data, encoding) => {
      this.push(Buffer.from(data, encoding));
    });
  }
} copy

Важнейшая особенность потока Duplex состоит в том, что стороны Readable и Writable работают независимо друг от друга, несмотря на то что они сосуществуют в одном экземпляре объекта.

Дуплексные потоки в режиме объектов

Для потоков Duplex параметр objectMode можно задать отдельно для стороны Readable или Writable с помощью параметров readableObjectMode и writableObjectMode соответственно.

Например, в следующем примере создаётся новый поток Transform (тип потока Duplex) со стороной Writable в режиме объектов, которая принимает числа JavaScript и преобразует их в шестнадцатеричные строки на стороне Readable.

const { Transform } = require('node:stream');

// All Transform streams are also Duplex Streams.
const myTransform = new Transform({
  writableObjectMode: true,

  transform(chunk, encoding, callback) {
    // Coerce the chunk to a number if necessary.
    chunk |= 0;

    // Transform the chunk into something else.
    const data = chunk.toString(16);

    // Push the data onto the readable queue.
    callback(null, '0'.repeat(data.length % 2) + data);
  },
});

myTransform.setEncoding('ascii');
myTransform.on('data', (chunk) => console.log(chunk));

myTransform.write(1);
// Prints: 01
myTransform.write(10);
// Prints: 0a
myTransform.write(100);
// Prints: 64 copy

Реализация преобразующего потока

Поток Transform — это поток Duplex, в котором выходные данные определённым образом вычисляются на основе входных. Примерами служат потоки zlib и crypto, сжимающие, шифрующие или расшифровывающие данные.

Выходные данные не обязательно должны иметь тот же размер, содержать то же количество фрагментов или поступать одновременно с входными. Например, поток Hash выдаёт только один фрагмент данных, когда входные данные заканчиваются. Поток zlib может выдавать данные, размер которых значительно меньше или больше размера входных данных.

Для реализации потока Transform расширяется класс stream.Transform.

Класс stream.Transform прототипно наследуется от stream.Duplex и реализует собственные версии методов writable._write() и readable._read(). Пользовательские реализации Transform должны реализовать метод transform._transform() и могут также реализовать метод transform._flush().

При использовании потоков Transform следует учитывать, что записываемые в поток данные могут приостановить его сторону Writable, если выходные данные на стороне Readable не считываются.

new stream.Transform([options])

• options <Object> Передаётся конструкторам Writable и Readable. Также содержит следующие поля:
  • transform <Function> Реализация метода stream._transform().
  • flush <Function> Реализация метода stream._flush().

const { Transform } = require('node:stream');

class MyTransform extends Transform {
  constructor(options) {
    super(options);
    // ...
  }
} copy

Или при использовании конструкторов в стиле до ES6:

const { Transform } = require('node:stream');
const util = require('node:util');

function MyTransform(options) {
  if (!(this instanceof MyTransform))
    return new MyTransform(options);
  Transform.call(this, options);
}
util.inherits(MyTransform, Transform); copy

Или при использовании упрощённого способа создания конструктора:

const { Transform } = require('node:stream');

const myTransform = new Transform({
  transform(chunk, encoding, callback) {
    // ...
  },
}); copy

Событие: 'end'

Событие 'end' относится к классу stream.Readable. Событие 'end' возникает после вывода всех данных, то есть после вызова обратного вызова в transform._flush(). При возникновении ошибки событие 'end' генерироваться не должно.

Событие: 'finish'

Событие 'finish' относится к классу stream.Writable. Событие 'finish' возникает после вызова stream.end() и обработки всех фрагментов методом stream._transform(). При возникновении ошибки событие 'finish' генерироваться не должно.

transform._flush(callback)

• callback <Function> Функция обратного вызова (необязательно с аргументом ошибки и данными), вызываемая после сброса оставшихся данных.

ЭТУ функцию нельзя вызывать непосредственно из кода приложения. Её должны реализовать дочерние классы; вызывать её могут только внутренние методы класса Readable.

В некоторых случаях при преобразовании в конце потока требуется выдать дополнительные данные. Например, поток сжатия zlib хранит некоторое внутреннее состояние, необходимое для оптимального сжатия выходных данных. Однако при завершении потока эти дополнительные данные необходимо сбросить, чтобы сжатые данные были полными.

Пользовательские реализации Transform могут реализовать метод transform._flush(). Он вызывается, когда больше нет записанных данных для обработки, но до возникновения события 'end', сигнализирующего о завершении потока Readable.

В реализации transform._flush() метод transform.push() можно вызывать ноль или более раз — в зависимости от необходимости. После завершения операции сброса необходимо вызвать функцию callback.

Перед именем метода transform._flush() стоит подчёркивание, поскольку он является внутренним для определяющего его класса и никогда не должен вызываться напрямую пользовательскими программами.

transform._transform(chunk, encoding, callback)

• chunk <Buffer> | <string> | <any> Фрагмент Buffer, который нужно преобразовать; он конвертируется из string, переданного в stream.write(). Если параметр decodeStrings потока имеет значение false или поток работает в режиме объектов, фрагмент не будет преобразован и будет иметь то же значение, что и переданное в stream.write().
• encoding <string> Если фрагмент является строкой, это тип кодировки. Если фрагмент — буфер, здесь указывается специальное значение 'buffer'. В этом случае его следует игнорировать.
• callback <Function> Функция обратного вызова (необязательно с аргументом ошибки и данными), вызываемая после обработки переданного chunk.

ЭТУ функцию нельзя вызывать непосредственно из кода приложения. Её должны реализовать дочерние классы; вызывать её могут только внутренние методы класса Readable.

Все реализации потока Transform должны предоставлять метод _transform() для приёма входных данных и формирования выходных. Реализация transform._transform() обрабатывает записанные байты, вычисляет результат, а затем передаёт его читаемой части с помощью метода transform.push().

Метод transform.push() можно вызывать ноль или более раз для формирования выходных данных из одного входного фрагмента — в зависимости от объёма результата, который необходимо выдать.

Для отдельного фрагмента входных данных выходные данные могут не формироваться.

Функцию callback необходимо вызывать только после полного потребления текущего фрагмента. Если при обработке входных данных произошла ошибка, первым аргументом, передаваемым в callback, должен быть объект Error; в противном случае — null. Если в callback передан второй аргумент, он будет передан методу transform.push(), но только если первый аргумент является ложным значением. Иными словами, следующий код эквивалентен:

transform.prototype._transform = function(data, encoding, callback) {
  this.push(data);
  callback();
};

transform.prototype._transform = function(data, encoding, callback) {
  callback(null, data);
}; copy

Перед именем метода transform._transform() стоит подчёркивание, поскольку он является внутренним для определяющего его класса и никогда не должен вызываться напрямую пользовательскими программами.

Метод transform._transform() никогда не вызывается параллельно; потоки используют механизм очереди, и для получения следующего фрагмента необходимо вызвать callback — синхронно или асинхронно.

Класс: stream.PassThrough

Класс stream.PassThrough представляет собой простую реализацию потока Transform, который передаёт входные байты на выход без изменений. В основном он предназначен для примеров и тестирования, но в некоторых случаях stream.PassThrough полезен как строительный блок для создания новых типов потоков.

Дополнительные примечания

Совместимость потоков с асинхронными генераторами и асинхронными итераторами

Благодаря поддержке асинхронных генераторов и итераторов в JavaScript асинхронные генераторы теперь фактически являются первоклассной языковой конструкцией для работы с потоками.

Ниже приведены распространённые сценарии взаимодействия потоков Node.js с асинхронными генераторами и асинхронными итераторами.

Чтение данных из читаемых потоков с помощью асинхронных итераторов

(async function() {
  for await (const chunk of readable) {
    console.log(chunk);
  }
})(); copy

Асинхронные итераторы регистрируют в потоке постоянный обработчик ошибок, чтобы предотвратить необработанные ошибки, возникающие после уничтожения потока.

Создание читаемых потоков с помощью асинхронных генераторов

Читаемый поток Node.js можно создать из асинхронного генератора с помощью служебного метода Readable.from():

const { Readable } = require('node:stream');

const ac = new AbortController();
const signal = ac.signal;

async function * generate() {
  yield 'a';
  await someLongRunningFn({ signal });
  yield 'b';
  yield 'c';
}

const readable = Readable.from(generate());
readable.on('close', () => {
  ac.abort();
});

readable.on('data', (chunk) => {
  console.log(chunk);
}); copy

Передача данных из асинхронных итераторов в записываемые потоки

При записи данных из асинхронного итератора в записываемый поток необходимо правильно обрабатывать обратное давление и ошибки. Метод stream.pipeline() берёт на себя обработку обратного давления и связанных с ним ошибок:

const fs = require('node:fs');
const { pipeline } = require('node:stream');
const { pipeline: pipelinePromise } = require('node:stream/promises');

const writable = fs.createWriteStream('./file');

const ac = new AbortController();
const signal = ac.signal;

const iterator = createIterator({ signal });

// Callback Pattern
pipeline(iterator, writable, (err, value) => {
  if (err) {
    console.error(err);
  } else {
    console.log(value, 'value returned');
  }
}).on('close', () => {
  ac.abort();
});

// Promise Pattern
pipelinePromise(iterator, writable)
  .then((value) => {
    console.log(value, 'value returned');
  })
  .catch((err) => {
    console.error(err);
    ac.abort();
  }); copy

Совместимость со старыми версиями Node.js

До Node.js 0.10 интерфейс потока Readable был проще, но при этом менее мощным и менее полезным.

• Вместо ожидания вызовов метода stream.read() события 'data' начинали генерироваться сразу. Приложениям, которым требовалось выполнить некоторую работу, чтобы решить, как обрабатывать данные, приходилось сохранять прочитанные данные в буферах, чтобы они не терялись.
• Метод stream.pause() носил рекомендательный, а не обязательный характер. Это означало, что необходимо было быть готовым получать события 'data' даже когда поток находился в состоянии паузы.

В Node.js 0.10 был добавлен класс Readable. Для обратной совместимости со старыми программами Node.js потоки Readable переходят в «потоковый режим» при добавлении обработчика события 'data' или вызове метода stream.resume(). В результате даже при отказе от нового метода stream.read() и события 'readable' больше не нужно беспокоиться о потере фрагментов события 'data'.

Хотя большинство приложений продолжат работать как обычно, в следующих случаях возникает особая ситуация:

• Обработчик события 'data' не добавлен.
• Метод stream.resume() ни разу не вызывается.
• Поток не подключён к какой-либо записываемой стороне назначения.

Например, рассмотрим следующий код:

// WARNING!  BROKEN!
net.createServer((socket) => {

  // We add an 'end' listener, but never consume the data.
  socket.on('end', () => {
    // It will never get here.
    socket.end('The message was received but was not processed.\n');
  });

}).listen(1337); copy

До Node.js 0.10 входящие данные сообщения просто отбрасывались. Однако в Node.js 0.10 и более поздних версиях сокет остаётся приостановленным навсегда.

В этой ситуации можно вызвать метод stream.resume(), чтобы возобновить передачу данных:

// Workaround.
net.createServer((socket) => {
  socket.on('end', () => {
    socket.end('The message was received but was not processed.\n');
  });

  // Start the flow of data, discarding it.
  socket.resume();
}).listen(1337); copy

Помимо перехода новых потоков Readable в потоковый режим, потоки в стиле до версии 0.10 можно обернуть в класс Readable с помощью метода readable.wrap().

readable.read(0)

В некоторых случаях необходимо запустить обновление базовых механизмов читаемого потока, не потребляя при этом данные. В таких ситуациях можно вызвать readable.read(0), который всегда возвращает null.

Если внутренний буфер чтения заполнен меньше чем до highWaterMark и в данный момент поток не читает данные, вызов stream.read(0) запустит низкоуровневый вызов stream._read().

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

readable.push('')

Использовать readable.push('') не рекомендуется.

Добавление строки <string>, <Buffer>, <TypedArray> или <DataView> нулевой длины в поток, не работающий в режиме объектов, имеет интересный побочный эффект. Поскольку это всё же вызов readable.push(), он завершит процесс чтения. Однако, поскольку аргументом является пустая строка, в читаемый буфер не добавляются данные, и пользователю нечего потреблять.

Несоответствие highWaterMark после вызова readable.setEncoding()

Использование readable.setEncoding() изменяет поведение highWaterMark в режиме, отличном от режима объектов.

Обычно размер текущего буфера сравнивается с highWaterMark в байтах. Однако после вызова setEncoding() функция сравнения начинает измерять размер буфера в символах.

В обычных случаях с latin1 или ascii проблем не возникает. Тем не менее при работе со строками, которые могут содержать многобайтовые символы, рекомендуется учитывать это поведение.