Stream pipeline что это

Node.js Streams для чайников или как работать с потоками

Я думаю многие не раз слышали про Node js Streams, но так ни разу и не использовали, либо использовали, не задумываясь как же они работают, запайпили (pipe) стрим и норм. Давайте же разберемся что такое стримы, запайпить (pipe), чанки (chunk — часть данных) и все такое))

Почему важно понимать как устроены стримы в Node js? Ответ прост: многие встроенные модули в Node js реализуют стримы, такие как HTTP requests/responses, fs read/write, zlib, crypto, TCP sockets и другие. Также стримы вам понадобятся, к примеру, при обработке больших файлов, при работе с картинками. Возможно вы и не будете писать свой собственный стрим, однако понимание как это работает сделает вас более компетентным разработчиком.

Итак, что же такое стрим(далее по тексту буду использовать вместо Stream (поток)). Стрим — это концепция, c помощью которой можно обрабатывать данные небольшими частями, что позволяет задействовать небольшой объем оперативной памяти. Также с ее помощью мы можем разбить обработку каждой части на независимые друг от друга модули (функции либо классы). Например, мы можем сразу сжать часть данных, потом зашифровать и записать в файл. Основная идея в том, чтобы не работать с данными целиком, а поочередно обрабатывать часть данных.

В Node js есть 4 вида стримов:

• Readable — чтение
• Writable — запись
• Duplex — чтение и запись
• Transform — вид Duplex потока, который может изменять данные

Простой пример

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

// 1 - (используя стримы) загружаем часть файла и отправляем ее, до тех пор пока не отправим весь файл const getFile = async (req, res, next) => < const fileStream = fs.createReadStream('path to file'); res.contentType('application/pdf'); fileStream.pipe(res); >; // 2 - (не используя стримы) загружаем файл полностью в память и затем отправляем const getFile = async (req, res, next) => < const file = fs.readFileSync('path to file'); res.contentType('application/pdf'); res.send(file); >; 

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

Далее в статье разберем каждый стрим по отдельности. Стрим можно создать используя наследование или с помощью функции-конструктора.

const < Readable >= require('stream'); // 1 - Используя конструктор const myReadable = new Readable(opt); // 2 - Наследуя класс class myReadable extends Readable < constructor(opt) < super(opt); >> 

Во всех примерах я буду использовать 2 способ.

Readable stream

Давайте рассмотрим как же нам создать Readable стрим в NodeJS.

const < Readable >= require('stream'); class myReadable extends Readable < constructor(opt) < super(opt); >_read(size) <> > 

Как видим из примера выше, этот класс принимает набор параметров. Мы рассмотрим только те, которые нужны для общего понимания работы Readable стрима, остальные вы можете посмотреть в документации. Нас интересует параметр highWaterMark и метод _read.

highWaterMark — это максимальное количество байтов внутреннего буфера стрима (по умолчанию 16кб) по достижению которого считывание из ресурса приостанавливается. Для того, чтобы продолжить считывание, нам нужно освободить внутренний буфер. Мы можем это сделать вызвав методы pipe, resume или подписавшись на событие data.

_read — это реализация приватного метода, который вызывается внутренними методами класса Readable. Он вызывается постоянно пока размер данных не достигнет highWaterMark.

Ну и последний метод, который нас интересует, это readable.push, он непосредственно и добавляет данные во внутренний буфер. Он возвращает true, но как только буфер будет заполнен, то вызов этого метода начнет возвращать false. Он может управляться методом readable._read.

Давайте теперь посмотрим пример для прояснения ситуации.

class Counter extends Readable < constructor(opt) < super(opt); this._max = 1000; this._index = 0; >_read() < this._index += 1; if (this._index >this._max) < this.push(null); >else < const buf = Buffer.from(`$`, 'utf8'); console.log(`Added: $. Could be added? `, this.push(buf)); > > > const counter = new Counter(< highWaterMark: 2 >); console.log(`Received: $`); 

Для начала скажу, что counter.read() — это не тот _read, который мы реализовали в классе. Тот метод является приватным, а этот — открытым, и он возвращает данные из внутреннего буфера. Когда мы выполним этот код, в консоли мы увидим следующее:

Что же тут произошло? При создании стрима new Counter(< highWaterMark: 2 >) мы указали, что размер нашего внутреннего буфера будет равняться 2-м байтам, т.е. может хранить 2 символа (1 символ = 1 байт). После вызова counter.read() стрим начинает считывание, записывает ‘1’ во внутренний буфер и возвращает его. Затем он продолжает считывание, записывает ‘2’. Когда он запишет ‘3’, то буфер будет заполнен, readable.push вернет false, и стрим будет ждать, пока внутренний буфер освободится. Т.к. в нашем примере нет логики на освобождения буфера, скрипт завершится.

Как и говорилось ранее, для того, чтобы чтение не прерывалось, нам нужно постоянно очищать внутренний буфер. Для этого мы подпишемся на событие data. Заменим последние 2 строчки следующим кодом.

const counter = new Counter(< highWaterMark: 2 >); counter.on('data', chunk => < console.log(`Received: $`); >); 

Теперь если мы запустим этот пример, то увидим, что все сработало как надо и в консоли выведутся цифры от 1 до 1000.

Writable stream

На самом деле он очень похож на Readable стрим, только предназначен для записи данных.

const < Writable >= require('stream'); class myWritable extends Writable < constructor(opt) < super(opt); >_write(chunk, encoding, callback) <> > 

Он принимает похожие параметры, как и Readable стрим. Нас интересуют highWaterMark и _write.

_write — это приватный метод, который вызывается внутренними методами класса Writable для записи порции данных. Он принимает 3 параметра: chunk (часть данных), encoding (кодировка, если chunk это строка), callback (функция, которая вызывается после успешной или неудачной записи).

highWaterMark — это максимальное количество байтов внутреннего буфера стрима (по умолчанию 16кб), по достижению которого stream.write начнет возвращать false.

Давайте перепишем предыдущий пример со счетчиком.

const < Writable >= require('stream'); class Counter extends Writable < _write(chunk, encoding, callback) < console.log(chunk.toString()); callback(); >> const counter = new Counter(< highWaterMark: 2 >); for (let i = 1; i < 1000; i += 1) < counter.write(Buffer.from(`$`, 'utf8')); > 

По сути все просто, но есть один интересный нюанс, о котором стоит помнить! При создании стрима new Counter(< highWaterMark: 2 >) мы указали, что размер нашего внутреннего буфера будет равняться 2-м байтам, т.е. может хранить 2 символа (1 символ = 1 байт). Когда же счетчик дойдет до десяти, то буфер будет заполняться при каждом вызове write, соответственно, если бы запись осуществлялась в медленный источник, то все остальные данные при вызове write сохранялись бы в оперативную память, что могло бы вызвать ее переполнение (в данном примере это конечно же не важно, так как наш буфер 2 байта, но вот с большими файлами об этом нужно помнить). Когда возникает такая ситуация, нам надо подождать, пока стрим запишет текущую порцию данных, освободит внутренний буфер (вызовет событие drain), и затем мы можем возобновить запись данных. Давайте перепишем наш пример.

const < Writable >= require('stream'); const < once >= require('events'); class Counter extends Writable < _write(chunk, encoding, callback) < console.log(chunk.toString()); callback(); >> const counter = new Counter(< highWaterMark: 2 >); (async () => < for (let i = 1; i < 1000; i += 1) < const canWrite = counter.write(Buffer.from(`$`, 'utf8')); console.log(`Can we write bunch of data? $`); if (!canWrite) < await events.once(counter, 'drain'); console.log('drain event fired.'); >> >)(); 

Метод events.once был добавлен в v11.13.0 и позволяет создать промис и подождать, пока определенное событие выполнится один раз. В этом примере мы проверяем, возможна ли запись данных в стрим, если нет, то ожидаем, пока буфер освободится, и продолжаем запись.

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

Duplex stream

Он объединяет в себе Readable и Writable стримы, то есть мы должны написать реализацию двух методов _read и _write.

const < Duplex >= require('stream'); class myDuplex extends Duplex < constructor(opt) < super(opt); >_read(size) <> _write(chunk, encoding, callback) <> > 

Здесь нам интересны 2 параметра, которые мы можем передать в конструктор, это readableHighWaterMark и writableHighWaterMark, которые позволяют нам указать размер внутреннего буфера для Readable, Writable стримов соответственно. Вот так будет выглядеть реализация предыдущих двух примеров с помощью Duplex стрима.

const < Duplex >= require('stream'); const events = require('events'); class Counter extends Duplex < constructor(opt) < super(opt); this._max = 1000; this._index = 0; >_read() < this._index += 1; if (this._index >this._max) < this.push(null); >else < const buf = Buffer.from(`$`, 'utf8'); this.push(buf); > > _write(chunk, encoding, callback) < console.log(chunk.toString()); callback(); >> const counter = new Counter(< readableHighWaterMark: 2, writableHighWaterMark: 2 >); (async () => < let chunk = counter.read(); while (chunk !== null) < const canWrite = counter.write(chunk); console.log(`Can we write bunch of data? $`); if (!canWrite) < await events.once(counter, 'drain'); console.log('drain event fired.'); >chunk = counter.read(); > >)(); 

Думаю, этот код не нуждается в пояснениях, так как он такой же, как и раньше, только в одном классе.

Transform stream

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

const < Transform >= require('stream'); class myTransform extends Transform < _ transform(chunk, encoding, callback) <>> 

Нас интересует метод _transform.

_transform — это приватный метод, который вызывается внутренними методами класса Transform для преобразования порции данных. Он принимает 3 параметра: chunk (часть данных), encoding (кодировка, если chunk это строка), callback (функция, которая вызывается после успешной или неудачной записи).

