Событие: 'close'

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

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

Событие: 'drain'

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

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

Событие: 'error'

• Тип: <Error>

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

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

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

Событие: 'finish'

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

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

Событие: 'pipe'

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

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

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

Событие: 'unpipe'

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

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

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

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

writable.cork()

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

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

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

writable.destroy([error])

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

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

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

const myStream = new Writable();

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

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

const myStream = new Writable();

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

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

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

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

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

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

writable.closed

• Тип: <boolean>

Равно true после генерации события 'close'.

writable.destroyed

• Тип: <boolean>

Равно true после вызова writable.destroy().

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

const myStream = new Writable();

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

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

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

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

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

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

writable.setDefaultEncoding(encoding)

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

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

writable.uncork()

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

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

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

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

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

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

writable.writable

• Тип: <boolean>

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

writable.writableAborted

• Тип: <boolean>

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

writable.writableEnded

• Тип: <boolean>

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

writable.writableCorked

• Тип: <integer>

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

writable.errored

• Тип: <Error>

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

writable.writableFinished

• Тип: <boolean>

Устанавливается в true непосредственно перед генерацией события 'finish'.

writable.writableHighWaterMark

• Тип: <number>

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

writable.writableLength

• Тип: <number>

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

writable.writableNeedDrain

• Тип: <boolean>

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

writable.writableObjectMode

• Тип: <boolean>

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

writable[Symbol.asyncDispose]()

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

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

• chunk Необязательные данные для записи. Для потоков, работающих не в объектном режиме, chunk должен быть значением типа <string>, <Buffer>, <TypedArray> или <DataView>. Для потоков в объектном режиме chunk может быть любым значением JavaScript, кроме null. <string> | <Buffer> | <TypedArray> | <DataView> | <any>
• encoding Кодировка, если chunk — строка. По умолчанию: 'utf8' <string> | <null>
• 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.');
}); copy

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

Событие: 'close'

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

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

Событие: 'data'

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

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

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

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

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

Событие: 'end'

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

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

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

Событие: 'error'

• Тип: <Error>

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

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

Событие: 'pause'

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

Событие: 'readable'

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

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

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

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

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

При запуске этого скрипта вывод будет следующим:

$ node test.js
readable: null
end copy

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

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

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

Событие: 'resume'

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

readable.destroy([error])

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

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

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

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

readable.closed

• Тип: <boolean>

Равно true после испускания 'close'.

readable.destroyed

• Тип: <boolean>

Равно true после вызова readable.destroy().

readable.isPaused()

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

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

const readable = new stream.Readable();

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

readable.pause()

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

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

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

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

readable.pipe(destination[, options])

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

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

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

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

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

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

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

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

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

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

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

readable.read([size])

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

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

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

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

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

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

const readable = getReadableStreamSomehow();

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

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

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

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

const chunks = [];

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

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

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

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

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

readable.readable

• Тип: <boolean>

Равно true, если безопасно вызвать readable.read(); это означает, что поток не был уничтожен и не испустил события 'error' или 'end'.

readable.readableAborted

• Тип: <boolean>

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

readable.readableDidRead

• Тип: <boolean>

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

readable.readableEncoding

• Тип: <null> | <string>

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

readable.readableEnded

• Тип: <boolean>

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

readable.errored

• Тип: <Error>

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

readable.readableFlowing

• Тип: <boolean>

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

readable.readableHighWaterMark

• Тип: <number>

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

readable.readableLength

• Тип: <number>

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

readable.readableObjectMode

• Тип: <boolean>

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

readable.resume()

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

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

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

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

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

readable.setEncoding(encoding)

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

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

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

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

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

readable.unpipe([destination])

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

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

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

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

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

readable.unshift(chunk[, encoding])

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

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

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

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

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

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

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

readable.wrap(stream)

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

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

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

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

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

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

readable[Symbol.asyncIterator]()

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

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

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

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

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

readable[Symbol.asyncDispose]()

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

readable.compose(stream[, options])

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

import { Readable } from 'node:stream';

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

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

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

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

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

readable.iterator([options])

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

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

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

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

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

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

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

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

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

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

showBoth(); copy

readable.map(fn[, options])

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

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

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

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

readable.filter(fn[, options])

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

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

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

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

readable.forEach(fn[, options])

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

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

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

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

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

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

readable.toArray([options])

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

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

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

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

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

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

readable.some(fn[, options])

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

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

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

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

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

readable.find(fn[, options])

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

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

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

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

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

readable.every(fn[, options])

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

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

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

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

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

readable.flatMap(fn[, options])

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

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

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

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

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

readable.drop(limit[, options])

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

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

import { Readable } from 'node:stream';

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

readable.take(limit[, options])

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

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

import { Readable } from 'node:stream';

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

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

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

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

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

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

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

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

console.log(folderSize); copy

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

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

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

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

console.log(folderSize); copy

duplex.allowHalfOpen

• Тип: <boolean>

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

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

transform.destroy([error])

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

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

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