stream

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

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

Генератор событий. Список определённых событий включает:

1. close
2. data
3. drain
4. end
5. error
6. finish
7. pause
8. pipe
9. readable
10. resume
11. unpipe

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

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

Вспомогательный метод для создания Duplex на основе веб-ReadableStream и WritableStream.

Вспомогательный метод для создания веб-ReadableStream и WritableStream на основе Duplex.

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

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

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

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

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

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

Геттер свойства encoding заданного потока Readable. Значение свойства encoding можно задать с помощью метода setEncoding.

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

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

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

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

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

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

Генератор событий. Определённые события документов включают:

1. close
2. data
3. end
4. error
5. pause
6. readable
7. resume

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

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

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

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

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

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

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

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

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

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

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

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

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

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

const readable = new stream.Readable();

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

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

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

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

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

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

Каждый вызов readable.read() возвращает фрагмент данных или null. Фрагменты не объединяются. Чтобы получить все данные, находящиеся в буфере, необходим цикл 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('');
});

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

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

Вызов read после генерации события 'end' вернёт null. Ошибка времени выполнения возникать не будет.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

import fs from '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);

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

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

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

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

// Pull off a header delimited by \n\n.
// Use unshift() if we get too much.
// Call the callback with (error, header, stream).
import { StringDecoder } from '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;
    }
  }
}

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

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

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

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

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

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

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

Вспомогательный метод для создания Readable из веб-объекта ReadableStream.

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

Вспомогательный метод для создания веб-объекта ReadableStream из Readable.

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

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

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

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

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

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

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

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

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

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

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

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

1. close
2. drain
3. error
4. finish
5. pipe
6. unpipe

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

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

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

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

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

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

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

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

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

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

Метод writable.uncork() сбрасывает все данные, буферизованные после вызова 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());

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

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

Метод 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 и использовать 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.');
});

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

Вспомогательный метод для создания Writable из веб-потока WritableStream.

Вспомогательный метод для создания веб-потока WritableStream из Writable.

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

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

Эмиттер событий. Определенные события в документации включают:

1. close
2. data
3. drain
4. end
5. error
6. finish
7. pause
8. pipe
9. readable
10. resume
11. unpipe