С помощью этого метода и будет происходить изменение порции данных. Внутри этого метода мы можем вызвать transform.push() ноль или несколько раз, который фиксирует изменения. Когда мы завершим преобразование данных, мы должны вызвать callback, который отправит все, что мы добавляли в transform.push(). Первый параметр этой callback функции — это ошибка. Также мы можем не использовать transform.push(), а отправить измененные данные вторым параметром в функцию callback (пример: callback(null, data)). Для того, чтобы понять как использовать этот вид стрима, давайте разберем метод stream.pipe.

stream.pipe — этот метод используется для соединения Readable стрима с Writable стримом, а также для создания цепочек стримов. Это значит, что мы можем считывать часть данных и передавать в следующий стрим для обработки, а потом в следующий и т д.

Давайте напишем Transform стрим, который будет добавлять символ * в начало и конец каждой части данных.

class CounterReader extends Readable < constructor(opt) < super(opt); this._max = 1000; this._index = 0; >_read() < this._index += 1; if (this._index >this._max) < this.push(null); >else < const buf = Buffer.from(`$`, 'utf8'); this.push(buf); > > > class CounterWriter extends Writable < _write(chunk, encoding, callback) < console.log(chunk.toString()); callback(); >> class CounterTransform extends Transform < _transform(chunk, encoding, callback) < try < const resultString = `*$*`; callback(null, resultString); > catch (err) < callback(err); >> > const counterReader = new CounterReader(< highWaterMark: 2 >); const counterWriter = new CounterWriter(< highWaterMark: 2 >); const counterTransform = new CounterTransform(< highWaterMark: 2 >); counterReader.pipe(counterTransform).pipe(counterWriter); 

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

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

Streams. Потоки в Node js

Streams — это концепция для передачи данных с одной программы в другую в операциях ввода/вывода. Streams позволяют передавать данные небольшими порциями, с помощью чего мы можем работать с довольно большими объемами файлов. При работе с потоками мы не расходуем много памяти т.к. нам не нужно загружать весь файл в память, а только часть файла для буферизации.

Виды потоков

• Readable — поток для считывания
• Writable — поток для записи
• Duplex — это поток в котором можно и читать данные и записывать, где эти 2 процесса происходят независимо друг от друга
• Transform — разновидность Duplex потока, которые могут изменять данные при их записи/чтению

Поток имеет внутренний буфер (Back Buffer) для временного хранения данных, которые были получены, пока они не будут обработаны соответствующим способом (методом read() или write() в соответствии с видом потока).

Размер буфера можно указать через параметр highWaterMark, где смысл этого параметра устанавливается опцией objectMode.

new StreamObject(); //по умолчанию 16384 (16kb) new StreamObject();//по умолчанию 16

Readable stream

Events

• readable — когда стрим готов читать со своего внутреннего буфера
• error — при ошибке
• end — когда считывать больше нечего

Methods

• read() — считать часть даты с внутреннего буфера
• read(N) — считать кусок размером в N байт

В Readable stream данные буферизуются, пока не будет вызван метод read(). Как только общий размер внутреннего буфера достигнет порогового значения, указанного в highWaterMark, поток временно прекратит чтение данных.

const fs = require('fs'); const streamRead = new fs.ReadStream(_filename); // _filename - ссылка на файл для чтения streamRead.on('readable', function() < let data = streamRead.read(); if (data !== null) < console.log(data); // фрагмент данных в виде буфера if (data !== null) < console.log(data.length); // длина прочитанного фрагмента файла console.log(data.toString()); // преобразование данных с буфера в строку >> >); streamRead.on('end', () => < console.log('The end'); >)

В этом примере мы поставили библиотеку fs для создания стрима и читаем данные из текстового файла. fs.ReadStream реализует стандартный поток чтения, который наследует от stream.Readable. Мы создали поток и он пытается прочитать данные с файла, когда он что-то прочитал — имитирует событие ‘readable’. Это событие оглашает, что данные прочитаны и находятся во внутреннем буфере. После, при помощи метода read() мы получаем данные из внутреннего буфера и уже можем его обработать.

Если data === null — значит данные на считывание закончились.

Writable stream

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

Events

• drain — когда внутренний буфер свободен для получения новых данных
• error — при ошибке
• finish — когда был вызван метод end() и все данные с внутреннего буфера были записаны

Methods

• write() — запись данных
• end() — закончить процесс

Создадим простой стрим для записи данных — 2 в степени 0, 2 в степени 1 и так до 10-й степени. Когда степень будет равна 10 — записываем полученное значение и закрываем стрим.

const fs = require('fs'); const streamWrite = new fs.WriteStream(_filename); // _filename - имя файла for (let i = 0; i | `; streamWrite.write(data); // записываем данные в файл if (i == 10) < streamWrite.end(); >> streamWrite.on('finish', () => < console.log('The end'); >)

Здесь мы создаем стрим для записи через fs.WriteStream, производим вычисление данных data при помощи цикла и записываем данные методом write. Когда нам больше нечего записывать, вызываем метод end. Если данные с внутреннего буфера записались все — срабатывает событие finish и выводиться сообщение в консоль.

Ничего сложного. Пойдем далее. Мы уже умеем писать стримы для записи и считывания так давайте сделаем микс — читаем данные, ищем нужные значения, если подходят — записываем.

const fs = require('fs'); const _filenameWrite = './demo/wr.txt'; const _filenameRead = './demo/demo.txt'; // создаем 2 стрима - на запись и считывание const streamRead = new fs.ReadStream(_filenameRead); const streamWrite = new fs.WriteStream(_filenameWrite); streamRead.on('readable', function() < if (data !== null) < // ищем есть ли в полученном блоке нужные строки, если находим - записываем в новый файл if (data.includes('Понтий Пилат')) < streamWrite.write(data); // записываем данные в файл >> >); streamRead.on('end', () => < console.log('The end read'); streamWrite.end(); >) streamWrite.on('finish', () => < console.log('The end write'); >)

Pipe

Pipe — это канал, который связывает поток для чтения и поток для записи. Он позволяет сразу передавать данные с чтения на запись и контролирует, чтоб к моменту передачи новых данных для записи, предыдущие были уже записаны.

const fs = require('fs'); const _filenameWrite = './demo/wr.txt'; const _filenameRead = './demo/demo.txt'; // создаем 2 стрима - на запись и считывание const streamRead = new fs.createReadStream(_filenameRead); const streamWrite = new fs.createWriteStream(_filenameWrite); streamRead.pipe(streamWrite);

У потока чтения вызывается метод pipe(), в который передается поток для записи.

Node.js v21.1.0 documentation

A stream is an abstract interface for working with streaming data in Node.js. The node:stream module provides an API for implementing the stream interface.

There are many stream objects provided by Node.js. For instance, a request to an HTTP server and process.stdout are both stream instances.

Streams can be readable, writable, or both. All streams are instances of EventEmitter .

To access the node:stream module:

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

The node:stream module is useful for creating new types of stream instances. It is usually not necessary to use the node:stream module to consume streams.

Organization of this document #

This document contains two primary sections and a third section for notes. The first section explains how to use existing streams within an application. The second section explains how to create new types of streams.

Types of streams #

There are four fundamental stream types within Node.js:

• Writable : streams to which data can be written (for example, fs.createWriteStream() ).
• Readable : streams from which data can be read (for example, fs.createReadStream() ).
• Duplex : streams that are both Readable and Writable (for example, net.Socket ).
• Transform : Duplex streams that can modify or transform the data as it is written and read (for example, zlib.createDeflate() ).

Streams Promises API #

Added in: v15.0.0

The stream/promises API provides an alternative set of asynchronous utility functions for streams that return Promise objects rather than using callbacks. The API is accessible via require(‘node:stream/promises’) or require(‘node:stream’).promises .

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

stream.pipeline(streams[, options]) #

Added in: v15.0.0

