Поток

Поток — это абстрактный интерфейс для работы с потоковыми данными в Node.js. Модуль node:stream предоставляет API для реализации интерфейса потока.

Node.js предоставляет множество объектов потоков. Например, запрос к HTTP-серверу и process.stdout являются экземплярами потоков.

Потоки могут быть читаемыми, записываемыми или поддерживать обе операции. Все потоки являются экземплярами EventEmitter.

Чтобы получить доступ к модулю node:stream:

const stream = require('node:stream');
jscopy

Модуль node:stream полезен для создания новых типов экземпляров потоков. Обычно для работы с потоками не требуется использовать модуль node:stream.

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

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

Типы потоков

В 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 промисов потоков

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

stream.pipeline(streams[, options])

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

• streams <Stream[]> | <AsyncIterable[]> | <Function[]>
• source <Stream> | <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>
  • cleanup <boolean> | <undefined> Если true, удаляет обработчики событий, зарегистрированные этой функцией, до выполнения промиса. По умолчанию: false.
• Возвращает: <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 также предоставляет вариант с функцией обратного вызова.

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

await finished(rs, { cleanup: true });
mjscopy

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

Все потоки, созданные 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
jscopy

Потоки 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-сокеты
• стандартный ввод дочернего процесса
• 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');
jscopy

Класс: 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);
    }
  }
}
jscopy

Событие: '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');
jscopy

Событие: '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);
jscopy

Событие: '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);
jscopy

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
cjscopy

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

const myStream = new Writable();

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

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

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

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

После вызова 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
cjscopy

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!
jscopy

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());
jscopy

Если метод 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();
});
jscopy

См. также: 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 по умолчанию приостановлены, пока их не подключат с помощью '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.');
});
jscopy

Поток 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 может переключиться обратно в приостановленный режим одним из следующих способов:

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

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

Для обеспечения обратной совместимости удаление обработчиков события 'data' не приводит к автоматической приостановке потока. Кроме того, если имеются назначения pipe, вызов 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.
jscopy

