Событие: '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> исходный поток, который прекратил передачу данных в этот поток для записи методом unpiped

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

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

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

writable.cork()

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

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

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

writable.destroy([error])

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

Уничтожает поток. При необходимости генерирует событие 'error' и событие 'close' (если только для emitClose не задано значение false). После этого вызова поток для записи завершает работу, а последующие вызовы write() или end() приведут к ошибке ERR_STREAM_DESTROYED. Это разрушительный и немедленный способ уничтожения потока. Данные, переданные предыдущими вызовами write(), могли не быть обработаны, что может привести к ошибке ERR_STREAM_DESTROYED. Вместо 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 <string> | <Buffer> | <TypedArray> | <DataView> | <any> Необязательные данные для записи. Для потоков, не работающих в режиме объектов, chunk должно быть значением типа <string>, <Buffer>, <TypedArray> или <DataView>. Для потоков в режиме объектов chunk может быть любым значением JavaScript, кроме null.
• encoding <string> Кодировка, если chunk является строкой
• callback <Function> Функция обратного вызова, вызываемая после завершения работы с потоком.
• Возвращает: <this>

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

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

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

writable.setDefaultEncoding(encoding)

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

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

writable.uncork()

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

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

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

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

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

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

writable.writable

• Тип: <boolean>

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

writable.writableAborted

• Тип: <boolean>

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

writable.writableEnded

• Тип: <boolean>

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

writable.writableCorked

• Тип: <integer>

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

writable.errored

• Тип: <Error>

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

writable.writableFinished

• Тип: <boolean>

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

writable.writableHighWaterMark

• Тип: <number>

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

writable.writableLength

• Тип: <number>

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

writable.writableNeedDrain

• Тип: <boolean>

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

writable.writableObjectMode

• Тип: <boolean>

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

writable[Symbol.asyncDispose]()

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

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

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

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

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

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

Запись данных, пока поток не может освободить буфер, особенно проблематична для потока Transform, поскольку потоки Transform по умолчанию приостановлены до тех пор, пока для них не будет настроена передача данных или не будет добавлен обработчик события '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>

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

Функции обратного вызова-обработчика будет передан один объект 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 КиБ, поскольку параметр highWaterMark не передан в fs.createReadStream().

readable[Symbol.asyncDispose]()

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

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'] copy

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

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

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

Из 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'.