• streams | | |
• source | | |
  • Returns: |
  • source
  • Returns: |
  • source
  • Returns: |
  • signal
  • end

  const < pipeline >= require('node:stream/promises'); const fs = require('node:fs'); const zlib = require('node:zlib'); async function run( ) < await pipeline( fs.createReadStream('archive.tar'), zlib.createGzip(), fs.createWriteStream('archive.tar.gz'), ); console.log('Pipeline succeeded.'); > run().catch(console.error);import < pipeline >from 'node:stream/promises'; import < createReadStream, createWriteStream >from 'node:fs'; import < createGzip >from 'node:zlib'; await pipeline( createReadStream('archive.tar'), createGzip(), createWriteStream('archive.tar.gz'), ); console.log('Pipeline succeeded.');copy

  To use an AbortSignal , pass it inside an options object, as the last argument. When the signal is aborted, destroy will be called on the underlying pipeline, with an AbortError .

  const < pipeline >= require('node:stream/promises'); const fs = require('node:fs'); const zlib = require('node:zlib'); async function run( ) < const ac = new AbortController(); const signal = ac.signal; setImmediate(() => ac.abort()); await pipeline( fs.createReadStream('archive.tar'), zlib.createGzip(), fs.createWriteStream('archive.tar.gz'), < signal >, ); > run().catch(console.error); // AbortErrorimport < pipeline >from 'node:stream/promises'; import < createReadStream, createWriteStream >from 'node:fs'; import < createGzip >from 'node:zlib'; const ac = new AbortController(); const < signal >= ac; setImmediate(() => ac.abort()); try < await pipeline( createReadStream('archive.tar'), createGzip(), createWriteStream('archive.tar.gz'), < signal >, ); > catch (err) < console.error(err); // AbortError >copy

  The pipeline API also supports async generators:

  const < pipeline >= require('node:stream/promises'); const fs = require('node:fs'); async function run( ) < await pipeline( fs.createReadStream('lowercase.txt'), async function* (source, < signal >) < source.setEncoding('utf8'); // Work with strings rather than `Buffer`s. for await (const chunk of source) < yield await processChunk(chunk, < signal >); > >, fs.createWriteStream('uppercase.txt'), ); console.log('Pipeline succeeded.'); > run().catch(console.error);import < pipeline >from 'node:stream/promises'; import < createReadStream, createWriteStream >from 'node:fs'; await pipeline( createReadStream('lowercase.txt'), async function* (source, < signal >) < source.setEncoding('utf8'); // Work with strings rather than `Buffer`s. for await (const chunk of source) < yield await processChunk(chunk, < signal >); > >, createWriteStream('uppercase.txt'), ); console.log('Pipeline succeeded.');copy

  Remember to handle the signal argument passed into the async generator. Especially in the case where the async generator is the source for the pipeline (i.e. first argument) or the pipeline will never complete.

  const < pipeline >= require('node:stream/promises'); const fs = require('node:fs'); async function run( ) < await pipeline( async function* (< signal >) < await someLongRunningfn(< signal >); yield 'asd'; >, fs.createWriteStream('uppercase.txt'), ); console.log('Pipeline succeeded.'); > run().catch(console.error);import < pipeline >from 'node:stream/promises'; import fs from 'node:fs'; await pipeline( async function* (< signal >) < await someLongRunningfn(< signal >); yield 'asd'; >, fs.createWriteStream('uppercase.txt'), ); console.log('Pipeline succeeded.');copy

  The pipeline API provides callback version:

  stream.finished(stream[, options]) #

  Added in: v15.0.0

  • stream
  • options
    • error |
    • readable |
    • writable |
    • signal : |

    const < finished >= require('node:stream/promises'); const fs = require('node:fs'); const rs = fs.createReadStream('archive.tar'); async function run( ) < await finished(rs); console.log('Stream is done reading.'); > run().catch(console.error); rs.resume(); // Drain the stream.import < finished >from 'node:stream/promises'; import < createReadStream >from 'node:fs'; const rs = createReadStream('archive.tar'); async function run( ) < await finished(rs); console.log('Stream is done reading.'); > run().catch(console.error); rs.resume(); // Drain the stream.copy

    The finished API also provides a callback version.

    Object mode #

    All streams created by Node.js APIs operate exclusively on strings and Buffer (or Uint8Array ) objects. It is possible, however, for stream implementations to work with other types of JavaScript values (with the exception of null , which serves a special purpose within streams). Such streams are considered to operate in «object mode».

    Stream instances are switched into object mode using the objectMode option when the stream is created. Attempting to switch an existing stream into object mode is not safe.

    Buffering #

    Both Writable and Readable streams will store data in an internal buffer.

    The amount of data potentially buffered depends on the highWaterMark option passed into the stream’s constructor. For normal streams, the highWaterMark option specifies a total number of bytes. For streams operating in object mode, the highWaterMark specifies a total number of objects.

    Data is buffered in Readable streams when the implementation calls stream.push(chunk) . If the consumer of the Stream does not call stream.read() , the data will sit in the internal queue until it is consumed.

    Once the total size of the internal read buffer reaches the threshold specified by highWaterMark , the stream will temporarily stop reading data from the underlying resource until the data currently buffered can be consumed (that is, the stream will stop calling the internal readable._read() method that is used to fill the read buffer).

    Data is buffered in Writable streams when the writable.write(chunk) method is called repeatedly. While the total size of the internal write buffer is below the threshold set by highWaterMark , calls to writable.write() will return true . Once the size of the internal buffer reaches or exceeds the highWaterMark , false will be returned.

    A key goal of the stream API, particularly the stream.pipe() method, is to limit the buffering of data to acceptable levels such that sources and destinations of differing speeds will not overwhelm the available memory.

    The highWaterMark option is a threshold, not a limit: it dictates the amount of data that a stream buffers before it stops asking for more data. It does not enforce a strict memory limitation in general. Specific stream implementations may choose to enforce stricter limits but doing so is optional.

    Because Duplex and Transform streams are both Readable and Writable , each maintains two separate internal buffers used for reading and writing, allowing each side to operate independently of the other while maintaining an appropriate and efficient flow of data. For example, net.Socket instances are Duplex streams whose Readable side allows consumption of data received from the socket and whose Writable side allows writing data to the socket. Because data may be written to the socket at a faster or slower rate than data is received, each side should operate (and buffer) independently of the other.

    The mechanics of the internal buffering are an internal implementation detail and may be changed at any time. However, for certain advanced implementations, the internal buffers can be retrieved using writable.writableBuffer or readable.readableBuffer . Use of these undocumented properties is discouraged.

    API for stream consumers #

    Almost all Node.js applications, no matter how simple, use streams in some manner. The following is an example of using streams in a Node.js application that implements an HTTP server:

    const http = require('node:http'); const server = http.createServer((req, res) => < // `req` is an http.IncomingMessage, which is a readable stream. // `res` is an http.ServerResponse, which is a writable stream. let body = ''; // Get the data as utf8 strings. // If an encoding is not set, Buffer objects will be received. req.setEncoding('utf8'); // Readable streams emit 'data' events once a listener is added. req.on('data', (chunk) => < body += chunk; >); // The 'end' event indicates that the entire body has been received. req.on('end', () => < try < const data = JSON.parse(body); // Write back something interesting to the user: res.write(typeof data); res.end(); > catch (er) < // uh oh! bad json! res.statusCode = 400; return res.end(`error: $ `); > >); >); server.listen(1337); // $ curl localhost:1337 -d "<>" // object // $ curl localhost:1337 -d "\"foo\"" // string // $ curl localhost:1337 -d "not json" // error: Unexpected token 'o', "not json" is not valid JSON copy

    Writable streams (such as res in the example) expose methods such as write() and end() that are used to write data onto the stream.

    Readable streams use the EventEmitter API for notifying application code when data is available to be read off the stream. That available data can be read from the stream in multiple ways.

    Both Writable and Readable streams use the EventEmitter API in various ways to communicate the current state of the stream.

    Applications that are either writing data to or consuming data from a stream are not required to implement the stream interfaces directly and will generally have no reason to call require(‘node:stream’) .

    Developers wishing to implement new types of streams should refer to the section API for stream implementers.

    Writable streams #

    Writable streams are an abstraction for a destination to which data is written.

    Examples of Writable streams include:

    • HTTP requests, on the client
    • HTTP responses, on the server
    • fs write streams
    • zlib streams
    • crypto streams
    • TCP sockets
    • child process stdin
    • process.stdout , process.stderr

    Some of these examples are actually Duplex streams that implement the Writable interface.

    All Writable streams implement the interface defined by the stream.Writable class.

    While specific instances of Writable streams may differ in various ways, all Writable streams follow the same fundamental usage pattern as illustrated in the example below:

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

    Class: stream.Writable #

    Added in: v0.9.4

    Event: ‘close’ #

    Add emitClose option to specify if ‘close’ is emitted on destroy.

    The ‘close’ event is emitted when the stream and any of its underlying resources (a file descriptor, for example) have been closed. The event indicates that no more events will be emitted, and no further computation will occur.

    A Writable stream will always emit the ‘close’ event if it is created with the emitClose option.

    Event: ‘drain’ #

    Added in: v0.9.4

    If a call to stream.write(chunk) returns false , the ‘drain’ event will be emitted when it is appropriate to resume writing data to the stream.

    // 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

    Event: ‘error’ #

    Added in: v0.9.4

    The ‘error’ event is emitted if an error occurred while writing or piping data. The listener callback is passed a single Error argument when called.

    The stream is closed when the ‘error’ event is emitted unless the autoDestroy option was set to false when creating the stream.

    After ‘error’ , no further events other than ‘close’ should be emitted (including ‘error’ events).

    Event: ‘finish’ #

    Added in: v0.9.4

    The ‘finish’ event is emitted after the stream.end() method has been called, and all data has been flushed to the underlying system.

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

    Event: ‘pipe’ #

    Added in: v0.9.4

    The ‘pipe’ event is emitted when the stream.pipe() method is called on a readable stream, adding this writable to its set of destinations.

    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

    Event: ‘unpipe’ #

    Added in: v0.9.4

    • src The source stream that unpiped this writable

    The ‘unpipe’ event is emitted when the stream.unpipe() method is called on a Readable stream, removing this Writable from its set of destinations.

    This is also emitted in case this Writable stream emits an error when a Readable stream pipes into it.

    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() #

    Added in: v0.11.2

    The writable.cork() method forces all written data to be buffered in memory. The buffered data will be flushed when either the stream.uncork() or stream.end() methods are called.

    The primary intent of writable.cork() is to accommodate a situation in which several small chunks are written to the stream in rapid succession. Instead of immediately forwarding them to the underlying destination, writable.cork() buffers all the chunks until writable.uncork() is called, which will pass them all to writable._writev() , if present. This prevents a head-of-line blocking situation where data is being buffered while waiting for the first small chunk to be processed. However, use of writable.cork() without implementing writable._writev() may have an adverse effect on throughput.

    writable.destroy([error]) #

    Work as a no-op on a stream that has already been destroyed.

    Destroy the stream. Optionally emit an ‘error’ event, and emit a ‘close’ event (unless emitClose is set to false ). After this call, the writable stream has ended and subsequent calls to write() or end() will result in an ERR_STREAM_DESTROYED error. This is a destructive and immediate way to destroy a stream. Previous calls to write() may not have drained, and may trigger an ERR_STREAM_DESTROYED error. Use end() instead of destroy if data should flush before close, or wait for the ‘drain’ event before destroying the stream.

    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

    Once destroy() has been called any further calls will be a no-op and no further errors except from _destroy() may be emitted as ‘error’ .

    Implementors should not override this method, but instead implement writable._destroy() .

    writable.closed #

    Added in: v18.0.0

    Is true after ‘close’ has been emitted.

    writable.destroyed #

    Added in: v8.0.0

    Is true after writable.destroy() has been called.

    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]) #

    The callback is invoked before ‘finish’ or on error.

    The callback is invoked if ‘finish’ or ‘error’ is emitted.

    This method now returns a reference to writable .

    The chunk argument can now be a Uint8Array instance.

    Calling the writable.end() method signals that no more data will be written to the Writable . The optional chunk and encoding arguments allow one final additional chunk of data to be written immediately before closing the stream.

    Calling the stream.write() method after calling stream.end() will raise an error.

    // 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) #

    This method now returns a reference to writable .

    Added in: v0.11.15

    • encoding The new default encoding
    • Returns:

    The writable.setDefaultEncoding() method sets the default encoding for a Writable stream.

    writable.uncork() #

    Added in: v0.11.2

    The writable.uncork() method flushes all data buffered since stream.cork() was called.

    When using writable.cork() and writable.uncork() to manage the buffering of writes to a stream, defer calls to writable.uncork() using process.nextTick() . Doing so allows batching of all writable.write() calls that occur within a given Node.js event loop phase.

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

    If the writable.cork() method is called multiple times on a stream, the same number of calls to writable.uncork() must be called to flush the buffered data.

    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.writable #

    Added in: v11.4.0

    Is true if it is safe to call writable.write() , which means the stream has not been destroyed, errored, or ended.

    writable.writableAborted #

    Added in: v18.0.0, v16.17.0
    Stability: 1 — Experimental

    Returns whether the stream was destroyed or errored before emitting ‘finish’ .

    writable.writableEnded #

    Added in: v12.9.0

    Is true after writable.end() has been called. This property does not indicate whether the data has been flushed, for this use writable.writableFinished instead.

    writable.writableCorked #

    Added in: v13.2.0, v12.16.0

    Number of times writable.uncork() needs to be called in order to fully uncork the stream.

    writable.errored #

    Added in: v18.0.0

    Returns error if the stream has been destroyed with an error.

    writable.writableFinished #

    Added in: v12.6.0

    Is set to true immediately before the ‘finish’ event is emitted.

    writable.writableHighWaterMark #

    Added in: v9.3.0

    Return the value of highWaterMark passed when creating this Writable .

    writable.writableLength #

    Added in: v9.4.0

    This property contains the number of bytes (or objects) in the queue ready to be written. The value provides introspection data regarding the status of the highWaterMark .

    writable.writableNeedDrain #

    Added in: v15.2.0, v14.17.0

    Is true if the stream’s buffer has been full and stream will emit ‘drain’ .

    writable.writableObjectMode #

    Added in: v12.3.0

    Getter for the property objectMode of a given Writable stream.

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

    The chunk argument can now be a Uint8Array instance.

    Passing null as the chunk parameter will always be considered invalid now, even in object mode.

    The writable.write() method writes some data to the stream, and calls the supplied callback once the data has been fully handled. If an error occurs, the callback will be called with the error as its first argument. The callback is called asynchronously and before ‘error’ is emitted.

    The return value is true if the internal buffer is less than the highWaterMark configured when the stream was created after admitting chunk . If false is returned, further attempts to write data to the stream should stop until the ‘drain’ event is emitted.

    While a stream is not draining, calls to write() will buffer chunk , and return false. Once all currently buffered chunks are drained (accepted for delivery by the operating system), the ‘drain’ event will be emitted. Once write() returns false, do not write more chunks until the ‘drain’ event is emitted. While calling write() on a stream that is not draining is allowed, Node.js will buffer all written chunks until maximum memory usage occurs, at which point it will abort unconditionally. Even before it aborts, high memory usage will cause poor garbage collector performance and high RSS (which is not typically released back to the system, even after the memory is no longer required). Since TCP sockets may never drain if the remote peer does not read the data, writing a socket that is not draining may lead to a remotely exploitable vulnerability.

    Writing data while the stream is not draining is particularly problematic for a Transform , because the Transform streams are paused by default until they are piped or a ‘data’ or ‘readable’ event handler is added.

    If the data to be written can be generated or fetched on demand, it is recommended to encapsulate the logic into a Readable and use stream.pipe() . However, if calling write() is preferred, it is possible to respect backpressure and avoid memory issues using the ‘drain’ event:

    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

    A Writable stream in object mode will always ignore the encoding argument.

    Readable streams #

    Readable streams are an abstraction for a source from which data is consumed.

    Examples of Readable streams include:

    • HTTP responses, on the client
    • HTTP requests, on the server
    • fs read streams
    • zlib streams
    • crypto streams
    • TCP sockets
    • child process stdout and stderr
    • process.stdin

    All Readable streams implement the interface defined by the stream.Readable class.

    Two reading modes #

    Readable streams effectively operate in one of two modes: flowing and paused. These modes are separate from object mode. A Readable stream can be in object mode or not, regardless of whether it is in flowing mode or paused mode.

    • In flowing mode, data is read from the underlying system automatically and provided to an application as quickly as possible using events via the EventEmitter interface.
    • In paused mode, the stream.read() method must be called explicitly to read chunks of data from the stream.

    All Readable streams begin in paused mode but can be switched to flowing mode in one of the following ways:

    • Adding a ‘data’ event handler.
    • Calling the stream.resume() method.
    • Calling the stream.pipe() method to send the data to a Writable .

    The Readable can switch back to paused mode using one of the following:

    • If there are no pipe destinations, by calling the stream.pause() method.
    • If there are pipe destinations, by removing all pipe destinations. Multiple pipe destinations may be removed by calling the stream.unpipe() method.

    The important concept to remember is that a Readable will not generate data until a mechanism for either consuming or ignoring that data is provided. If the consuming mechanism is disabled or taken away, the Readable will attempt to stop generating the data.

    For backward compatibility reasons, removing ‘data’ event handlers will not automatically pause the stream. Also, if there are piped destinations, then calling stream.pause() will not guarantee that the stream will remain paused once those destinations drain and ask for more data.

    If a Readable is switched into flowing mode and there are no consumers available to handle the data, that data will be lost. This can occur, for instance, when the readable.resume() method is called without a listener attached to the ‘data’ event, or when a ‘data’ event handler is removed from the stream.

    Adding a ‘readable’ event handler automatically makes the stream stop flowing, and the data has to be consumed via readable.read() . If the ‘readable’ event handler is removed, then the stream will start flowing again if there is a ‘data’ event handler.

    Three states #

    The «two modes» of operation for a Readable stream are a simplified abstraction for the more complicated internal state management that is happening within the Readable stream implementation.

    Specifically, at any given point in time, every Readable is in one of three possible states:

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

    When readable.readableFlowing is null , no mechanism for consuming the stream’s data is provided. Therefore, the stream will not generate data. While in this state, attaching a listener for the ‘data’ event, calling the readable.pipe() method, or calling the readable.resume() method will switch readable.readableFlowing to true , causing the Readable to begin actively emitting events as data is generated.

    Calling readable.pause() , readable.unpipe() , or receiving backpressure will cause the readable.readableFlowing to be set as false , temporarily halting the flowing of events but not halting the generation of data. While in this state, attaching a listener for the ‘data’ event will not switch readable.readableFlowing to true .

    const < PassThrough, Writable > = require('node:stream'); const pass = new PassThrough(); const writable = new Writable(); pass.pipe(writable); pass.unpipe(writable); // readableFlowing is now false. pass.on('data', (chunk) => < console.log(chunk.toString()); >); // readableFlowing is still false. pass.write('ok'); // Will not emit 'data'. pass.resume(); // Must be called to make stream emit 'data'. // readableFlowing is now true. copy

    While readable.readableFlowing is false , data may be accumulating within the stream’s internal buffer.

    Choose one API style #

    The Readable stream API evolved across multiple Node.js versions and provides multiple methods of consuming stream data. In general, developers should choose one of the methods of consuming data and should never use multiple methods to consume data from a single stream. Specifically, using a combination of on(‘data’) , on(‘readable’) , pipe() , or async iterators could lead to unintuitive behavior.

    Class: stream.Readable #

    Added in: v0.9.4

    Event: ‘close’ #

    Add emitClose option to specify if ‘close’ is emitted on destroy.

    The ‘close’ event is emitted when the stream and any of its underlying resources (a file descriptor, for example) have been closed. The event indicates that no more events will be emitted, and no further computation will occur.

    A Readable stream will always emit the ‘close’ event if it is created with the emitClose option.

    Event: ‘data’ #

    Added in: v0.9.4

    The ‘data’ event is emitted whenever the stream is relinquishing ownership of a chunk of data to a consumer. This may occur whenever the stream is switched in flowing mode by calling readable.pipe() , readable.resume() , or by attaching a listener callback to the ‘data’ event. The ‘data’ event will also be emitted whenever the readable.read() method is called and a chunk of data is available to be returned.

    Attaching a ‘data’ event listener to a stream that has not been explicitly paused will switch the stream into flowing mode. Data will then be passed as soon as it is available.

    The listener callback will be passed the chunk of data as a string if a default encoding has been specified for the stream using the readable.setEncoding() method; otherwise the data will be passed as a Buffer .

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

    Event: ‘end’ #

    Added in: v0.9.4

    The ‘end’ event is emitted when there is no more data to be consumed from the stream.

    The ‘end’ event will not be emitted unless the data is completely consumed. This can be accomplished by switching the stream into flowing mode, or by calling stream.read() repeatedly until all data has been consumed.

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

    Event: ‘error’ #

    Added in: v0.9.4

    The ‘error’ event may be emitted by a Readable implementation at any time. Typically, this may occur if the underlying stream is unable to generate data due to an underlying internal failure, or when a stream implementation attempts to push an invalid chunk of data.

    The listener callback will be passed a single Error object.

    Event: ‘pause’ #

    Added in: v0.9.4

    The ‘pause’ event is emitted when stream.pause() is called and readableFlowing is not false .

    Event: ‘readable’ #

    The ‘readable’ is always emitted in the next tick after .push() is called.

    Using ‘readable’ requires calling .read() .

    The ‘readable’ event is emitted when there is data available to be read from the stream or when the end of the stream has been reached. Effectively, the ‘readable’ event indicates that the stream has new information. If data is available, stream.read() will return that data.

    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

    If the end of the stream has been reached, calling stream.read() will return null and trigger the ‘end’ event. This is also true if there never was any data to be read. For instance, in the following example, foo.txt is an empty file:

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

    The output of running this script is:

    $ node test.js readable: null end copy

    In some cases, attaching a listener for the ‘readable’ event will cause some amount of data to be read into an internal buffer.

    In general, the readable.pipe() and ‘data’ event mechanisms are easier to understand than the ‘readable’ event. However, handling ‘readable’ might result in increased throughput.

    If both ‘readable’ and ‘data’ are used at the same time, ‘readable’ takes precedence in controlling the flow, i.e. ‘data’ will be emitted only when stream.read() is called. The readableFlowing property would become false . If there are ‘data’ listeners when ‘readable’ is removed, the stream will start flowing, i.e. ‘data’ events will be emitted without calling .resume() .

    Event: ‘resume’ #

    Added in: v0.9.4

    The ‘resume’ event is emitted when stream.resume() is called and readableFlowing is not true .

    readable.destroy([error]) #

    Work as a no-op on a stream that has already been destroyed.

    Destroy the stream. Optionally emit an ‘error’ event, and emit a ‘close’ event (unless emitClose is set to false ). After this call, the readable stream will release any internal resources and subsequent calls to push() will be ignored.

    Once destroy() has been called any further calls will be a no-op and no further errors except from _destroy() may be emitted as ‘error’ .

    Implementors should not override this method, but instead implement readable._destroy() .

    readable.closed #

    Added in: v18.0.0

    Is true after ‘close’ has been emitted.

    readable.destroyed #

    Added in: v8.0.0

    Is true after readable.destroy() has been called.

    readable.isPaused() #

    Added in: v0.11.14

    • Returns:

    The readable.isPaused() method returns the current operating state of the Readable . This is used primarily by the mechanism that underlies the readable.pipe() method. In most typical cases, there will be no reason to use this method directly.

    const readable = new stream.Readable(); readable.isPaused(); // === false readable.pause(); readable.isPaused(); // === true readable.resume(); readable.isPaused(); // === false copy

    readable.pause() #

    Added in: v0.9.4

    • Returns:

    The readable.pause() method will cause a stream in flowing mode to stop emitting ‘data’ events, switching out of flowing mode. Any data that becomes available will remain in the internal buffer.

    const readable = getReadableStreamSomehow(); readable.on('data', (chunk) => < console.log(`Received $ 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

    The readable.pause() method has no effect if there is a ‘readable’ event listener.

    readable.pipe(destination[, options]) #

    Added in: v0.9.4

    • destination The destination for writing data
    • options Pipe options
      • end End the writer when the reader ends. Default: true .

      The readable.pipe() method attaches a Writable stream to the readable , causing it to switch automatically into flowing mode and push all of its data to the attached Writable . The flow of data will be automatically managed so that the destination Writable stream is not overwhelmed by a faster Readable stream.

      The following example pipes all of the data from the readable into a file named 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

      It is possible to attach multiple Writable streams to a single Readable stream.

      The readable.pipe() method returns a reference to the destination stream making it possible to set up chains of piped streams:

      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

      By default, stream.end() is called on the destination Writable stream when the source Readable stream emits ‘end’ , so that the destination is no longer writable. To disable this default behavior, the end option can be passed as false , causing the destination stream to remain open:

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

      One important caveat is that if the Readable stream emits an error during processing, the Writable destination is not closed automatically. If an error occurs, it will be necessary to manually close each stream in order to prevent memory leaks.

      The process.stderr and process.stdout Writable streams are never closed until the Node.js process exits, regardless of the specified options.

      readable.read([size]) #

      Added in: v0.9.4

      The readable.read() method reads data out of the internal buffer and returns it. If no data is available to be read, null is returned. By default, the data is returned as a Buffer object unless an encoding has been specified using the readable.setEncoding() method or the stream is operating in object mode.

      The optional size argument specifies a specific number of bytes to read. If size bytes are not available to be read, null will be returned unless the stream has ended, in which case all of the data remaining in the internal buffer will be returned.

      If the size argument is not specified, all of the data contained in the internal buffer will be returned.

      The size argument must be less than or equal to 1 GiB.

      The readable.read() method should only be called on Readable streams operating in paused mode. In flowing mode, readable.read() is called automatically until the internal buffer is fully drained.

      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 $ 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

      Each call to readable.read() returns a chunk of data, or null . The chunks are not concatenated. A while loop is necessary to consume all data currently in the buffer. When reading a large file .read() may return null , having consumed all buffered content so far, but there is still more data to come not yet buffered. In this case a new ‘readable’ event will be emitted when there is more data in the buffer. Finally the ‘end’ event will be emitted when there is no more data to come.

      Therefore to read a file’s whole contents from a readable , it is necessary to collect chunks across multiple ‘readable’ events:

      const chunks = []; readable.on('readable', () => < let chunk; while (null !== (chunk = readable.read())) < chunks.push(chunk); > >); readable.on('end', () => < const content = chunks.join(''); >); copy

      A Readable stream in object mode will always return a single item from a call to readable.read(size) , regardless of the value of the size argument.

      If the readable.read() method returns a chunk of data, a ‘data’ event will also be emitted.

      Calling stream.read([size]) after the ‘end’ event has been emitted will return null . No runtime error will be raised.

      readable.readable #

      Added in: v11.4.0

      Is true if it is safe to call readable.read() , which means the stream has not been destroyed or emitted ‘error’ or ‘end’ .

      readable.readableAborted #

      Added in: v16.8.0
      Stability: 1 — Experimental

      Returns whether the stream was destroyed or errored before emitting ‘end’ .

      readable.readableDidRead #

      Added in: v16.7.0, v14.18.0
      Stability: 1 — Experimental

      Returns whether ‘data’ has been emitted.

      readable.readableEncoding #

      Added in: v12.7.0

      Getter for the property encoding of a given Readable stream. The encoding property can be set using the readable.setEncoding() method.

      readable.readableEnded #

      Added in: v12.9.0

      Becomes true when ‘end’ event is emitted.

      readable.errored #

      Added in: v18.0.0

      Returns error if the stream has been destroyed with an error.

      readable.readableFlowing #

      Added in: v9.4.0

      This property reflects the current state of a Readable stream as described in the Three states section.

      readable.readableHighWaterMark #

      Added in: v9.3.0

      Returns the value of highWaterMark passed when creating this Readable .

      readable.readableLength #

      Added in: v9.4.0

      This property contains the number of bytes (or objects) in the queue ready to be read. The value provides introspection data regarding the status of the highWaterMark .

      readable.readableObjectMode #

      Added in: v12.3.0

      Getter for the property objectMode of a given Readable stream.

      readable.resume() #

      The resume() has no effect if there is a ‘readable’ event listening.

      • Returns:

      The readable.resume() method causes an explicitly paused Readable stream to resume emitting ‘data’ events, switching the stream into flowing mode.

      The readable.resume() method can be used to fully consume the data from a stream without actually processing any of that data:

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

      The readable.resume() method has no effect if there is a ‘readable’ event listener.

      readable.setEncoding(encoding) #

      Added in: v0.9.4

      • encoding The encoding to use.
      • Returns:

      The readable.setEncoding() method sets the character encoding for data read from the Readable stream.

      By default, no encoding is assigned and stream data will be returned as Buffer objects. Setting an encoding causes the stream data to be returned as strings of the specified encoding rather than as Buffer objects. For instance, calling readable.setEncoding(‘utf8’) will cause the output data to be interpreted as UTF-8 data, and passed as strings. Calling readable.setEncoding(‘hex’) will cause the data to be encoded in hexadecimal string format.

      The Readable stream will properly handle multi-byte characters delivered through the stream that would otherwise become improperly decoded if simply pulled from the stream as Buffer objects.

      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]) #

      Added in: v0.9.4

      • destination Optional specific stream to unpipe
      • Returns:

      The readable.unpipe() method detaches a Writable stream previously attached using the stream.pipe() method.

      If the destination is not specified, then all pipes are detached.

      If the destination is specified, but no pipe is set up for it, then the method does nothing.

      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]) #

      The chunk argument can now be a Uint8Array instance.

      Passing chunk as null signals the end of the stream (EOF) and behaves the same as readable.push(null) , after which no more data can be written. The EOF signal is put at the end of the buffer and any buffered data will still be flushed.

      The readable.unshift() method pushes a chunk of data back into the internal buffer. This is useful in certain situations where a stream is being consumed by code that needs to «un-consume» some amount of data that it has optimistically pulled out of the source, so that the data can be passed on to some other party.

      The stream.unshift(chunk) method cannot be called after the ‘end’ event has been emitted or a runtime error will be thrown.

      Developers using stream.unshift() often should consider switching to use of a Transform stream instead. See the API for stream implementers section for more information.

      // 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

      Unlike stream.push(chunk) , stream.unshift(chunk) will not end the reading process by resetting the internal reading state of the stream. This can cause unexpected results if readable.unshift() is called during a read (i.e. from within a stream._read() implementation on a custom stream). Following the call to readable.unshift() with an immediate stream.push(») will reset the reading state appropriately, however it is best to simply avoid calling readable.unshift() while in the process of performing a read.

      readable.wrap(stream) #

      Added in: v0.9.4

      Prior to Node.js 0.10, streams did not implement the entire node:stream module API as it is currently defined. (See Compatibility for more information.)

      When using an older Node.js library that emits ‘data’ events and has a stream.pause() method that is advisory only, the readable.wrap() method can be used to create a Readable stream that uses the old stream as its data source.

      It will rarely be necessary to use readable.wrap() but the method has been provided as a convenience for interacting with older Node.js applications and libraries.

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

      Symbol.asyncIterator support is no longer experimental.

      • Returns: to fully consume the stream.

      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

      If the loop terminates with a break , return , or a throw , the stream will be destroyed. In other terms, iterating over a stream will consume the stream fully. The stream will be read in chunks of size equal to the highWaterMark option. In the code example above, data will be in a single chunk if the file has less then 64 KiB of data because no highWaterMark option is provided to fs.createReadStream() .

      readable[Symbol.asyncDispose]() #

      Added in: v20.4.0, v18.18.0
      Stability: 1 — Experimental

      Calls readable.destroy() with an AbortError and returns a promise that fulfills when the stream is finished.

      readable.compose(stream[, options]) #

      Added in: v19.1.0, v18.13.0
      Stability: 1 — Experimental

      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

      See stream.compose for more information.

      readable.iterator([options]) #

      Added in: v16.3.0
      Stability: 1 — Experimental

      • options
        • destroyOnReturn When set to false , calling return on the async iterator, or exiting a for await. of iteration using a break , return , or throw will not destroy the stream. Default: true .

        The iterator created by this method gives users the option to cancel the destruction of the stream if the for await. of loop is exited by return , break , or throw , or if the iterator should destroy the stream if the stream emitted an error during iteration.

        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]) #

        added highWaterMark in options.

        Added in: v17.4.0, v16.14.0

        Stability: 1 — Experimental

        This method allows mapping over the stream. The fn function will be called for every chunk in the stream. If the fn function returns a promise — that promise will be await ed before being passed to the result stream.

        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]) #

        added highWaterMark in options.

        Added in: v17.4.0, v16.14.0

        Stability: 1 — Experimental

        This method allows filtering the stream. For each chunk in the stream the fn function will be called and if it returns a truthy value, the chunk will be passed to the result stream. If the fn function returns a promise — that promise will be await ed.

        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]) #

        Added in: v17.5.0, v16.15.0
        Stability: 1 — Experimental

        This method allows iterating a stream. For each chunk in the stream the fn function will be called. If the fn function returns a promise — that promise will be await ed.

        This method is different from for await. of loops in that it can optionally process chunks concurrently. In addition, a forEach iteration can only be stopped by having passed a signal option and aborting the related AbortController while for await. of can be stopped with break or return . In either case the stream will be destroyed.

        This method is different from listening to the ‘data’ event in that it uses the readable event in the underlying machinary and can limit the number of concurrent fn calls.

        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]) #

        Added in: v17.5.0, v16.15.0
        Stability: 1 — Experimental

        • options
          • signal allows cancelling the toArray operation if the signal is aborted.

          This method allows easily obtaining the contents of a stream.

          As this method reads the entire stream into memory, it negates the benefits of streams. It’s intended for interoperability and convenience, not as the primary way to consume streams.

          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]) #

          Added in: v17.5.0, v16.15.0
          Stability: 1 — Experimental

          This method is similar to Array.prototype.some and calls fn on each chunk in the stream until the awaited return value is true (or any truthy value). Once an fn call on a chunk awaited return value is truthy, the stream is destroyed and the promise is fulfilled with true . If none of the fn calls on the chunks return a truthy value, the promise is fulfilled with 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]) #

          Added in: v17.5.0, v16.17.0
          Stability: 1 — Experimental

          This method is similar to Array.prototype.find and calls fn on each chunk in the stream to find a chunk with a truthy value for fn . Once an fn call’s awaited return value is truthy, the stream is destroyed and the promise is fulfilled with value for which fn returned a truthy value. If all of the fn calls on the chunks return a falsy value, the promise is fulfilled with 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]) #

          Added in: v17.5.0, v16.15.0
          Stability: 1 — Experimental

          This method is similar to Array.prototype.every and calls fn on each chunk in the stream to check if all awaited return values are truthy value for fn . Once an fn call on a chunk awaited return value is falsy, the stream is destroyed and the promise is fulfilled with false . If all of the fn calls on the chunks return a truthy value, the promise is fulfilled with 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]) #

          Added in: v17.5.0, v16.15.0
          Stability: 1 — Experimental

          This method returns a new stream by applying the given callback to each chunk of the stream and then flattening the result.

          It is possible to return a stream or another iterable or async iterable from fn and the result streams will be merged (flattened) into the returned stream.

          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]) #

          Added in: v17.5.0, v16.15.0
          Stability: 1 — Experimental

          This method returns a new stream with the first limit chunks dropped.

          import < Readable > from 'node:stream'; await Readable.from([1, 2, 3, 4]).drop(2).toArray(); // [3, 4] copy

          readable.take(limit[, options]) #

          Added in: v17.5.0, v16.15.0
          Stability: 1 — Experimental

          This method returns a new stream with the first limit chunks.

          import < Readable > from 'node:stream'; await Readable.from([1, 2, 3, 4]).take(2).toArray(); // [1, 2] copy

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

          Added in: v17.5.0, v16.15.0
          Stability: 1 — Experimental

          This method calls fn on each chunk of the stream in order, passing it the result from the calculation on the previous element. It returns a promise for the final value of the reduction.

          If no initial value is supplied the first chunk of the stream is used as the initial value. If the stream is empty, the promise is rejected with a TypeError with the ERR_INVALID_ARGS code property.

          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

          The reducer function iterates the stream element-by-element which means that there is no concurrency parameter or parallelism. To perform a reduce concurrently, you can extract the async function to readable.map method.

          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, ) => totalSize + size, 0); console.log(folderSize); copy

          Duplex and transform streams #

          Class: stream.Duplex #

          Instances of Duplex now return true when checking instanceof stream.Writable .

          Duplex streams are streams that implement both the Readable and Writable interfaces.

          Examples of Duplex streams include:

          duplex.allowHalfOpen #

          Added in: v0.9.4

          If false then the stream will automatically end the writable side when the readable side ends. Set initially by the allowHalfOpen constructor option, which defaults to true .

          This can be changed manually to change the half-open behavior of an existing Duplex stream instance, but must be changed before the ‘end’ event is emitted.

          Class: stream.Transform #

          Added in: v0.9.4

          Transform streams are Duplex streams where the output is in some way related to the input. Like all Duplex streams, Transform streams implement both the Readable and Writable interfaces.

          Examples of Transform streams include:

          transform.destroy([error]) #

          Work as a no-op on a stream that has already been destroyed.

          • error
          • Returns:

          Destroy the stream, and optionally emit an ‘error’ event. After this call, the transform stream would release any internal resources. Implementors should not override this method, but instead implement readable._destroy() . The default implementation of _destroy() for Transform also emit ‘close’ unless emitClose is set in false.

          Once destroy() has been called, any further calls will be a no-op and no further errors except from _destroy() may be emitted as ‘error’ .

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

          Added support for ReadableStream and WritableStream .

          The signal option was added.

          The finished(stream, cb) will wait for the ‘close’ event before invoking the callback. The implementation tries to detect legacy streams and only apply this behavior to streams which are expected to emit ‘close’ .

          Emitting ‘close’ before ‘end’ on a Readable stream will cause an ERR_STREAM_PREMATURE_CLOSE error.

          Callback will be invoked on streams which have already finished before the call to finished(stream, cb) .

          A readable and/or writable stream/webstream.

          • options
            • error If set to false , then a call to emit(‘error’, err) is not treated as finished. Default: true .
            • readable When set to false , the callback will be called when the stream ends even though the stream might still be readable. Default: true .
            • writable When set to false , the callback will be called when the stream ends even though the stream might still be writable. Default: true .
            • signal allows aborting the wait for the stream finish. The underlying stream will not be aborted if the signal is aborted. The callback will get called with an AbortError . All registered listeners added by this function will also be removed.
            • cleanup remove all registered stream listeners. Default: false .

            A function to get notified when a stream is no longer readable, writable or has experienced an error or a premature close event.

            const < finished >= require('node:stream'); const fs = require('node:fs'); const rs = fs.createReadStream('archive.tar'); finished(rs, (err) => < if (err) < console.error('Stream failed.', err); > else < console.log('Stream is done reading.'); > >); rs.resume(); // Drain the stream. copy

            Especially useful in error handling scenarios where a stream is destroyed prematurely (like an aborted HTTP request), and will not emit ‘end’ or ‘finish’ .

            The finished API provides promise version.

            stream.finished() leaves dangling event listeners (in particular ‘error’ , ‘end’ , ‘finish’ and ‘close’ ) after callback has been invoked. The reason for this is so that unexpected ‘error’ events (due to incorrect stream implementations) do not cause unexpected crashes. If this is unwanted behavior then the returned cleanup function needs to be invoked in the callback:

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

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

            stream.pipeline(streams, callback) #

            Added support for webstreams.

            Passing an invalid callback to the callback argument now throws ERR_INVALID_ARG_TYPE instead of ERR_INVALID_CALLBACK .

            The pipeline(. cb) will wait for the ‘close’ event before invoking the callback. The implementation tries to detect legacy streams and only apply this behavior to streams which are expected to emit ‘close’ .

            Add support for async generators.

            • streams | | | | | |
            • source | | | |
              • Returns: |
              • source
              • Returns:
              • source
              • Returns: |
              • err
              • val Resolved value of Promise returned by destination .

              A module method to pipe between streams and generators forwarding errors and properly cleaning up and provide a callback when the pipeline is complete.

              const < pipeline >= require('node:stream'); const fs = require('node:fs'); const zlib = require('node:zlib'); // Use the pipeline API to easily pipe a series of streams // together and get notified when the pipeline is fully done. // A pipeline to gzip a potentially huge tar file efficiently: pipeline( fs.createReadStream('archive.tar'), zlib.createGzip(), fs.createWriteStream('archive.tar.gz'), (err) => < if (err) < console.error('Pipeline failed.', err); > else < console.log('Pipeline succeeded.'); > >, ); copy

              The pipeline API provides a promise version.

              stream.pipeline() will call stream.destroy(err) on all streams except:

              • Readable streams which have emitted ‘end’ or ‘close’ .
              • Writable streams which have emitted ‘finish’ or ‘close’ .

              stream.pipeline() leaves dangling event listeners on the streams after the callback has been invoked. In the case of reuse of streams after failure, this can cause event listener leaks and swallowed errors. If the last stream is readable, dangling event listeners will be removed so that the last stream can be consumed later.

              stream.pipeline() closes all the streams when an error is raised. The IncomingRequest usage with pipeline could lead to an unexpected behavior once it would destroy the socket without sending the expected response. See the example below:

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

              stream.compose(. streams) #

              Added support for stream class.

              Added support for webstreams.

              Stability: 1 — stream.compose is experimental.

              Combines two or more streams into a Duplex stream that writes to the first stream and reads from the last. Each provided stream is piped into the next, using stream.pipeline . If any of the streams error then all are destroyed, including the outer Duplex stream.

              Because stream.compose returns a new stream that in turn can (and should) be piped into other streams, it enables composition. In contrast, when passing streams to stream.pipeline , typically the first stream is a readable stream and the last a writable stream, forming a closed circuit.

              If passed a Function it must be a factory method taking a source Iterable .

              import < compose, Transform > from 'node:stream'; const removeSpaces = new Transform(< transform(chunk, encoding, callback) < callback(null, String(chunk).replace(' ', '')); >, >); async function* toUpper(source) < for await (const chunk of source) < yield String(chunk).toUpperCase(); > > let res = ''; for await (const buf of compose(removeSpaces, toUpper).end('hello world')) < res += buf; >console.log(res); // prints 'HELLOWORLD' copy

              stream.compose can be used to convert async iterables, generators and functions into streams.

              • AsyncIterable converts into a readable Duplex . Cannot yield null .
              • AsyncGeneratorFunction converts into a readable/writable transform Duplex . Must take a source AsyncIterable as first parameter. Cannot yield null .
              • AsyncFunction converts into a writable Duplex . Must return either null or undefined .

              import < compose >from 'node:stream'; import < finished >from 'node:stream/promises'; // Convert AsyncIterable into readable Duplex. const s1 = compose(async function*() < yield 'Hello'; yield 'World'; >()); // Convert AsyncGenerator into transform Duplex. const s2 = compose(async function*(source) < for await (const chunk of source) < yield String(chunk).toUpperCase(); > >); let res = ''; // Convert AsyncFunction into writable Duplex. const s3 = compose(async function(source) < for await (const chunk of source) < res += chunk; >>); await finished(compose(s1, s2, s3)); console.log(res); // prints 'HELLOWORLD' copy

              See readable.compose(stream) for stream.compose as operator.

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

              Added in: v12.3.0, v10.17.0

              • iterable Object implementing the Symbol.asyncIterator or Symbol.iterator iterable protocol. Emits an ‘error’ event if a null value is passed.
              • options Options provided to new stream.Readable([options]) . By default, Readable.from() will set options.objectMode to true , unless this is explicitly opted out by setting options.objectMode to false .
              • Returns:

              A utility method for creating readable streams out of iterators.

              const < Readable > = require('node:stream'); async function * generate( ) < yield 'hello'; yield 'streams'; > const readable = Readable.from(generate()); readable.on('data', (chunk) => < console.log(chunk); >); copy

              Calling Readable.from(string) or Readable.from(buffer) will not have the strings or buffers be iterated to match the other streams semantics for performance reasons.

              If an Iterable object containing promises is passed as an argument, it might result in unhandled rejection.

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

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

              Added in: v17.0.0
              Stability: 1 — Experimental

              • readableStream
              • options
                • encoding
                • highWaterMark
                • objectMode
                • signal

                Saved searches

                Use saved searches to filter your results more quickly

                Cancel Create saved search

                You signed in with another tab or window. Reload to refresh your session. You signed out in another tab or window. Reload to refresh your session. You switched accounts on another tab or window. Reload to refresh your session.

                nodejs / undici Public

                An HTTP/1.1 client, written from scratch for Node.js

                License

                nodejs/undici

                This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

                Switch branches/tags
                Branches Tags
                Could not load branches
                Nothing to show
                Could not load tags
                Nothing to show

                Name already in use

                A tag already exists with the provided branch name. Many Git commands accept both tag and branch names, so creating this branch may cause unexpected behavior. Are you sure you want to create this branch?

                Cancel Create

                • Local
                • Codespaces

                HTTPS GitHub CLI
                Use Git or checkout with SVN using the web URL.
                Work fast with our official CLI. Learn more about the CLI.

                Sign In Required

                Please sign in to use Codespaces.

                Launching GitHub Desktop

                If nothing happens, download GitHub Desktop and try again.

                Launching GitHub Desktop

                If nothing happens, download GitHub Desktop and try again.

                Launching Xcode

                If nothing happens, download Xcode and try again.

                Launching Visual Studio Code

                Your codespace will open once ready.

                There was a problem preparing your codespace, please try again.

                Latest commit

                3a77cbb Oct 27, 2023

                Git stats

                Files

                Failed to load latest commit information.

                Latest commit message
                Commit time
                October 11, 2023 10:25
                September 30, 2023 07:53
                September 8, 2023 14:42
                June 23, 2023 12:42
                October 25, 2022 10:09
                October 17, 2023 16:59
                April 21, 2023 10:14
                March 17, 2022 10:02
                October 27, 2023 10:36
                October 11, 2023 23:38
                October 26, 2023 22:06
                October 23, 2023 09:05
                April 2, 2021 14:42
                August 25, 2021 18:42
                November 13, 2022 17:46
                March 26, 2021 11:20
                February 10, 2022 17:54
                April 23, 2023 16:50
                March 26, 2021 13:15
                October 25, 2022 09:58
                June 23, 2023 12:42
                August 31, 2021 15:38
                August 24, 2020 11:10
                April 2, 2021 10:44
                September 21, 2023 13:37
                December 8, 2021 07:27
                October 23, 2023 07:42
                September 20, 2023 15:02
                November 11, 2021 06:36
                September 3, 2023 04:44
                October 26, 2023 13:47

                README.md

                undici

                An HTTP/1.1 client, written from scratch for Node.js.

                Undici means eleven in Italian. 1.1 -> 11 -> Eleven -> Undici. It is also a Stranger Things reference.

                Have a question about using Undici? Open a Q&A Discussion or join our official OpenJS Slack channel.

                Install

                npm i undici 

                Benchmarks

                The benchmark is a simple hello world example using a number of unix sockets (connections) with a pipelining depth of 10 running on Node 20.6.0.

                Connections 1

                Tests	Samples	Result	Tolerance	Difference with slowest
                http — no keepalive	15	5.32 req/sec	± 2.61 %	—
                http — keepalive	10	5.35 req/sec	± 2.47 %	+ 0.44 %
                undici — fetch	15	41.85 req/sec	± 2.49 %	+ 686.04 %
                undici — pipeline	40	50.36 req/sec	± 2.77 %	+ 845.92 %
                undici — stream	15	60.58 req/sec	± 2.75 %	+ 1037.72 %
                undici — request	10	61.19 req/sec	± 2.60 %	+ 1049.24 %
                undici — dispatch	20	64.84 req/sec	± 2.81 %	+ 1117.81 %

                Connections 50

                Tests	Samples	Result	Tolerance	Difference with slowest
                undici — fetch	30	2107.19 req/sec	± 2.69 %	—
                http — no keepalive	10	2698.90 req/sec	± 2.68 %	+ 28.08 %
                http — keepalive	10	4639.49 req/sec	± 2.55 %	+ 120.17 %
                undici — pipeline	40	6123.33 req/sec	± 2.97 %	+ 190.59 %
                undici — stream	50	9426.51 req/sec	± 2.92 %	+ 347.35 %
                undici — request	10	10162.88 req/sec	± 2.13 %	+ 382.29 %
                undici — dispatch	50	11191.11 req/sec	± 2.98 %	+ 431.09 %

                Quick Start

                import  request > from 'undici' const  statusCode, headers, trailers, body > = await request('http://localhost:3000/foo') console.log('response received', statusCode) console.log('headers', headers) for await (const data of body)  console.log('data', data) > console.log('trailers', trailers)

                Body Mixins

                The body mixins are the most common way to format the request/response body. Mixins include:

                import  request > from 'undici' const  statusCode, headers, trailers, body > = await request('http://localhost:3000/foo') console.log('response received', statusCode) console.log('headers', headers) console.log('data', await body.json()) console.log('trailers', trailers)

                Note: Once a mixin has been called then the body cannot be reused, thus calling additional mixins on .body , e.g. .body.json(); .body.text() will result in an error TypeError: unusable being thrown and returned through the Promise rejection.

                Should you need to access the body in plain-text after using a mixin, the best practice is to use the .text() mixin first and then manually parse the text to the desired format.

                For more information about their behavior, please reference the body mixin from the Fetch Standard.

                Common API Methods

                This section documents our most commonly used API methods. Additional APIs are documented in their own files within the docs folder and are accessible via the navigation list on the left side of the docs site.

                undici.request([url, options]): Promise

                • url string | URL | UrlObject
                • optionsRequestOptions
                  • dispatcher Dispatcher — Default: getGlobalDispatcher
                  • method String — Default: PUT if options.body , otherwise GET
                  • maxRedirections Integer — Default: 0

                  Returns a promise with the result of the Dispatcher.request method.

                  undici.stream([url, options, ]factory): Promise

                  • url string | URL | UrlObject
                  • optionsStreamOptions
                    • dispatcher Dispatcher — Default: getGlobalDispatcher
                    • method String — Default: PUT if options.body , otherwise GET
                    • maxRedirections Integer — Default: 0

                    Returns a promise with the result of the Dispatcher.stream method.

                    Calls options.dispatcher.stream(options, factory) .

                    See Dispatcher.stream for more details.

                    undici.pipeline([url, options, ]handler): Duplex

                    • url string | URL | UrlObject
                    • optionsPipelineOptions
                      • dispatcher Dispatcher — Default: getGlobalDispatcher
                      • method String — Default: PUT if options.body , otherwise GET
                      • maxRedirections Integer — Default: 0

                      Calls options.dispatch.pipeline(options, handler) .

                      undici.connect([url, options]): Promise

                      Starts two-way communications with the requested resource using HTTP CONNECT.

                      • url string | URL | UrlObject
                      • optionsConnectOptions
                        • dispatcher Dispatcher — Default: getGlobalDispatcher
                        • maxRedirections Integer — Default: 0

                        Returns a promise with the result of the Dispatcher.connect method.

                        undici.fetch(input[, init]): Promise

                        • https://developer.mozilla.org/en-US/docs/Web/API/WindowOrWorkerGlobalScope/fetch
                        • https://fetch.spec.whatwg.org/#fetch-method

                        Only supported on Node 16.8+.

                        Basic usage example:

                        import  fetch > from 'undici' const res = await fetch('https://example.com') const json = await res.json() console.log(json)

                        You can pass an optional dispatcher to fetch as:

                        import  fetch, Agent > from 'undici' const res = await fetch('https://example.com',  // Mocks are also supported dispatcher: new Agent( keepAliveTimeout: 10, keepAliveMaxTimeout: 10 >) >) const json = await res.json() console.log(json)

                        request.body

                        A body can be of the following types:

                        • ArrayBuffer
                        • ArrayBufferView
                        • AsyncIterables
                        • Blob
                        • Iterables
                        • String
                        • URLSearchParams
                        • FormData

                        In this implementation of fetch, request.body now accepts Async Iterables . It is not present in the Fetch Standard.

                        import  fetch > from 'undici' const data =  async *[Symbol.asyncIterator]()  yield 'hello' yield 'world' >, > await fetch('https://example.com',  body: data, method: 'POST', duplex: 'half' >)

                        request.duplex

                        In this implementation of fetch, request.duplex must be set if request.body is ReadableStream or Async Iterables . And fetch requests are currently always be full duplex. More detail refer to Fetch Standard.

                        response.body

                        Nodejs has two kinds of streams: web streams, which follow the API of the WHATWG web standard found in browsers, and an older Node-specific streams API. response.body returns a readable web stream. If you would prefer to work with a Node stream you can convert a web stream using .fromWeb() .

                        import  fetch > from 'undici' import  Readable > from 'node:stream' const response = await fetch('https://example.com') const readableWebStream = response.body const readableNodeStream = Readable.fromWeb(readableWebStream)

                        Specification Compliance

                        This section documents parts of the Fetch Standard that Undici does not support or does not fully implement.

                        Garbage Collection

                        The Fetch Standard allows users to skip consuming the response body by relying on garbage collection to release connection resources. Undici does not do the same. Therefore, it is important to always either consume or cancel the response body.

                        Garbage collection in Node is less aggressive and deterministic (due to the lack of clear idle periods that browsers have through the rendering refresh rate) which means that leaving the release of connection resources to the garbage collector can lead to excessive connection usage, reduced performance (due to less connection re-use), and even stalls or deadlocks when running out of connections.

                        // Do const headers = await fetch(url) .then(async res =>  for await (const chunk of res.body)  // force consumption of body > return res.headers >) // Do not const headers = await fetch(url) .then(res => res.headers)

                        However, if you want to get only headers, it might be better to use HEAD request method. Usage of this method will obviate the need for consumption or cancelling of the response body. See MDN — HTTP — HTTP request methods — HEAD for more details.

                        const headers = await fetch(url,  method: 'HEAD' >) .then(res => res.headers)

                        Forbidden and Safelisted Header Names

                        • https://fetch.spec.whatwg.org/#cors-safelisted-response-header-name
                        • https://fetch.spec.whatwg.org/#forbidden-header-name
                        • https://fetch.spec.whatwg.org/#forbidden-response-header-name
                        • wintercg/fetch#6

                        The Fetch Standard requires implementations to exclude certain headers from requests and responses. In browser environments, some headers are forbidden so the user agent remains in full control over them. In Undici, these constraints are removed to give more control to the user.

                        undici.upgrade([url, options]): Promise

                        Upgrade to a different protocol. See MDN — HTTP — Protocol upgrade mechanism for more details.

                        • url string | URL | UrlObject
                        • optionsUpgradeOptions
                          • dispatcher Dispatcher — Default: getGlobalDispatcher
                          • maxRedirections Integer — Default: 0

                          Returns a promise with the result of the Dispatcher.upgrade method.

                          undici.setGlobalDispatcher(dispatcher)

                          • dispatcher Dispatcher

                          Sets the global dispatcher used by Common API Methods.

                          undici.getGlobalDispatcher()

                          Gets the global dispatcher used by Common API Methods.

                          undici.setGlobalOrigin(origin)

                          • origin string | URL | undefined

                          Sets the global origin used in fetch .

                          If undefined is passed, the global origin will be reset. This will cause Response.redirect , new Request() , and fetch to throw an error when a relative path is passed.

                          setGlobalOrigin('http://localhost:3000') const response = await fetch('/api/ping') console.log(response.url) // http://localhost:3000/api/ping

                          undici.getGlobalOrigin()

                          Gets the global origin used in fetch .

                          UrlObject

                          • port string | number (optional)
                          • path string (optional)
                          • pathname string (optional)
                          • hostname string (optional)
                          • origin string (optional)
                          • protocol string (optional)
                          • search string (optional)

                          Specification Compliance

                          This section documents parts of the HTTP/1.1 specification that Undici does not support or does not fully implement.

                          Expect

                          Undici does not support the Expect request header field. The request body is always immediately sent and the 100 Continue response will be ignored.

                          Pipelining

                          Undici will only use pipelining if configured with a pipelining factor greater than 1 .

                          Undici always assumes that connections are persistent and will immediately pipeline requests, without checking whether the connection is persistent. Hence, automatic fallback to HTTP/1.0 or HTTP/1.1 without pipelining is not supported.

                          Undici will immediately pipeline when retrying requests after a failed connection. However, Undici will not retry the first remaining requests in the prior pipeline and instead error the corresponding callback/promise/stream.

                          Undici will abort all running requests in the pipeline when any of them are aborted.

                          • Refs: https://tools.ietf.org/html/rfc2616#section-8.1.2.2
                          • Refs: https://tools.ietf.org/html/rfc7230#section-6.3.2

                          Manual Redirect

                          Since it is not possible to manually follow an HTTP redirect on the server-side, Undici returns the actual response instead of an opaqueredirect filtered one when invoked with a manual redirect. This aligns fetch() with the other implementations in Deno and Cloudflare Workers.

                          Workarounds

                          Network address family autoselection.

                          If you experience problem when connecting to a remote server that is resolved by your DNS servers to a IPv6 (AAAA record) first, there are chances that your local router or ISP might have problem connecting to IPv6 networks. In that case undici will throw an error with code UND_ERR_CONNECT_TIMEOUT .

                          If the target server resolves to both a IPv6 and IPv4 (A records) address and you are using a compatible Node version (18.3.0 and above), you can fix the problem by providing the autoSelectFamily option (support by both undici.request and undici.Agent ) which will enable the family autoselection algorithm when establishing the connection.

                          Collaborators

                          • Daniele Belardi, https://www.npmjs.com/~dnlup
                          • Ethan Arrowood, https://www.npmjs.com/~ethan_arrowood
                          • Matteo Collina, https://www.npmjs.com/~matteo.collina
                          • Matthew Aitken, https://www.npmjs.com/~khaf
                          • Robert Nagy, https://www.npmjs.com/~ronag
                          • Szymon Marczak, https://www.npmjs.com/~szmarczak
                          • Tomas Della Vedova, https://www.npmjs.com/~delvedor

                          Releasers

                          • Ethan Arrowood, https://www.npmjs.com/~ethan_arrowood
                          • Matteo Collina, https://www.npmjs.com/~matteo.collina
                          • Robert Nagy, https://www.npmjs.com/~ronag
                          • Matthew Aitken, https://www.npmjs.com/~khaf
                          Похожие публикации:

                          1. Transmission cli как пользоваться
                          2. Как изменить кодировку csv файла на utf 8 python 3
                          3. Как составить зашифрованное письмо любимому
                          4. Что будет если поставить блок питания меньшей мощности