Пока 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.`);
});
jscopy

Событие: '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.');
});
jscopy

Событие: '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);
  }
});
jscopy

Если достигнут конец потока, вызов 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');
});
jscopy

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

$ node test.js
readable: null
end
consolecopy

В некоторых случаях добавление слушателя события '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
jscopy

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);
});
jscopy

Метод 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);
jscopy

К одному потоку 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);
jscopy

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

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

Важно учитывать, что если во время обработки поток 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.');
});
jscopy

Каждый вызов 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('');
});
jscopy

Поток 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.');
  });
jscopy

Метод 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);
});
jscopy

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);
jscopy

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;
    }
  }
}
jscopy

В отличие от 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.
});
jscopy

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);
jscopy

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

readable[Symbol.for('Stream.toAsyncStreamable')]()

• Возвращает: <AsyncIterable> Объект AsyncIterable<Uint8Array[]>, выдающий пакетные фрагменты из потока.

Если включён флаг --experimental-stream-iter, потоки Readable реализуют протокол Stream.toAsyncStreamable, что позволяет эффективно считывать их с помощью API stream/iter.

Это предоставляет пакетный асинхронный итератор, который извлекает данные из внутреннего буфера потока пакетами Uint8Array[], снижая средние затраты на Promise для каждого фрагмента по сравнению со стандартным путём Symbol.asyncIterator. В потоках с байтовым режимом фрагменты выдаются напрямую в виде экземпляров Buffer (которые являются подклассами Uint8Array). В потоках в режиме объектов или потоках с заданной кодировкой каждый фрагмент перед пакетной обработкой приводится к Uint8Array.

Возвращаемый итератор помечен как проверенный источник, поэтому from() передаёт его дальше без дополнительной нормализации.

import { Readable } from 'node:stream';
import { text, from } from 'node:stream/iter';

const readable = new Readable({
  read() { this.push('hello'); this.push(null); },
});

// Readable is automatically consumed via toAsyncStreamable
console.log(await text(from(readable))); // 'hello'

const { Readable } = require('node:stream');
const { text, from } = require('node:stream/iter');

async function run() {
  const readable = new Readable({
    read() { this.push('hello'); this.push(null); },
  });

  console.log(await text(from(readable))); // 'hello'
}

run().catch(console.error);

Если флаг --experimental-stream-iter не включён, вызов этого метода выбрасывает ERR_STREAM_ITER_MISSING_FLAG.

readable[Symbol.asyncDispose]()

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

readable.compose(stream[, options])

• stream <Writable> | <Duplex> | <WritableStream> | <TransformStream> | <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(['text passed through', 'composed stream']).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // prints ['text', 'passed', 'through', 'composed', 'stream']
mjscopy

readable.compose(s) эквивалентен stream.compose(readable, s).

Этот метод также позволяет передать <AbortSignal>, при прерывании которого составной поток будет уничтожен.

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

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();
jscopy

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 возвращает Promise, этот Promise будет 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.
}
mjscopy

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 возвращает промис, этот промис будет awaited.

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);
}
mjscopy

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 возвращает промис, этот промис будет awaited.

Этот метод отличается от циклов 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
mjscopy

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]

const resolver = new Resolver();

// 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();
mjscopy

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
mjscopy

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
mjscopy

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
mjscopy

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);
}
mjscopy

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]
mjscopy

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]
mjscopy

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);
mjscopy

Функция-редьюсер перебирает поток поэлементно, то есть параметр 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);
mjscopy

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

Класс: stream.Duplex

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

Примеры двунаправленных потоков Duplex:

• TCP-сокеты
• потоки zlib
• криптографические потоки

duplex.allowHalfOpen

• Тип: <boolean>

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

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

Класс: stream.Transform

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

Примеры потоков преобразования Transform:

• потоки zlib
• криптографические потоки

transform.destroy([error])

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

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

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

stream.duplexPair([options])

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

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

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

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

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

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

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

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

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

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

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

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

stream.pipeline(streams, callback)

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

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

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.');
    }
  },
);
jscopy

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!!!');
    }
  });
});
jscopy

stream.compose(...streams)

• streams <Stream[]> | <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'
mjscopy

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'
mjscopy

Для удобства метод readable.compose(stream) доступен для потоков <Readable> и <Duplex> как оболочка для этой функции.

stream.isErrored(stream)

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

Возвращает, произошла ли в потоке ошибка.

stream.isReadable(stream)

• stream <Readable> | <Duplex> | <ReadableStream>
• Возвращает: <boolean> | <null> — возвращает null только если stream не является допустимым Readable, Duplex или ReadableStream.

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

stream.isWritable(stream)

• stream <Writable> | <Duplex> | <WritableStream>
• Возвращает: <boolean> | <null> — возвращает null только если stream не является допустимым Writable, Duplex или WritableStream.

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

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);
});
jscopy

Вызов 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
]);
jscopy

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.Readable.toWeb(streamReadable[, options])

• streamReadable <stream.Readable>
• options <Object>
  • strategy <Object>
    • highWaterMark <number> Максимальный размер внутренней очереди (созданного ReadableStream), после которого при чтении из указанного stream.Readable применяется обратное давление. Если значение не задано, оно будет взято из указанного stream.Readable.
    • size <Function> Функция, определяющая размер переданного фрагмента данных. Если значение не задано, размер всех фрагментов будет равен 1.
      • chunk <any>
      • Возвращает: <number>
  • type <string> Задаёт тип создаваемого ReadableStream. Должно быть значение 'bytes' или undefined.
• Возвращает: <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> | <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
]);
jscopy

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[, options])

• streamDuplex <stream.Duplex>
• options <Object>
  • readableType <string> Задаёт тип ReadableStream половины созданной пары для чтения и записи. Должно быть значение 'bytes' или undefined. (options.type — устаревший псевдоним этого параметра.)
• Возвращает: <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();
jscopy

Или с использованием 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;
    }
  }
})();
jscopy

Или с использованием 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();
});
jscopy

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 });
    // ...
  }
}
jscopy

При расширении потоков учитывайте, какие параметры пользователь может и должен задавать перед передачей их базовому конструктору. Например, если реализация предполагает определенные значения параметров 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...
  },
});
jscopy

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

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

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

new stream.Writable([options])

• options <Object>
  • highWaterMark <number> Размер буфера, при котором stream.write() начинает возвращать false. По умолчанию: 65536 (64 КиБ) или 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);
    // ...
  }
}
jscopy

Или при использовании конструкторов в стиле до 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);
jscopy

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

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

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

Вызов 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();
jscopy

writable._construct(callback)

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

Метод _construct() НЕ ДОЛЖЕН вызываться напрямую. Он может быть реализован дочерними классами и, если реализован, будет вызываться только внутренними методами класса Writable.

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

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);
    }
  }
}
jscopy

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();
    }
  },
});
jscopy

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

В следующем примере показана довольно простая (и в некотором смысле бессмысленная) реализация пользовательского потока 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();
    }
  }
}
jscopy

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

Декодирование буферов — распространенная задача, например при использовании преобразователей, входными данными для которых служит строка. При многобайтовой кодировке символов, такой как 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?.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: €
jscopy

Реализация потока для чтения

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

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

new stream.Readable([options])

• options <Object>
  • highWaterMark <number> Максимальное количество байтов, хранимых во внутреннем буфере до прекращения чтения из базового ресурса. По умолчанию: 65536 (64 КиБ) или 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);
    // ...
  }
}
jscopy

Или при использовании конструкторов в стиле до 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);
jscopy

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

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

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

Вызов 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();
jscopy

readable._construct(callback)

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

Метод _construct() НЕ ДОЛЖЕН вызываться напрямую. Он может быть реализован дочерними классами и, если реализован, будет вызываться только внутренними методами класса Readable.

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

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);
    }
  }
}
jscopy

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();
  }
}
jscopy

Метод 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.
    }
  },
});
jscopy

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

Ниже приведен простой пример потока 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);
    }
  }
}
jscopy

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

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

Поскольку JavaScript не поддерживает множественное наследование, класс stream.Duplex расширяется для реализации потока 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);
    // ...
  }
}
jscopy

Или при использовании конструкторов в стиле до 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);
jscopy

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

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

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

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

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');
    }
  },
);
jscopy

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

Ниже приведен простой пример потока 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));
    });
  }
}
jscopy

Важнейшая особенность потока 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
jscopy

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

Поток 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);
    // ...
  }
}
jscopy

Или при использовании конструкторов в стиле до 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);
jscopy

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

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

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

Событие: '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);
};
jscopy

Перед именем метода 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);
  }
})();
jscopy

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

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

Читаемый поток 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);
});
jscopy

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

При записи в записываемый поток из асинхронного итератора необходимо правильно обрабатывать обратное давление и ошибки. Метод 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();
  });
jscopy

Совместимость со старыми версиями 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);
jscopy

До 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);
jscopy

Помимо переключения новых потоков 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 это не создает проблем. Но при работе со строками, содержащими многобайтовые символы, рекомендуется учитывать такое поведение.